mount: refresh open file handles from the metadata event entry

The invalidation callback looked the path up again after the
subscription event arrived. A transient lookup failure, or a stale
cached result, left the open handle pinned to its old entry with no
retry: the cursor had already advanced, so only another event for the
same path would recover it. Apply the entry the event itself carries:
NewEntry for in-place updates and rename destinations, and keep the
last entry when the path is vacated so unlinked-but-open reads still
work.

weed/mount/meta_cache/meta_cache.go

@@ -564,7 +564,7 @@ func (mc *MetaCache) handleApplyRequest(req metadataApplyRequest) error {
 
 type metadataInvalidation struct {
 	path  util.FullPath
-	entry *filer_pb.Entry
+	entry *filer_pb.Entry // entry now at path per the event; nil when the path was vacated (delete, rename away)
 }
 
 type metadataResponseSideEffects struct {
@@ -975,15 +975,17 @@ func collectEntryInvalidations(resp *filer_pb.SubscribeMetadataResponse) []metad
 	var invalidations []metadataInvalidation
 	if message.OldEntry != nil && message.NewEntry != nil {
 		oldKey := util.NewFullPath(resp.Directory, message.OldEntry.Name)
-		invalidations = append(invalidations, metadataInvalidation{path: oldKey, entry: message.OldEntry})
 		// Normalize NewParentPath: empty means same directory as resp.Directory
 		newDir := resp.Directory
 		if message.NewParentPath != "" {
 			newDir = message.NewParentPath
 		}
 		if message.OldEntry.Name != message.NewEntry.Name || resp.Directory != newDir {
+			invalidations = append(invalidations, metadataInvalidation{path: oldKey})
 			newKey := util.NewFullPath(newDir, message.NewEntry.Name)
 			invalidations = append(invalidations, metadataInvalidation{path: newKey, entry: message.NewEntry})
+		} else {
+			invalidations = append(invalidations, metadataInvalidation{path: oldKey, entry: message.NewEntry})
 		}
 		return invalidations
 	}
@@ -999,7 +1001,7 @@ func collectEntryInvalidations(resp *filer_pb.SubscribeMetadataResponse) []metad
 
 	if filer_pb.IsDelete(resp) && message.OldEntry != nil {
 		oldKey := util.NewFullPath(resp.Directory, message.OldEntry.Name)
-		invalidations = append(invalidations, metadataInvalidation{path: oldKey, entry: message.OldEntry})
+		invalidations = append(invalidations, metadataInvalidation{path: oldKey})
 	}
 
 	return invalidations

weed/mount/meta_cache/meta_cache_apply_test.go

@@ -409,6 +409,65 @@ func TestApplyMetadataResponsePurgesHiddenDestinationPath(t *testing.T) {
 	}
 }
 
+// The entry attached to each invalidation is what an open file handle gets
+// refreshed with, so it must be the entry now at that path — or nil when the
+// path was vacated and the handle should keep its last entry.
+func TestCollectEntryInvalidationsCarryAuthoritativeEntries(t *testing.T) {
+	newEntry := &filer_pb.Entry{
+		Name:       "file.txt",
+		Attributes: &filer_pb.FuseAttributes{FileSize: 42},
+	}
+
+	update := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry:      &filer_pb.Entry{Name: "file.txt"},
+			NewEntry:      newEntry,
+			NewParentPath: "/dir",
+		},
+	}
+	got := collectEntryInvalidations(update)
+	if len(got) != 1 || got[0].path != "/dir/file.txt" || got[0].entry != newEntry {
+		t.Fatalf("in-place update invalidations = %+v, want [{/dir/file.txt NewEntry}]", got)
+	}
+
+	rename := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/src",
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry:      &filer_pb.Entry{Name: "file.tmp"},
+			NewEntry:      newEntry,
+			NewParentPath: "/dst",
+		},
+	}
+	got = collectEntryInvalidations(rename)
+	if len(got) != 2 || got[0].path != "/src/file.tmp" || got[0].entry != nil ||
+		got[1].path != "/dst/file.txt" || got[1].entry != newEntry {
+		t.Fatalf("rename invalidations = %+v, want [{/src/file.tmp nil} {/dst/file.txt NewEntry}]", got)
+	}
+
+	create := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: newEntry,
+		},
+	}
+	got = collectEntryInvalidations(create)
+	if len(got) != 1 || got[0].path != "/dir/file.txt" || got[0].entry != newEntry {
+		t.Fatalf("create invalidations = %+v, want [{/dir/file.txt NewEntry}]", got)
+	}
+
+	deleteResp := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "file.txt"},
+		},
+	}
+	got = collectEntryInvalidations(deleteResp)
+	if len(got) != 1 || got[0].path != "/dir/file.txt" || got[0].entry != nil {
+		t.Fatalf("delete invalidations = %+v, want [{/dir/file.txt nil}]", got)
+	}
+}
+
 func newTestMetaCache(t *testing.T, cached map[util.FullPath]bool) (*MetaCache, map[util.FullPath]bool, *recordedPaths, *recordedPaths) {
 	t.Helper()
 

weed/mount/weedfs.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/seaweedfs/go-fuse/v2/fuse"
 	"google.golang.org/grpc"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/seaweedfs/seaweedfs/weed/cluster"
 	"github.com/seaweedfs/seaweedfs/weed/filer"
@@ -302,26 +303,7 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 		}, func(path util.FullPath) bool {
 			return wfs.inodeToPath.IsChildrenCached(path)
 		}, func(filePath util.FullPath, entry *filer_pb.Entry) {
-			// Find inode if it is not a deleted path
-			if inode, inodeFound := wfs.inodeToPath.GetInode(filePath); inodeFound {
-				// Find open file handle
-				if fh, fhFound := wfs.fhMap.FindFileHandle(inode); fhFound {
-					fhActiveLock := fh.wfs.fhLockTable.AcquireLock("invalidateFunc", fh.fh, util.ExclusiveLock)
-					defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock)
-
-					// Recreate dirty pages
-					fh.dirtyPages.Destroy()
-					fh.dirtyPages = newPageWriter(fh, wfs.option.ChunkSizeLimit)
-
-					// Update handle entry
-					newEntry, status := wfs.maybeLoadEntry(filePath)
-					if status == fuse.OK {
-						if fh.GetEntry().GetEntry() != newEntry {
-							fh.SetEntry(newEntry)
-						}
-					}
-				}
-			}
+			wfs.invalidateOpenFileHandle(filePath, entry)
 		}, func(dirPath util.FullPath) {
 			if wfs.inodeToPath.RecordDirectoryUpdate(dirPath, time.Now(), wfs.dirHotWindow, wfs.dirHotThreshold) {
 				wfs.markDirectoryReadThrough(dirPath)
@@ -694,6 +676,41 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, fuse.Status)
 	return filer.FromPbEntry(dir, entry), fuse.OK
 }
 
