From 8398af35721c82eae00b4039b2005a871a28a7e8 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 2 Sep 2026 19:56:11 -0700 Subject: [PATCH] filer: route exclusive and conditional creates to the entry's ring owner (#11109) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 | 10 +- weed/pb/filer.proto | 6 +- weed/pb/filer_pb/filer.pb.go | 22 ++- weed/pb/filer_pb/filer_vtproto.pb.go | 33 ++++ weed/server/filer_grpc_server.go | 30 +++ .../filer_grpc_server_create_route_test.go | 175 ++++++++++++++++++ weed/server/filer_grpc_server_route.go | 9 + 7 files changed, 274 insertions(+), 11 deletions(-) create mode 100644 weed/server/filer_grpc_server_create_route_test.go diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index ca1c0114c..324cbf5bb 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/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 locations_map = 1; diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index dbd18dfaf..fd440bbae 100644 --- a/weed/pb/filer.proto +++ b/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 diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 14e532d49..46bacb313 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/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. - Condition *WriteCondition `protobuf:"bytes,7,opt,name=condition,proto3" json:"condition,omitempty"` + // 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" + diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index 6d944c6b2..0b7b4d153 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/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:]) diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index b6ed2d411..b86d82475 100644 --- a/weed/server/filer_grpc_server.go +++ b/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) diff --git a/weed/server/filer_grpc_server_create_route_test.go b/weed/server/filer_grpc_server_create_route_test.go new file mode 100644 index 000000000..f5f4dc99b --- /dev/null +++ b/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") + } +} diff --git a/weed/server/filer_grpc_server_route.go b/weed/server/filer_grpc_server_route.go index b52d3844b..d60254bfb 100644 --- a/weed/server/filer_grpc_server_route.go +++ b/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) +}