filer: session lease + reaping for POSIX locks (#9666)

* filer: session lease + reaping for POSIX locks

A mount renews its session lease by keepalive (new KEEP_ALIVE op); the
owner filer records last-seen per session and a background sweeper reaps
the locks of leased sessions that stop renewing — a dead or partitioned
mount. Only sessions that have renewed are leased, so this is inert until
mounts run with -posixLock.

* mount: route POSIX advisory locks to the owner filer (-posixLock) (#9665)

mount: route POSIX advisory locks to the owner filer under -dlm

With -dlm, GetLk/SetLk/SetLkw and the flush/release cleanup paths go to
the inode's owner filer via the PosixLock RPC instead of the local table,
so flock/fcntl are honored across mounts. Advisory locking rides the same
switch as whole-file write coordination — and is therefore off under
writeback cache, which implies single-writer. The mount calls its filer
and relies on filer-side forwarding to reach the owner. Keys are the inode
identity (HardLinkId else path); SetLkw is client-side polling with the
FUSE cancel channel (no server wait queue); a per-mount session id
namespaces owners; a local hint avoids a release RPC on every close.

* mount,filer: bound posix-lock release RPCs and stop the reaper on shutdown

The unlock/release RPCs run off the syscall path (close/flush) and used
context.Background() with no deadline, so a slow or unreachable filer could
hang close() indefinitely; bound them to 5s (they still aren't cancelled by
an interrupt). The lease-reaping sweeper now selects on a stop channel that
FilerServer.Shutdown closes, instead of looping for the process lifetime.

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

@@ -396,6 +396,7 @@ enum PosixLockOp {
     GET_LK = 2;             // report a conflicting lock, if any
     RELEASE_POSIX_OWNER = 3; // drop the owner's fcntl locks (flush-time)
     RELEASE_FLOCK_OWNER = 4; // drop the owner's flock locks (release-time)
+    KEEP_ALIVE = 5;         // renew the session's lease on this owner (lock.sid)
 }
 
 message PosixLockResponse {

weed/filer/posixlock/manager.go

@@ -1,6 +1,9 @@
 package posixlock
 
-import "sync"
+import (
+	"sync"
+	"time"
+)
 
 // Manager is the owner filer's in-memory authority for POSIX advisory locks
 // across inodes. Lock state lives here, not in replicated metadata: it is
@@ -13,18 +16,58 @@ import "sync"
 // "hl:"+hex(HardLinkId) for a hardlinked inode — so all names of one inode share
 // a Set. The Manager is safe for concurrent use.
 type Manager struct {
-	mu    sync.Mutex
-	byKey map[string]*Set            // inode key -> held locks
-	bySid map[uint64]map[string]bool // session -> keys it currently holds locks on
+	mu       sync.Mutex
+	byKey    map[string]*Set            // inode key -> held locks
+	bySid    map[uint64]map[string]bool // session -> keys it currently holds locks on
+	lastSeen map[uint64]time.Time       // session -> last keepalive; only renewing sessions are leased
 }
 
 func NewManager() *Manager {
 	return &Manager{
-		byKey: make(map[string]*Set),
-		bySid: make(map[uint64]map[string]bool),
+		byKey:    make(map[string]*Set),
+		bySid:    make(map[uint64]map[string]bool),
+		lastSeen: make(map[uint64]time.Time),
 	}
 }
 
+// Renew records a keepalive from a session, placing it under lease management.
+// Only sessions that have renewed are subject to ReapExpired, so a session that
+// never sends keepalives (e.g. before the mount keepalive exists) is never reaped.
+func (m *Manager) Renew(sid uint64) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.lastSeen[sid] = time.Now()
+}
+
+// ReapExpired releases the locks of every leased session whose last keepalive is
+// older than ttl — a dead or partitioned mount. Sessions that never renewed are
+// left untouched. Returns the reaped session ids.
+func (m *Manager) ReapExpired(ttl time.Duration) []uint64 {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	cutoff := time.Now().Add(-ttl)
+	var reaped []uint64
+	for sid, seen := range m.lastSeen {
+		if seen.After(cutoff) {
+			continue
+		}
+		for key := range m.bySid[sid] {
+			s := m.byKey[key]
+			if s == nil {
+				continue
+			}
+			s.ReleaseSession(sid)
+			if s.Empty() {
+				delete(m.byKey, key)
+			}
+		}
+		delete(m.bySid, sid)
+		delete(m.lastSeen, sid)
+		reaped = append(reaped, sid)
+	}
+	return reaped
+}
+
 // TryLock grants lk on key, or returns the conflicting lock and false. The set
 // is created on first use and dropped again when it empties.
 func (m *Manager) TryLock(key string, lk Range) (Range, bool) {

weed/filer/posixlock/manager_test.go

@@ -6,6 +6,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"testing"
+	"time"
 )
 
 func TestManagerGrantAndConflict(t *testing.T) {
@@ -109,6 +110,37 @@ func TestManagerReleaseSessionReapsAcrossKeys(t *testing.T) {
 	}
 }
 
+func TestManagerReapsOnlyStaleLeasedSessions(t *testing.T) {
+	m := NewManager()
+	// Session 1: holds a lock, leased but stale (renewed long ago).
+	m.TryLock("a", Range{Start: 0, End: 99, Type: Write, Sid: 1, Owner: 1})
+	m.Renew(1)
+	m.lastSeen[1] = time.Now().Add(-time.Hour)
+	// Session 2: holds a lock, leased and fresh.
+	m.TryLock("b", Range{Start: 0, End: 99, Type: Write, Sid: 2, Owner: 1})
+	m.Renew(2)
+	// Session 3: holds a lock but never renewed (no lease) — must not be reaped.
+	m.TryLock("c", Range{Start: 0, End: 99, Type: Write, Sid: 3, Owner: 1})
+
+	reaped := m.ReapExpired(30 * time.Second)
+
+	if len(reaped) != 1 || reaped[0] != 1 {
+		t.Fatalf("only the stale leased session should be reaped, got %v", reaped)
+	}
+	if _, ok := m.byKey["a"]; ok {
+		t.Fatal("stale session's lock should be gone")
+	}
+	if _, ok := m.byKey["b"]; !ok {
+		t.Fatal("fresh session's lock must remain")
+	}
+	if _, ok := m.byKey["c"]; !ok {
+		t.Fatal("never-renewed session must not be reaped")
+	}
+	if _, ok := m.lastSeen[1]; ok {
+		t.Fatal("reaped session's lease entry should be cleared")
+	}
+}
+
 // Mutual exclusion under concurrent whole-file flock churn through the Manager:
 // at most one owner may believe it holds the exclusive lock at any instant.
 func TestManagerConcurrentFlockMutualExclusion(t *testing.T) {

weed/pb/filer.proto

@@ -396,6 +396,7 @@ enum PosixLockOp {
     GET_LK = 2;             // report a conflicting lock, if any
     RELEASE_POSIX_OWNER = 3; // drop the owner's fcntl locks (flush-time)
     RELEASE_FLOCK_OWNER = 4; // drop the owner's flock locks (release-time)
+    KEEP_ALIVE = 5;         // renew the session's lease on this owner (lock.sid)
 }
 
 message PosixLockResponse {

weed/pb/filer_pb/filer.pb.go

@@ -144,6 +144,7 @@ const (
 	PosixLockOp_GET_LK              PosixLockOp = 2 // report a conflicting lock, if any
 	PosixLockOp_RELEASE_POSIX_OWNER PosixLockOp = 3 // drop the owner's fcntl locks (flush-time)
 	PosixLockOp_RELEASE_FLOCK_OWNER PosixLockOp = 4 // drop the owner's flock locks (release-time)
+	PosixLockOp_KEEP_ALIVE          PosixLockOp = 5 // renew the session's lease on this owner (lock.sid)
 )
 
 // Enum value maps for PosixLockOp.
@@ -154,6 +155,7 @@ var (
 		2: "GET_LK",
 		3: "RELEASE_POSIX_OWNER",
 		4: "RELEASE_FLOCK_OWNER",
+		5: "KEEP_ALIVE",
 	}
 	PosixLockOp_value = map[string]int32{
 		"TRY_LOCK":            0,
@@ -161,6 +163,7 @@ var (
 		"GET_LK":              2,
 		"RELEASE_POSIX_OWNER": 3,
 		"RELEASE_FLOCK_OWNER": 4,
+		"KEEP_ALIVE":          5,
 	}
 )
 
@@ -7153,7 +7156,7 @@ const file_filer_proto_rawDesc = "" +
 	"\x15EXISTING_IS_DIRECTORY\x10\x03\x12\x14\n" +
 	"\x10EXISTING_IS_FILE\x10\x04\x12\x18\n" +
 	"\x14ENTRY_ALREADY_EXISTS\x10\x05\x12\x17\n" +
-	"\x13PRECONDITION_FAILED\x10\x06*e\n" +
+	"\x13PRECONDITION_FAILED\x10\x06*u\n" +
 	"\vPosixLockOp\x12\f\n" +
 	"\bTRY_LOCK\x10\x00\x12\n" +
 	"\n" +
@@ -7161,7 +7164,9 @@ const file_filer_proto_rawDesc = "" +
 	"\n" +
 	"\x06GET_LK\x10\x02\x12\x17\n" +
 	"\x13RELEASE_POSIX_OWNER\x10\x03\x12\x17\n" +
-	"\x13RELEASE_FLOCK_OWNER\x10\x042\xbc\x16\n" +
+	"\x13RELEASE_FLOCK_OWNER\x10\x04\x12\x0e\n" +
+	"\n" +
+	"KEEP_ALIVE\x10\x052\xbc\x16\n" +
 	"\fSeaweedFiler\x12g\n" +
 	"\x14LookupDirectoryEntry\x12%.filer_pb.LookupDirectoryEntryRequest\x1a&.filer_pb.LookupDirectoryEntryResponse\"\x00\x12N\n" +
 	"\vListEntries\x12\x1c.filer_pb.ListEntriesRequest\x1a\x1d.filer_pb.ListEntriesResponse\"\x000\x01\x12L\n" +

weed/server/filer_grpc_server_posix_lock.go

@@ -3,6 +3,7 @@ package weed_server
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer/posixlock"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -10,6 +11,35 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 )
 
+const (
+	// posixLockSessionTTL is how long a mount's lease survives without a
+	// keepalive before its locks are reaped; posixLockSweepInterval is how often
+	// each filer checks. The mount renews well within the TTL.
+	posixLockSessionTTL    = 15 * time.Second
+	posixLockSweepInterval = 5 * time.Second
+)
+
+// startPosixLockSweeper periodically reaps the locks of leased sessions (mounts)
+// that stopped sending keepalives. Sessions that never renew are never reaped, so
+// this is inert until mounts run with -posixLock.
+func (fs *FilerServer) startPosixLockSweeper() {
+	fs.posixLockSweeperStop = make(chan struct{})
+	go func() {
+		ticker := time.NewTicker(posixLockSweepInterval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-fs.posixLockSweeperStop:
+				return
+			case <-ticker.C:
+				if reaped := fs.posixLocks.ReapExpired(posixLockSessionTTL); len(reaped) > 0 {
+					glog.V(2).Infof("posix lock: reaped %d expired session(s): %v", len(reaped), reaped)
+				}
+			}
+		}
+	}()
+}
+
 // PosixLock applies one advisory lock operation against the in-memory lock table
 // of the inode's owner filer. The owner is resolved from req.Key on the same
 // ring the gateway and DLM use; a non-owner filer forwards the request one hop
@@ -76,6 +106,8 @@ func (fs *FilerServer) PosixLock(ctx context.Context, req *filer_pb.PosixLockReq
 		fs.posixLocks.ReleasePosixOwner(req.Key, lk.Sid, lk.Owner)
 	case filer_pb.PosixLockOp_RELEASE_FLOCK_OWNER:
 		fs.posixLocks.ReleaseFlockOwner(req.Key, lk.Sid, lk.Owner)
+	case filer_pb.PosixLockOp_KEEP_ALIVE:
+		fs.posixLocks.Renew(lk.Sid)
 	default:
 		return &filer_pb.PosixLockResponse{}, fmt.Errorf("unknown posix lock op %v", req.Op)
 	}

weed/server/filer_grpc_server_posix_lock_test.go

@@ -88,6 +88,22 @@ func TestPosixLockReleasePosixOwnerKeepsFlock(t *testing.T) {
 	}
 }
 
+func TestPosixLockKeepAlive(t *testing.T) {
+	fs := newPosixTestServer()
+	resp, err := fs.PosixLock(context.Background(), &filer_pb.PosixLockRequest{
+		Key: "s3.fuse.lock:/x", Op: filer_pb.PosixLockOp_KEEP_ALIVE,
+		Lock: pbLock(0, 0, posixlock.Unlock, 7, 0, 0, false),
+	})
+	if err != nil || resp == nil {
+		t.Fatalf("keep_alive should succeed: err=%v", err)
+	}
+	// A renewed session that goes stale is reaped; a never-renewed one is not.
+	fs.posixLocks.TryLock("s3.fuse.lock:/x", posixlock.Range{Start: 0, End: 9, Type: posixlock.Write, Sid: 7, Owner: 1})
+	if reaped := fs.posixLocks.ReapExpired(0); len(reaped) != 1 || reaped[0] != 7 {
+		t.Fatalf("renewed session 7 should be reapable at ttl=0, got %v", reaped)
+	}
+}
+
 // A request whose key is owned by another filer is forwarded to it; the owner
 // applies it and the sender does not. The owner's ring points back at the bogus
 // sender, so without is_moved on the forwarded hop it would re-forward and fail.

weed/server/filer_server.go

@@ -139,6 +139,8 @@ type FilerServer struct {
 	// here rather than in replicated metadata: it is transient coordination, so
 	// keeping it off the meta-log avoids churn.
 	posixLocks *posixlock.Manager
+	// posixLockSweeperStop stops the lease-reaping sweeper goroutine on Shutdown.
+	posixLockSweeperStop chan struct{}
 }
 
 func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) (fs *FilerServer, err error) {
@@ -180,6 +182,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 		entryLockTable:        util.NewLockTable[util.FullPath](),
 		posixLocks:            posixlock.NewManager(),
 	}
+	fs.startPosixLockSweeper()
 	fs.mountPeerRegistry = filer.NewMountPeerRegistry()
 	go fs.runMountPeerRegistrySweeper()
 	fs.listenersCond = sync.NewCond(&fs.listenersLock)
@@ -322,6 +325,9 @@ func (fs *FilerServer) checkWithMaster() {
 // This prevents data corruption when the process receives SIGTERM during active uploads.
 func (fs *FilerServer) Shutdown() {
 	glog.V(0).Infof("Shutting down filer")
+	if fs.posixLockSweeperStop != nil {
+		close(fs.posixLockSweeperStop)
+	}
 	fs.filer.Shutdown()
 }