+// invalidateOpenFileHandle refreshes an open file handle from a metadata
+// subscription event. The event entry is applied directly: a second lookup
+// here can fail transiently or serve stale cached metadata, and since the
+// subscription cursor has already advanced past the event, the handle would
+// stay pinned to its old entry until an unrelated event arrives. A nil entry
+// means the path no longer holds one (delete, rename away); the handle keeps
+// its last entry so unlinked-but-open reads still work.
+func (wfs *WFS) invalidateOpenFileHandle(filePath util.FullPath, entry *filer_pb.Entry) {
+	inode, inodeFound := wfs.inodeToPath.GetInode(filePath)
+	if !inodeFound {
+		return
+	}
+	fh, fhFound := wfs.fhMap.FindFileHandle(inode)
+	if !fhFound {
+		return
+	}
+	fhActiveLock := wfs.fhLockTable.AcquireLock("invalidateFunc", fh.fh, util.ExclusiveLock)
+	defer wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock)
+
+	fh.dirtyPages.Destroy()
+	fh.dirtyPages = newPageWriter(fh, wfs.option.ChunkSizeLimit)
+
+	if entry == nil {
+		return
+	}
+	newEntry := proto.Clone(entry).(*filer_pb.Entry)
+	if newEntry.Attributes == nil {
+		newEntry.Attributes = &filer_pb.FuseAttributes{}
+	}
+	if wfs.option.UidGidMapper != nil {
+		newEntry.Attributes.Uid, newEntry.Attributes.Gid = wfs.option.UidGidMapper.FilerToLocal(newEntry.Attributes.Uid, newEntry.Attributes.Gid)
+	}
+	fh.SetEntry(newEntry)
+}
+
 func (wfs *WFS) LookupFn() wdclient.LookupFileIdFunctionType {
 	if wfs.option.VolumeServerAccess == "filerProxy" {
 		return func(ctx context.Context, fileId string) (targetUrls []string, err error) {

weed/mount/weedfs_invalidate_open_handle_test.go

@@ -0,0 +1,121 @@
+package mount
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/mount/meta_cache"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+func newInvalidateTestWFS(t *testing.T) *WFS {
+	t.Helper()
+
+	// Map filer uid 2000 to local uid 1000 to verify the event entry gets the
+	// same id translation a filer lookup would apply.
+	uidGidMapper, err := meta_cache.NewUidGidMapper("1000:2000", "")
+	if err != nil {
+		t.Fatalf("create uid/gid mapper: %v", err)
+	}
+
+	root := util.FullPath("/")
+	wfs := &WFS{
+		signature:         1,
+		inodeToPath:       NewInodeToPath(root, 0),
+		fhMap:             NewFileHandleToInode(),
+		fhLockTable:       util.NewLockTable[FileHandleId](),
+		hardLinkLockTable: util.NewLockTable[string](),
+		option: &Option{
+			ChunkSizeLimit:     1024,
+			ConcurrentReaders:  1,
+			VolumeServerAccess: "filerProxy",
+			// Nothing listens here: any secondary lookup during invalidation
+			// fails, like the transient filer error that pins a handle.
+			FilerAddresses: []pb.ServerAddress{
+				pb.NewServerAddressWithGrpcPort("127.0.0.1:1", 1),
+			},
+			GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
+			UidGidMapper:   uidGidMapper,
+		},
+	}
+
+	wfs.metaCache = meta_cache.NewMetaCache(
+		filepath.Join(t.TempDir(), "meta"),
+		uidGidMapper,
+		root,
+		false,
+		func(path util.FullPath) { wfs.inodeToPath.MarkChildrenCached(path) },
+		func(path util.FullPath) bool { return wfs.inodeToPath.IsChildrenCached(path) },
+		wfs.invalidateOpenFileHandle,
+		nil,
+	)
+	t.Cleanup(wfs.metaCache.Shutdown)
+
+	return wfs
+}
+
+// An update event must refresh an open file handle from the entry the event
+// itself carries. A second lookup can fail transiently or serve stale cached
+// metadata, and the subscription cursor has already advanced, so a missed
+// refresh leaves the handle pinned to the old entry until an unrelated event
+// for the same path arrives.
+func TestUpdateEventRefreshesOpenFileHandle(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+
+	inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false)
+	fh := wfs.fhMap.AcquireFileHandle(wfs, inode, &filer_pb.Entry{
+		Name:       "file",
+		Attributes: &filer_pb.FuseAttributes{FileSize: 88},
+	})
+
+	updateResp := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "file"},
+			NewEntry: &filer_pb.Entry{
+				Name:       "file",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 180020, Uid: 2000},
+				Chunks:     []*filer_pb.FileChunk{{FileId: "1,ab1", Size: 180020}},
+			},
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), updateResp, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply update event: %v", err)
+	}
+	wfs.metaCache.WaitForEntryInvalidations()
+
+	entry := fh.GetEntry().GetEntry()
+	if entry.Attributes.FileSize != 180020 {
+		t.Fatalf("open handle file size = %d, want 180020", entry.Attributes.FileSize)
+	}
+	if len(entry.GetChunks()) != 1 {
+		t.Fatalf("open handle chunks = %d, want 1", len(entry.GetChunks()))
+	}
+	if entry.Attributes.Uid != 1000 {
+		t.Fatalf("open handle uid = %d, want filer uid 2000 mapped to local 1000", entry.Attributes.Uid)
+	}
+
+	// A delete leaves the handle with its last entry so unlinked-but-open
+	// reads keep working.
+	deleteResp := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "file"},
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), deleteResp, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply delete event: %v", err)
+	}
+	wfs.metaCache.WaitForEntryInvalidations()
+
+	if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 180020 {
+		t.Fatalf("open handle file size after delete = %d, want 180020", size)
+	}
+}