mount: let the release test's parked flush talk to a live filer

Marking the handle deleted raced with the flush's unlocked early read
of isDeleted. Pointing the fixture at a live fake filer lets the flush
complete on its own once the handle lock frees, no flag write needed.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/mount/weedfs_file_io.go

@@ -140,6 +140,12 @@ func (wfs *WFS) invalidateOpenMtimeCache(inode uint64) {
 }
 
 func (wfs *WFS) Release(cancel <-chan struct{}, in *fuse.ReleaseIn) {
+	// Drop the file description's flock locks first: close() frees them
+	// regardless of how writeback fares, and a flush stuck on an unreachable
+	// filer must not pin a dead holder's lock in the table.
+	if in.ReleaseFlags&fuse.FUSE_RELEASE_FLOCK_UNLOCK != 0 {
+		wfs.releaseFlockOwner(in.NodeId, in.LockOwner)
+	}
 	// Flush is usually sent before Release, but the FUSE protocol does not
 	// guarantee it. Route every Release through doFlush so a dirty handle
 	// (e.g. a deferred create with no intervening Flush) is not dropped.
@@ -153,8 +159,5 @@ func (wfs *WFS) Release(cancel <-chan struct{}, in *fuse.ReleaseIn) {
 			glog.Warningf("release fh %d inode %d: fallback flush failed: %v", in.Fh, in.NodeId, status)
 		}
 	}
-	if in.ReleaseFlags&fuse.FUSE_RELEASE_FLOCK_UNLOCK != 0 {
-		wfs.releaseFlockOwner(in.NodeId, in.LockOwner)
-	}
 	wfs.ReleaseHandle(FileHandleId(in.Fh))
 }

weed/mount/weedfs_release_flock_test.go

@@ -0,0 +1,77 @@
+package mount
+
+import (
+	"math"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/go-fuse/v2/fuse"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// A close() frees the file description's flock locks even when the handle
+// flush is stuck — e.g. a filer that stopped answering. If the locks outlive
+// the blocked flush, a restarted process taking the same file's flock sees
+// "file already locked" until the mount restarts.
+func TestReleaseDropsFlockWhileFlushBlocked(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+	wfs.posixLocks = NewPosixLockTable()
+	startFakeFiler(t, wfs, &fakeFilerServer{})
+
+	inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/f"), time.Now().Unix(), false, false, 0, false)
+	fh, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, &filer_pb.Entry{
+		Name:       "f",
+		Attributes: &filer_pb.FuseAttributes{},
+	}, 0, 0)
+	fh.dirtyMetadata = true
+
+	const owner = uint64(42)
+	if st := wfs.posixLocks.SetLk(inode, lockRange{Start: 0, End: math.MaxUint64, Typ: f_WRLCK, Owner: owner, IsFlock: true}); st != fuse.OK {
+		t.Fatalf("take flock: %v", st)
+	}
+
+	// Stall the release flush at the handle lock flushMetadataToFiler waits on.
+	active := wfs.fhLockTable.AcquireLock("test", fh.fh, util.ExclusiveLock)
+	released := false
+	releaseLock := func() {
+		if !released {
+			wfs.fhLockTable.ReleaseLock(fh.fh, active)
+			released = true
+		}
+	}
+	defer releaseLock()
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		wfs.Release(nil, &fuse.ReleaseIn{
+			InHeader:     fuse.InHeader{NodeId: inode},
+			Fh:           uint64(fh.fh),
+			LockOwner:    owner,
+			ReleaseFlags: fuse.FUSE_RELEASE_FLOCK_UNLOCK,
+		})
+	}()
+
+	// The dead holder's flock must be gone while the flush is still parked.
+	deadline := time.Now().Add(10 * time.Second)
+	for {
+		var out fuse.LkOut
+		wfs.posixLocks.GetLk(inode, lockRange{Start: 0, End: math.MaxUint64, Typ: f_WRLCK, Owner: 7, IsFlock: true}, &out)
+		if out.Lk.Typ == f_UNLCK {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("flock still held while the release flush is blocked")
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+
+	// Let the parked flush through: the live filer answers the metadata update.
+	releaseLock()
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		t.Fatal("release never completed after the flush unblocked")
+	}
+}