filer: serialize same-path mutations with a per-path lock (#9639)

CreateEntry is a FindEntry-then-write with no lock, so concurrent creates to the
same path race: OExcl can admit two creators, and a conditional check-then-act
has no atomicity. Add a per-path exclusive lock (util.LockTable, which evicts
idle keys so it stays bounded) on the FilerServer and take it in CreateEntry, so
the existence check and the write are atomic on this filer.

This is the local serialization point that lets callers route a key's writes to
its owner filer and drop the distributed lock for that key. AppendToEntry keeps
its distributed lock for now; it can move to the per-path lock once its callers
route to the owner.

weed/server/filer_grpc_server.go

@@ -10,7 +10,6 @@ import (
 	"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"
@@ -186,6 +185,13 @@ 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 := newEntry.FullPath
+	pathLock := fs.entryLockTable.AcquireLock("CreateEntry", fullpath, util.ExclusiveLock)
+	defer fs.entryLockTable.ReleaseLock(fullpath, pathLock)
+
 	ctx, eventSink := filer.WithMetadataEventSink(ctx)
 	createErr := fs.filer.CreateEntry(ctx, newEntry, req.OExcl, req.IsFromOtherCluster, req.Signatures, req.SkipCheckParentDirectory, so.MaxFileNameLength)
 

weed/server/filer_grpc_server_create_lock_test.go

@@ -0,0 +1,68 @@
+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, alreadyExists, unexpected 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},
+				},
+			})
+			switch {
+			case err != nil:
+				atomic.AddInt32(&unexpected, 1)
+			case resp.Error == "":
+				atomic.AddInt32(&success, 1)
+			case resp.ErrorCode == filer_pb.FilerError_ENTRY_ALREADY_EXISTS:
+				atomic.AddInt32(&alreadyExists, 1)
+			default:
+				atomic.AddInt32(&unexpected, 1)
+			}
+		}()
+	}
+	close(start)
+	wg.Wait()
+
+	// Exactly one winner; every loser fails with ENTRY_ALREADY_EXISTS and nothing
+	// else, so an unrelated failure can't masquerade as a passing test.
+	if success != 1 || alreadyExists != racers-1 || unexpected != 0 {
+		t.Fatalf("winners=%d already_exists=%d unexpected=%d (racers=%d)", success, alreadyExists, unexpected, racers)
+	}
+}

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

@@ -124,6 +124,14 @@ 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.
+	// CreateEntry takes it today; UpdateEntry and DeleteEntry are intended to take
+	// it too as their callers route a key's writes to this node, making it the
+	// local serialization point for read-modify-write operations that replaces
+	// the distributed lock for that key. 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 +170,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()