mount: resolve queued invalidations against the local store

Invalidations apply asynchronously, so the entry an event carries can be
a stale snapshot by the time it reaches the handle: a local flush may
install newer state while the event sits in the queue, and the flush's
own event is dedup-suppressed, so rolling the handle back would never
heal. The apply loop has already ordered the event and any later state
into the local store, so resolve the refresh there first; fall back to
the event entry only for read-through directories, where the store
holds nothing and the event is the freshest ordered information
available.

weed/mount/weedfs.go

@@ -677,12 +677,11 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, fuse.Status)
 }
 
 // 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.
+// subscription event. No filer lookup happens here: it can fail transiently,
+// 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 {
@@ -698,6 +697,17 @@ func (wfs *WFS) invalidateOpenFileHandle(filePath util.FullPath, entry *filer_pb
 	fh.dirtyPages.Destroy()
 	fh.dirtyPages = newPageWriter(fh, wfs.option.ChunkSizeLimit)
 
+	// Invalidations apply asynchronously, so the event entry may be a stale
+	// snapshot by now: a local flush can install newer state while the event
+	// sits in the queue, and the flush's own event is dedup-suppressed, so a
+	// rollback would never heal. The apply loop has already ordered this event
+	// and any later state into the local store, so prefer the store's entry;
+	// it misses only for read-through directories, where the event entry is
+	// the freshest ordered information available.
+	if localEntry, findErr := wfs.metaCache.FindEntry(context.Background(), filePath); findErr == nil && localEntry != nil {
+		fh.SetEntry(localEntry.ToProtoEntry())
+		return
+	}
 	if entry == nil {
 		return
 	}

weed/mount/weedfs_invalidate_open_handle_test.go

@@ -119,3 +119,60 @@ func TestUpdateEventRefreshesOpenFileHandle(t *testing.T) {
 		t.Fatalf("open handle file size after delete = %d, want 180020", size)
 	}
 }
+
+// A queued invalidation must not roll the handle back over newer state a
+// local flush installed while the event sat in the queue. The local store is
+// ordered by the apply loop, so it resolves the refresh for cached
+// directories.
+func TestQueuedEventDoesNotRollBackNewerLocalState(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+
+	wfs.inodeToPath.MarkChildrenCached(util.FullPath("/"))
+	wfs.inodeToPath.Lookup(util.FullPath("/dir"), time.Now().Unix(), true, false, 0, false)
+	wfs.inodeToPath.MarkChildrenCached(util.FullPath("/dir"))
+
+	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},
+	})
+
+	updateEvent := func(size uint64) *filer_pb.SubscribeMetadataResponse {
+		return &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: size},
+				},
+				NewParentPath: "/dir",
+			},
+		}
+	}
+
+	// Hold the handle lock so the queued invalidation cannot apply yet.
+	testLock := wfs.fhLockTable.AcquireLock("test", fh.fh, util.ExclusiveLock)
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), updateEvent(100), meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		wfs.fhLockTable.ReleaseLock(fh.fh, testLock)
+		t.Fatalf("apply subscriber event: %v", err)
+	}
+
+	// A local flush lands after the event was queued: newer state goes into
+	// the handle and, via the local apply, into the local store.
+	fh.SetEntry(&filer_pb.Entry{
+		Name:       "file",
+		Attributes: &filer_pb.FuseAttributes{FileSize: 200},
+	})
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), updateEvent(200), meta_cache.LocalMetadataResponseApplyOptions); err != nil {
+		wfs.fhLockTable.ReleaseLock(fh.fh, testLock)
+		t.Fatalf("apply local event: %v", err)
+	}
+	wfs.fhLockTable.ReleaseLock(fh.fh, testLock)
+
+	wfs.metaCache.WaitForEntryInvalidations()
+
+	if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 200 {
+		t.Fatalf("open handle file size = %d, want 200 (queued size-100 event must not roll back the newer local state)", size)
+	}
+}