fix(mount): keep async flush when LockOwner has no POSIX locks (#9300)

FlushIn.LockOwner is populated by the kernel for any fd that may have
participated in locking, not only when locks were actually taken. The
previous Flush logic treated any non-zero LockOwner as a closing lock
holder and forced a synchronous flush, which silently disabled the
writebackCache async-flush path (introduced in #8727) for most
ordinary close() calls.

Consult the POSIX lock table before forcing sync: only owners that
currently hold a non-flock byte-range lock need the synchronous path
to coordinate with blocked SetLkw waiters. Other closes go async as
intended.

weed/mount/posix_file_lock.go

@@ -234,6 +234,31 @@ func (plt *PosixLockTable) releaseMatching(inode uint64, matches func(lockRange)
 	plt.maybeCleanupInode(inode, il)
 }
 
+// HasPosixOwner reports whether owner currently holds any POSIX byte-range
+// locks on inode. FUSE may provide a non-zero FlushIn.LockOwner even when no
+// locks were taken, so callers should consult the lock table before treating a
+// flush as lock-sensitive.
+func (plt *PosixLockTable) HasPosixOwner(inode uint64, owner uint64) bool {
+	if owner == 0 {
+		return false
+	}
+	il := plt.getInodeLocks(inode)
+	if il == nil {
+		return false
+	}
+	il.mu.Lock()
+	defer il.mu.Unlock()
+	if il.dead {
+		return false
+	}
+	for _, lk := range il.locks {
+		if !lk.IsFlock && lk.Owner == owner {
+			return true
+		}
+	}
+	return false
+}
+
 // releaseWakeRef drops the temporary reference that keeps inodeLocks live while
 // a woken waiter retries its SetLkw acquisition.
 func releaseWakeRef(il *inodeLocks, waiter *lockWaiter) {

weed/mount/posix_file_lock_test.go

@@ -274,6 +274,32 @@ func TestReleasePosixOwnerDoesNotReleaseFlockLocks(t *testing.T) {
 	}
 }
 
+func TestHasPosixOwnerIgnoresMissingOwnerAndFlock(t *testing.T) {
+	plt := NewPosixLockTable()
+	inode := uint64(1)
+
+	if plt.HasPosixOwner(inode, 1) {
+		t.Fatal("missing owner should not be reported as holding POSIX locks")
+	}
+
+	if s := plt.SetLk(inode, lockRange{Start: 0, End: math.MaxUint64, Typ: syscall.F_WRLCK, Owner: 1, Pid: 10, IsFlock: true}); s != fuse.OK {
+		t.Fatalf("set flock: %v", s)
+	}
+	if plt.HasPosixOwner(inode, 1) {
+		t.Fatal("flock owner should not be reported as a POSIX lock owner")
+	}
+
+	if s := plt.SetLk(inode, lockRange{Start: 0, End: 99, Typ: syscall.F_WRLCK, Owner: 2, Pid: 20}); s != fuse.OK {
+		t.Fatalf("set POSIX lock: %v", s)
+	}
+	if !plt.HasPosixOwner(inode, 2) {
+		t.Fatal("POSIX lock owner was not reported")
+	}
+	if plt.HasPosixOwner(inode, 0) {
+		t.Fatal("zero owner should not be reported")
+	}
+}
+
 func TestWakeEligibleWaitersKeepsInodeUntilWakeRefReleased(t *testing.T) {
 	plt := NewPosixLockTable()
 	inode := uint64(1)

weed/mount/weedfs_file_sync.go

@@ -65,9 +65,12 @@ func (wfs *WFS) Flush(cancel <-chan struct{}, in *fuse.FlushIn) fuse.Status {
 		return fuse.OK
 	}
 
-	// When a closing lock owner is present, flush synchronously before waking any
-	// blocked POSIX lock waiters so write-serialized callers cannot overtake each other.
-	allowAsync := in.LockOwner == 0
+	// FlushIn.LockOwner is populated by some FUSE kernels even when the process
+	// did not hold byte-range locks. Only force the synchronous close path when
+	// this owner actually has POSIX locks to release; otherwise writebackCache
+	// would silently degrade to a blocking flush for ordinary close().
+	hasPosixLocks := wfs.posixLocks.HasPosixOwner(in.NodeId, in.LockOwner)
+	allowAsync := !hasPosixLocks
 	status := wfs.doFlush(fh, in.Uid, in.Gid, allowAsync)
 	if in.LockOwner != 0 {
 		wfs.posixLocks.ReleasePosixOwner(in.NodeId, in.LockOwner)