mount: stable-read open fence, causal log timestamp on the cache response

An event applied while the open's lookup was in flight could already be
reflected in the returned entry while sitting above the pre-lookup
cursor, so the fence under-covered and the queued invalidation replaced
the fresher entry. The open now re-reads until the cursor is stable
across the lookup; on sustained churn the last pre-lookup value stands,
which can only under-fence, never block a newer event.

A pre-RPC ping likewise cannot vouch for events committed during the
RPC itself. CacheRemoteObjectToLocalCluster now returns a log timestamp
stamped before the filer reads the entry — events are logged after
their store write on that same clock, so everything at or below it is
reflected in the returned entry, making the fence causal with the
response. The mount prefers the response event's timestamp, then this
log timestamp, and keeps the pre-RPC ping only for filers that return
neither.

weed/mount/filehandle.go

@@ -3,6 +3,7 @@ package mount
 import (
 	"os"
 	"sync"
+	"sync/atomic"
 
 	"github.com/seaweedfs/go-fuse/v2/fuse"
 	"github.com/seaweedfs/seaweedfs/weed/cluster"
@@ -39,6 +40,13 @@ type FileHandle struct {
 	isDeleted bool
 	isRenamed bool // set by Rename before waiting for async flush; skips old-path metadata flush
 
+	// lastLocalEntryTsNs is the filer log timestamp of the newest filer-
+	// acknowledged local mutation reflected in this handle's entry (from the
+	// metadata event a CreateEntry/CacheRemoteObject response carries).
+	// Subscription events at or before this timestamp are old news for the
+	// handle and must not roll it back.
+	lastLocalEntryTsNs atomic.Int64
+
 	// dlmLock holds the distributed lock for cross-mount write coordination.
 	// Non-nil only when -dlm is enabled and the file was opened for writing.
 	// Acquired in AcquireHandle, released in ReleaseHandle.
@@ -119,6 +127,34 @@ func (fh *FileHandle) SetEntry(entry *filer_pb.Entry) {
 	fh.invalidateChunkCache()
 }
 
+// noteFilerAck advances the watermark after a filer RPC acknowledged state
+// now reflected in this handle: to the response event's log timestamp when
+// one was returned, otherwise to baselineTsNs — the latest filer log position
+// known before the RPC was issued, a safe lower bound for the state the
+// filer served.
+func (fh *FileHandle) noteFilerAck(baselineTsNs int64, event *filer_pb.SubscribeMetadataResponse) {
+	if tsNs := event.GetTsNs(); tsNs != 0 {
+		fh.advanceLocalEntryTs(tsNs)
+		return
+	}
+	fh.advanceLocalEntryTs(baselineTsNs)
+}
+
+// advanceLocalEntryTs records the filer log timestamp of a filer-acknowledged
+// local mutation now reflected in the handle's entry. Monotonic: an older
+// timestamp never regresses the watermark.
+func (fh *FileHandle) advanceLocalEntryTs(tsNs int64) {
+	if tsNs == 0 {
+		return
+	}
+	for {
+		current := fh.lastLocalEntryTsNs.Load()
+		if tsNs <= current || fh.lastLocalEntryTsNs.CompareAndSwap(current, tsNs) {
+			return
+		}
+	}
+}
+
 func (fh *FileHandle) ResetDirtyPages() {
 	fh.dirtyPages.Destroy()
 	fh.dirtyPages = newPageWriter(fh, fh.wfs.option.ChunkSizeLimit)

weed/mount/filehandle_read.go

@@ -188,12 +188,22 @@ func (fh *FileHandle) downloadRemoteEntry(entry *LockedEntry) error {
 		}
 
 		glog.V(4).Infof("download entry: %v", request)
+		// Barrier through this closure's client: on failover WithFilerClient
+		// retries against another filer while the current-filer index still
+		// points at the failed one. Only a fallback for old filers — the
+		// response's own log timestamp is causal with the returned entry,
+		// while a pre-RPC ping cannot cover events committed during the RPC.
+		baselineTsNs := fh.wfs.filerBarrierTsNsWith(client)
 		resp, err := client.CacheRemoteObjectToLocalCluster(context.Background(), request)
 		if err != nil {
 			return fmt.Errorf("CacheRemoteObjectToLocalCluster file %s: %v", fileFullPath, err)
 		}
 
 		fh.SetEntry(resp.Entry)
+		if resp.GetLogTsNs() > baselineTsNs {
+			baselineTsNs = resp.GetLogTsNs()
+		}
+		fh.noteFilerAck(baselineTsNs, resp.GetMetadataEvent())
 
 		// Async: a sync apply deadlocks against the apply loop's invalidate, which needs this read's file-handle lock.
 		event := resp.GetMetadataEvent()

weed/mount/filehandle_read_remote_test.go

@@ -93,7 +93,7 @@ func TestReadUncachedRemoteEntryDoesNotDeadlock(t *testing.T) {
 		func(path util.FullPath) { wfs.inodeToPath.MarkChildrenCached(path) },
 		func(path util.FullPath) bool { return wfs.inodeToPath.IsChildrenCached(path) },
 		// Mirror weedfs.go's invalidateFunc: take the file handle exclusive lock.
-		func(path util.FullPath, _ *filer_pb.Entry) {
+		func(path util.FullPath, _ *filer_pb.Entry, _ int64) {
 			inode, ok := wfs.inodeToPath.GetInode(path)
 			if !ok {
 				return

weed/mount/meta_cache/meta_cache.go

@@ -6,6 +6,7 @@ import (
 	"math"
 	"os"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"golang.org/x/sync/singleflight"
@@ -31,7 +32,7 @@ type MetaCache struct {
 	uidGidMapper         *UidGidMapper
 	markCachedFn         func(fullpath util.FullPath)
 	isCachedFn           func(fullpath util.FullPath) bool
-	invalidateFunc       func(fullpath util.FullPath, entry *filer_pb.Entry)
+	invalidateFunc       func(fullpath util.FullPath, entry *filer_pb.Entry, eventTsNs int64)
 	onDirectoryUpdate    func(dir util.FullPath)
 	pinnedChildFn        func(*filer.Entry) bool // a child a rebuild must not drop (local-only, not yet on the filer); nil disables
 	visitGroup           singleflight.Group      // deduplicates concurrent EnsureVisited calls for the same path
@@ -42,6 +43,7 @@ type MetaCache struct {
 	buildingDirs         map[util.FullPath]*directoryBuildState
 	dedupRing            dedupRingBuffer
 	includeSystemEntries bool
+	latestEventTsNs      atomic.Int64 // newest filer log timestamp seen in any applied event
 
 	// Entry invalidations run on a worker, not inline on the apply loop:
 	// invalidateFunc takes the fh lock, which a flush can hold while waiting on
@@ -96,7 +98,7 @@ type metadataApplyRequest struct {
 }
 
 func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPath, includeSystemEntries bool,
-	markCachedFn func(path util.FullPath), isCachedFn func(path util.FullPath) bool, invalidateFunc func(util.FullPath, *filer_pb.Entry), onDirectoryUpdate func(dir util.FullPath)) *MetaCache {
+	markCachedFn func(path util.FullPath), isCachedFn func(path util.FullPath) bool, invalidateFunc func(util.FullPath, *filer_pb.Entry, int64), onDirectoryUpdate func(dir util.FullPath)) *MetaCache {
 	leveldbStore, virtualStore := openMetaStore(dbFolder)
 	mc := &MetaCache{
 		root:                 root,
@@ -107,8 +109,8 @@ func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPat
 		uidGidMapper:         uidGidMapper,
 		onDirectoryUpdate:    onDirectoryUpdate,
 		includeSystemEntries: includeSystemEntries,
-		invalidateFunc: func(fullpath util.FullPath, entry *filer_pb.Entry) {
-			invalidateFunc(fullpath, entry)
+		invalidateFunc: func(fullpath util.FullPath, entry *filer_pb.Entry, eventTsNs int64) {
+			invalidateFunc(fullpath, entry, eventTsNs)
 		},
 		applyCh:      make(chan metadataApplyRequest, 128),
 		applyDone:    make(chan struct{}),
@@ -117,7 +119,7 @@ func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPat
 	}
 	mc.invalidateWorker = util.NewAsyncBatchWorker(func(batch []metadataInvalidation) {
 		for _, invalidation := range batch {
-			mc.invalidateFunc(invalidation.path, invalidation.entry)
+			mc.invalidateFunc(invalidation.path, invalidation.entry, invalidation.tsNs)
 		}
 	})
 	go mc.runApplyLoop()
@@ -564,7 +566,8 @@ 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)
+	tsNs  int64           // the event's filer log timestamp; 0 for locally built events
 }
 
 type metadataResponseSideEffects struct {
@@ -582,6 +585,20 @@ func (mc *MetaCache) applyMetadataResponseNow(ctx context.Context, resp *filer_p
 		return mc.applyMetadataResponseDirect(ctx, resp, options, false)
 	}
 
+	for _, immediateEvent := range immediateEvents {
+		if err := mc.applyMetadataResponseDirect(ctx, immediateEvent, MetadataResponseApplyOptions{}, false); err != nil {
+			return err
+		}
+	}
+	// The cursor must cover a buffered event before its invalidation is
+	// observable, or an open racing the queue fences too low and the event
+	// later replaces newer state — a buffered event may never reach
+	// applyMetadataResponseDirect (build abort, snapshot-covered on
+	// completion). Sound without the store write: the building directory is
+	// read-through, so opens there consult the filer, which is at least this
+	// new. Immediate fragments were applied above, so their store writes are
+	// not outrun either.
+	mc.advanceLatestEventTs(resp.TsNs)
 	// Apply side effects but skip directory notifications for dirs that are
 	// currently being built. Notifying a building dir can trigger
 	// markDirectoryReadThrough → DeleteFolderChildren, wiping entries that
@@ -594,11 +611,6 @@ func (mc *MetaCache) applyMetadataResponseNow(ctx context.Context, resp *filer_p
 		}
 		state.bufferedEvents = append(state.bufferedEvents, events...)
 	}
-	for _, immediateEvent := range immediateEvents {
-		if err := mc.applyMetadataResponseDirect(ctx, immediateEvent, MetadataResponseApplyOptions{}, false); err != nil {
-			return err
-		}
-	}
 	return nil
 }
 
@@ -606,6 +618,10 @@ func (mc *MetaCache) applyMetadataResponseDirect(ctx context.Context, resp *file
 	if _, err := mc.applyMetadataResponseLocked(ctx, resp, options, allowUncachedInsert); err != nil {
 		return err
 	}
+	// Advance only after the store write: readers pairing a cursor capture
+	// with a store read (the open-time handle fence) rely on the cursor never
+	// leading the store.
+	mc.advanceLatestEventTs(resp.TsNs)
 	mc.applyMetadataSideEffects(resp, options)
 	return nil
 }
@@ -650,6 +666,26 @@ func (mc *MetaCache) WaitForEntryInvalidations() {
 	mc.invalidateWorker.Drain()
 }
 
+func (mc *MetaCache) advanceLatestEventTs(tsNs int64) {
+	if tsNs == 0 {
+		return
+	}
+	for {
+		current := mc.latestEventTsNs.Load()
+		if tsNs <= current || mc.latestEventTsNs.CompareAndSwap(current, tsNs) {
+			return
+		}
+	}
+}
+
+// LatestEventTsNs returns the newest filer log timestamp seen in any applied
+// event. A filer RPC issued after reading this serves state at least this
+// new, so it is a safe watermark baseline when the RPC response carries no
+// metadata event of its own.
+func (mc *MetaCache) LatestEventTsNs() int64 {
+	return mc.latestEventTsNs.Load()
+}
+
 func (mc *MetaCache) applyMetadataResponseLocked(ctx context.Context, resp *filer_pb.SubscribeMetadataResponse, _ MetadataResponseApplyOptions, allowUncachedInsert bool) (metadataResponseSideEffects, error) {
 	message := resp.GetEventNotification()
 	if message == nil {
@@ -755,6 +791,15 @@ func (mc *MetaCache) completeDirectoryBuildNow(ctx context.Context, dirPath util
 	}
 
 	mc.markCachedFn(dirPath)
+
+	// Re-invalidate every buffered event, replayed or snapshot-covered: the
+	// invalidation issued when an event arrived ran against a mid-build store
+	// that could miss the path entirely or predate the listing insert, so an
+	// open handle can hold older state than the completed directory. Enqueued
+	// after markCachedFn so the refresh resolves against the published store.
+	for _, event := range state.bufferedEvents {
+		mc.applyMetadataSideEffects(event, MetadataResponseApplyOptions{InvalidateEntries: true})
+	}
 	return nil
 }
 
@@ -975,15 +1020,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, tsNs: resp.TsNs})
 			newKey := util.NewFullPath(newDir, message.NewEntry.Name)
-			invalidations = append(invalidations, metadataInvalidation{path: newKey, entry: message.NewEntry})
+			invalidations = append(invalidations, metadataInvalidation{path: newKey, entry: message.NewEntry, tsNs: resp.TsNs})
+		} else {
+			invalidations = append(invalidations, metadataInvalidation{path: oldKey, entry: message.NewEntry, tsNs: resp.TsNs})
 		}
 		return invalidations
 	}
@@ -994,12 +1041,12 @@ func collectEntryInvalidations(resp *filer_pb.SubscribeMetadataResponse) []metad
 			newDir = message.NewParentPath
 		}
 		newKey := util.NewFullPath(newDir, message.NewEntry.Name)
-		invalidations = append(invalidations, metadataInvalidation{path: newKey, entry: message.NewEntry})
+		invalidations = append(invalidations, metadataInvalidation{path: newKey, entry: message.NewEntry, tsNs: resp.TsNs})
 	}
 
 	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, tsNs: resp.TsNs})
 	}
 
 	return invalidations

weed/mount/meta_cache/meta_cache_apply_test.go

@@ -409,6 +409,66 @@ 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",
+		TsNs:      77,
+		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 || got[0].tsNs != 77 {
+		t.Fatalf("in-place update invalidations = %+v, want [{/dir/file.txt NewEntry ts 77}]", 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()
 
@@ -436,7 +496,7 @@ func newTestMetaCache(t *testing.T, cached map[util.FullPath]bool) (*MetaCache,
 			defer cachedMu.Unlock()
 			return cached[path]
 		},
-		func(path util.FullPath, entry *filer_pb.Entry) {
+		func(path util.FullPath, entry *filer_pb.Entry, eventTsNs int64) {
 			invalidations.record(path)
 		},
 		func(dir util.FullPath) {

weed/mount/meta_cache/meta_cache_deadlock_test.go

@@ -58,7 +58,7 @@ func TestApplyLoopInvalidateDoesNotDeadlockWithLockHoldingEnqueuer(t *testing.T)
 			defer cachedMu.Unlock()
 			return cached[path]
 		},
-		func(path util.FullPath, entry *filer_pb.Entry) {
+		func(path util.FullPath, entry *filer_pb.Entry, _ int64) {
 			// Mirrors the wfs invalidateFunc: it takes the open file
 			// handle's exclusive lock before refreshing the handle entry.
 			enteredOnce.Do(func() { close(invalidateEntered) })

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"
@@ -301,27 +302,8 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 			wfs.inodeToPath.MarkChildrenCached(path)
 		}, 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)
-						}
-					}
-				}
-			}
+		}, func(filePath util.FullPath, entry *filer_pb.Entry, eventTsNs int64) {
+			wfs.invalidateOpenFileHandle(filePath, entry, eventTsNs)
 		}, func(dirPath util.FullPath) {
 			if wfs.inodeToPath.RecordDirectoryUpdate(dirPath, time.Now(), wfs.dirHotWindow, wfs.dirHotThreshold) {
 				wfs.markDirectoryReadThrough(dirPath)
@@ -694,6 +676,102 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, fuse.Status)
 	return filer.FromPbEntry(dir, entry), fuse.OK
 }
 
+// latestKnownFilerTsNs is the newest filer log timestamp seen in any applied
+// event — the watermark baseline for filer RPCs whose response carries no
+// metadata event. Nil-safe for partially constructed test instances.
+func (wfs *WFS) latestKnownFilerTsNs() int64 {
+	if wfs.metaCache == nil {
+		return 0
+	}
+	return wfs.metaCache.LatestEventTsNs()
+}
+
+// filerBarrierTsNsWith returns a timestamp at or below the filer's current
+// log position, read through the same client the fenced operation uses — a
+// barrier from a different filer would not vouch for the state that filer
+// serves. Metadata events are stamped with the filer clock, and a self-ping
+// reads that same clock, so unlike latestKnownFilerTsNs this also covers
+// events already committed but not yet delivered to the subscription. Call
+// before the RPC whose result the barrier fences. Best-effort: one bounded
+// attempt, falling back to the newest delivered event.
+func (wfs *WFS) filerBarrierTsNsWith(client filer_pb.SeaweedFilerClient) int64 {
+	baseline := wfs.latestKnownFilerTsNs()
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	if resp, err := client.Ping(ctx, &filer_pb.PingRequest{}); err == nil && resp.StartTimeNs > baseline {
+		baseline = resp.StartTimeNs
+	}
+	return baseline
+}
+
+// filerBarrierTsNs is filerBarrierTsNsWith against the current filer, for
+// operations that target it outside a filer client callback (the server-side
+// copy posts to the current filer over HTTP).
+func (wfs *WFS) filerBarrierTsNs() int64 {
+	baseline := wfs.latestKnownFilerTsNs()
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	_ = pb.WithGrpcClient(ctx, false, wfs.signature, func(conn *grpc.ClientConn) error {
+		baseline = wfs.filerBarrierTsNsWith(filer_pb.NewSeaweedFilerClient(conn))
+		return nil
+	}, wfs.getCurrentFiler().ToGrpcAddress(), false, wfs.option.GrpcDialOption)
+	return baseline
+}
+
+// invalidateOpenFileHandle refreshes an open file handle from a metadata
+// 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, eventTsNs int64) {
+	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)
+
+	// Invalidations apply asynchronously, so this event may be old news by
+	// now: a local flush can land while the event sits in the queue, and the
+	// flush's own event is dedup-suppressed, so a rollback would never heal.
+	// Both timestamps come from the filer log, so this orders exactly.
+	if eventTsNs != 0 && eventTsNs <= fh.lastLocalEntryTsNs.Load() {
+		return
+	}
+
+	fh.dirtyPages.Destroy()
+	fh.dirtyPages = newPageWriter(fh, wfs.option.ChunkSizeLimit)
+
+	// Prefer the local store's entry when the parent directory is cached: the
+	// apply loop has already ordered this event and anything newer (e.g. a
+	// local flush) into it. An uncached parent receives no store writes, so a
+	// hit there could be a stale leftover masking this event — fall through
+	// to the event entry instead, the freshest information for that case.
+	dir, _ := filePath.DirAndName()
+	if wfs.metaCache.IsDirectoryCached(util.FullPath(dir)) {
+		if localEntry, findErr := wfs.metaCache.FindEntry(context.Background(), filePath); findErr == nil && localEntry != nil {
+			fh.SetEntry(localEntry.ToProtoEntry())
+			return
+		}
+	}
+	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_file_copy_range.go

@@ -216,6 +216,7 @@ func (wfs *WFS) tryServerSideWholeFileCopy(cancel <-chan struct{}, in *fuse.Copy
 
 	glog.V(1).Infof("CopyFileRange server-side copy %s => %s (%d bytes)", copyRequest.srcPath, copyRequest.dstPath, copyRequest.sourceSize)
 
+	baselineTsNs := wfs.filerBarrierTsNs()
 	entry, outcome, err := performServerSideWholeFileCopy(cancel, wfs, copyRequest)
 	switch outcome {
 	case serverSideWholeFileCopyCommitted:
@@ -224,6 +225,7 @@ func (wfs *WFS) tryServerSideWholeFileCopy(cancel <-chan struct{}, in *fuse.Copy
 		} else {
 			glog.V(1).Infof("CopyFileRange server-side copy %s => %s completed (%d bytes)", copyRequest.srcPath, copyRequest.dstPath, copyRequest.sourceSize)
 		}
+		fhOut.advanceLocalEntryTs(baselineTsNs)
 		wfs.applyServerSideWholeFileCopyResult(fhIn, fhOut, copyRequest.dstPath, entry, copyRequest.sourceSize)
 		return uint32(copyRequest.sourceSize), true, fuse.OK
 	case serverSideWholeFileCopyAmbiguous:

weed/mount/weedfs_file_copy_range_test.go

@@ -346,7 +346,7 @@ func newCopyRangeTestWFSWithMetaCache(t *testing.T) *WFS {
 		func(path util.FullPath) bool {
 			return wfs.inodeToPath.IsChildrenCached(path)
 		},
-		func(util.FullPath, *filer_pb.Entry) {},
+		func(util.FullPath, *filer_pb.Entry, int64) {},
 		nil,
 	)
 	t.Cleanup(func() {

weed/mount/weedfs_file_mkrm_test.go

@@ -128,7 +128,7 @@ func newCreateTestWFS(t *testing.T) (*WFS, *createEntryTestServer) {
 		func(path util.FullPath) bool {
 			return wfs.inodeToPath.IsChildrenCached(path)
 		},
-		func(util.FullPath, *filer_pb.Entry) {},
+		func(util.FullPath, *filer_pb.Entry, int64) {},
 		nil,
 	)
 	wfs.inodeToPath.MarkChildrenCached(root)

weed/mount/weedfs_file_sync.go

@@ -260,11 +260,13 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n
 
 	wfs.mapPbIdFromLocalToFiler(request.Entry)
 
+	baselineTsNs := wfs.latestKnownFilerTsNs()
 	resp, err := wfs.streamCreateEntry(ctx, request)
 	if err != nil {
 		glog.Errorf("fh flush create %s: %v", fileFullPath, err)
 		return fmt.Errorf("fh flush create %s: %v", fileFullPath, err)
 	}
+	fh.noteFilerAck(baselineTsNs, resp.GetMetadataEvent())
 
 	event := resp.GetMetadataEvent()
 	if event == nil {

weed/mount/weedfs_filehandle.go

@@ -17,9 +17,33 @@ func (wfs *WFS) AcquireHandle(inode uint64, flags, uid, gid uint32) (fileHandle
 	// data that was just written asynchronously.
 	wfs.waitForPendingAsyncFlush(inode)
 
+	// Fence baseline for a freshly looked-up entry: the lookup below reads
+	// the local store or the filer, both of which reflect every event applied
+	// so far, so invalidations already queued at or before this cursor are
+	// old news for the new handle. Captured before the lookup — never after —
+	// so an event arriving mid-open cannot inflate the fence past the state
+	// the lookup actually returned.
+	baselineTsNs := wfs.latestKnownFilerTsNs()
+
 	var entry *filer_pb.Entry
 	var path util.FullPath
-	path, _, entry, status = wfs.maybeReadEntry(inode)
+	var existingFh *FileHandle
+	path, existingFh, entry, status = wfs.maybeReadEntry(inode)
+	// Stable-read: an event applied while the lookup was in flight may
+	// already be reflected in the returned entry yet sit above the
+	// pre-lookup cursor, so its queued invalidation would replace fresher
+	// state. Re-read until the cursor is stable across the lookup. Bounded:
+	// on sustained churn the last pre-lookup value stands, which can only
+	// under-fence — a too-low fence lets an event re-apply, never blocks a
+	// newer one.
+	for attempt := 0; attempt < 3 && existingFh == nil && status == fuse.OK; attempt++ {
+		currentTsNs := wfs.latestKnownFilerTsNs()
+		if currentTsNs == baselineTsNs {
+			break
+		}
+		baselineTsNs = currentTsNs
+		path, existingFh, entry, status = wfs.maybeReadEntry(inode)
+	}
 	if status == fuse.OK {
 		if wormEnforced, _ := wfs.wormEnforcedForEntry(path, entry); wormEnforced && flags&fuse.O_ANYWRITE != 0 {
 			return nil, fuse.EPERM
@@ -39,6 +63,11 @@ func (wfs *WFS) AcquireHandle(inode uint64, flags, uid, gid uint32) (fileHandle
 		// need to AcquireFileHandle again to ensure correct handle counter
 		fileHandle = wfs.fhMap.AcquireFileHandle(wfs, inode, entry)
 		fileHandle.RememberPath(path)
+		// An existing handle's entry did not come from the lookup above, so
+		// the fence would overstate what it reflects.
+		if existingFh == nil {
+			fileHandle.advanceLocalEntryTs(baselineTsNs)
+		}
 
 		// Acquire distributed lock for write opens. The lock is held with
 		// auto-renewal until the file handle is released (close).

weed/mount/weedfs_invalidate_open_handle_test.go

@@ -0,0 +1,924 @@
+package mount
+
+import (
+	"context"
+	"net"
+	"path/filepath"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/go-fuse/v2/fuse"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"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)
+	}
+}
+
+// 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)
+	}
+}
+
+// During a directory build, an event touching the building directory is
+// buffered: its store write is deferred to build completion while its
+// invalidation runs immediately, against a store that may not reflect the
+// listing yet. Build completion must re-invalidate every buffered event —
+// including snapshot-covered ones — so the handle lands on the completed
+// directory's state.
+func TestBufferedBuildEventReinvalidatesOnCompletion(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+
+	wfs.inodeToPath.MarkChildrenCached(util.FullPath("/"))
+	wfs.inodeToPath.Lookup(util.FullPath("/dir"), time.Now().Unix(), true, false, 0, false)
+
+	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},
+	})
+
+	if err := wfs.metaCache.BeginDirectoryBuild(context.Background(), util.FullPath("/dir")); err != nil {
+		t.Fatalf("begin build: %v", err)
+	}
+
+	// Covered by the upcoming listing snapshot (TsNs 900 <= snapshot 1000);
+	// its immediate invalidation runs before the listing inserts the newer
+	// entry, so the handle picks up the event's state.
+	event := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      900,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "file"},
+			NewEntry: &filer_pb.Entry{
+				Name:       "file",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 100},
+			},
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), event, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply buffered event: %v", err)
+	}
+	wfs.metaCache.WaitForEntryInvalidations()
+	if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 100 {
+		t.Fatalf("open handle file size mid-build = %d, want 100 (event state)", size)
+	}
+
+	// The listing then inserts the newer entry the snapshot already covers.
+	if err := wfs.metaCache.InsertEntry(context.Background(), &filer.Entry{
+		FullPath: "/dir/file",
+		Attr: filer.Attr{
+			Crtime:   time.Unix(1, 0),
+			Mtime:    time.Unix(1, 0),
+			Mode:     0100644,
+			FileSize: 300,
+		},
+	}); err != nil {
+		t.Fatalf("insert listing entry: %v", err)
+	}
+
+	if err := wfs.metaCache.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 1000); err != nil {
+		t.Fatalf("complete build: %v", err)
+	}
+	wfs.metaCache.WaitForEntryInvalidations()
+	if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 300 {
+		t.Fatalf("open handle file size after build completion = %d, want 300 (snapshot-covered event must re-invalidate)", size)
+	}
+}
+
+// A hit in the local store only resolves an invalidation when the parent
+// directory is cached. An uncached parent receives no store writes, so a
+// leftover entry there is stale and must not mask the event.
+func TestUncachedDirStaleStoreEntryDoesNotMaskEvent(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},
+	})
+
+	// Leftover store entry under a parent that is not children-cached.
+	if err := wfs.metaCache.InsertEntry(context.Background(), &filer.Entry{
+		FullPath: "/dir/file",
+		Attr: filer.Attr{
+			Crtime:   time.Unix(1, 0),
+			Mtime:    time.Unix(1, 0),
+			Mode:     0100644,
+			FileSize: 88,
+		},
+	}); err != nil {
+		t.Fatalf("insert stale entry: %v", err)
+	}
+
+	updateResp := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      1000,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "file"},
+			NewEntry: &filer_pb.Entry{
+				Name:       "file",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 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()
+
+	if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 180020 {
+		t.Fatalf("open handle file size = %d, want 180020 (stale store entry must not mask the event)", size)
+	}
+}
+
+// In a read-through directory neither a local flush nor the event reaches the
+// local store, so ordering falls to the filer log timestamps: an event at or
+// before the handle's last filer-acknowledged local mutation is old news and
+// must not roll the handle back.
+func TestQueuedEventOlderThanFlushedStateIsIgnored(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},
+	})
+
+	// Hold the handle lock so the queued invalidation cannot apply yet.
+	testLock := wfs.fhLockTable.AcquireLock("test", fh.fh, util.ExclusiveLock)
+	older := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      1000,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "file"},
+			NewEntry: &filer_pb.Entry{
+				Name:       "file",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 100},
+			},
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), older, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		wfs.fhLockTable.ReleaseLock(fh.fh, testLock)
+		t.Fatalf("apply subscriber event: %v", err)
+	}
+
+	// A local flush lands: the filer acknowledged it with a later log
+	// timestamp than the queued event.
+	fh.SetEntry(&filer_pb.Entry{
+		Name:       "file",
+		Attributes: &filer_pb.FuseAttributes{FileSize: 200},
+	})
+	fh.advanceLocalEntryTs(2000)
+	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 (event at TsNs 1000 predates the flush at 2000)", size)
+	}
+}
+
+type fakeFilerServer struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	lookupSize    uint64
+	pingTsNs      int64
+	cacheSize     uint64
+	cacheLogTsNs  int64
+	lookupCalls   atomic.Int32
+	lookupStarted chan struct{} // closed when the first lookup arrives
+	lookupGate    chan struct{} // first lookup waits here when non-nil
+}
+
+func (s *fakeFilerServer) CacheRemoteObjectToLocalCluster(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) {
+	// No MetadataEvent: the object was already cached by another client.
+	return &filer_pb.CacheRemoteObjectToLocalClusterResponse{
+		Entry: &filer_pb.Entry{
+			Name:       req.Name,
+			Attributes: &filer_pb.FuseAttributes{FileSize: s.cacheSize, FileMode: 0100644},
+		},
+		LogTsNs: s.cacheLogTsNs,
+	}, nil
+}
+
+func (s *fakeFilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	if s.lookupGate != nil && s.lookupCalls.Add(1) == 1 {
+		close(s.lookupStarted)
+		<-s.lookupGate
+	}
+	return &filer_pb.LookupDirectoryEntryResponse{
+		Entry: &filer_pb.Entry{
+			Name:       req.Name,
+			Attributes: &filer_pb.FuseAttributes{FileSize: s.lookupSize, FileMode: 0100644},
+		},
+	}, nil
+}
+
+func (s *fakeFilerServer) Ping(ctx context.Context, req *filer_pb.PingRequest) (*filer_pb.PingResponse, error) {
+	return &filer_pb.PingResponse{StartTimeNs: s.pingTsNs}, nil
+}
+
+func (s *fakeFilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) {
+	return &filer_pb.UpdateEntryResponse{
+		MetadataEvent: &filer_pb.SubscribeMetadataResponse{
+			Directory: req.Directory,
+			TsNs:      2000,
+			EventNotification: &filer_pb.EventNotification{
+				OldEntry:      &filer_pb.Entry{Name: req.Entry.Name},
+				NewEntry:      req.Entry,
+				NewParentPath: req.Directory,
+			},
+		},
+	}, nil
+}
+
+// saveEntry (truncate, setattr) must advance the open handle's watermark from
+// the acknowledged mutation's log timestamp, or an older queued event rolls
+// the mutation back in a read-through directory.
+func TestSaveEntryKeepsOpenHandleAheadOfOlderEvents(t *testing.T) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	t.Cleanup(func() { _ = listener.Close() })
+	server := pb.NewGrpcServer()
+	filer_pb.RegisterSeaweedFilerServer(server, &fakeFilerServer{})
+	go server.Serve(listener)
+	t.Cleanup(server.Stop)
+
+	wfs := newInvalidateTestWFS(t)
+	wfs.option.FilerAddresses = []pb.ServerAddress{
+		pb.NewServerAddressWithGrpcPort("127.0.0.1:1", listener.Addr().(*net.TCPAddr).Port),
+	}
+
+	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},
+	})
+
+	testLock := wfs.fhLockTable.AcquireLock("test", fh.fh, util.ExclusiveLock)
+	older := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      1000,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "file"},
+			NewEntry: &filer_pb.Entry{
+				Name:       "file",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 100},
+			},
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), older, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		wfs.fhLockTable.ReleaseLock(fh.fh, testLock)
+		t.Fatalf("apply subscriber event: %v", err)
+	}
+
+	// A truncate-style mutation: the filer acknowledges it at TsNs 2000 and
+	// the handle takes the new entry.
+	saved := &filer_pb.Entry{
+		Name:       "file",
+		Attributes: &filer_pb.FuseAttributes{FileSize: 200},
+	}
+	if code := wfs.saveEntry(util.FullPath("/dir/file"), saved); code != fuse.OK {
+		wfs.fhLockTable.ReleaseLock(fh.fh, testLock)
+		t.Fatalf("saveEntry status = %v, want OK", code)
+	}
+	fh.SetEntry(saved)
+	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 (saveEntry at TsNs 2000 outranks the queued event at 1000)", size)
+	}
+}
+
+// When a filer response carries no metadata event, the watermark falls back
+// to the latest filer log timestamp already seen: the filer served state at
+// least that new, so queued events at or before it must not roll the fresh
+// handle state back.
+func TestNilAckEventFallsBackToLatestSeenTs(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},
+	})
+
+	// An unrelated event advances the mount's known filer log position.
+	unrelated := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      1500,
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{
+				Name:       "other",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 1},
+			},
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), unrelated, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply unrelated event: %v", err)
+	}
+	if got := wfs.metaCache.LatestEventTsNs(); got != 1500 {
+		t.Fatalf("LatestEventTsNs = %d, want 1500", got)
+	}
+
+	testLock := wfs.fhLockTable.AcquireLock("test", fh.fh, util.ExclusiveLock)
+	older := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      1000,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "file"},
+			NewEntry: &filer_pb.Entry{
+				Name:       "file",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 100},
+			},
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), older, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		wfs.fhLockTable.ReleaseLock(fh.fh, testLock)
+		t.Fatalf("apply subscriber event: %v", err)
+	}
+
+	// A filer RPC that returned no event (remote cache already populated,
+	// server-side copy) installs fresh state; the baseline captured before
+	// the RPC stands in for the missing event timestamp.
+	baselineTsNs := wfs.metaCache.LatestEventTsNs()
+	fh.SetEntry(&filer_pb.Entry{
+		Name:       "file",
+		Attributes: &filer_pb.FuseAttributes{FileSize: 200},
+	})
+	fh.noteFilerAck(baselineTsNs, nil)
+	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 event at TsNs 1000 predates the known log position 1500)", size)
+	}
+}
+
+// startFakeFiler serves fake on a local port and points wfs at it.
+func startFakeFiler(t *testing.T, wfs *WFS, fake *fakeFilerServer) {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	t.Cleanup(func() { _ = listener.Close() })
+	server := pb.NewGrpcServer()
+	filer_pb.RegisterSeaweedFilerServer(server, fake)
+	go server.Serve(listener)
+	t.Cleanup(server.Stop)
+	wfs.option.FilerAddresses = []pb.ServerAddress{
+		pb.NewServerAddressWithGrpcPort("127.0.0.1:1", listener.Addr().(*net.TCPAddr).Port),
+	}
+}
+
+// A handle opened while an older event sits in the invalidation queue is
+// fenced at open time: its entry came from a lookup that reflects every
+// event applied so far, so the queued event must not replace it.
+func TestQueuedEventDoesNotRollBackHandleOpenedAfterEnqueue(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+	startFakeFiler(t, wfs, &fakeFilerServer{lookupSize: 200})
+
+	// Stall the single invalidation worker on an unrelated handle's lock so
+	// queued events outlive the open below.
+	blockerInode := wfs.inodeToPath.Lookup(util.FullPath("/dir/blocker"), time.Now().Unix(), false, false, 0, false)
+	blockerFh := wfs.fhMap.AcquireFileHandle(wfs, blockerInode, &filer_pb.Entry{
+		Name:       "blocker",
+		Attributes: &filer_pb.FuseAttributes{FileSize: 1},
+	})
+	blockerLock := wfs.fhLockTable.AcquireLock("test", blockerFh.fh, util.ExclusiveLock)
+	blockerEvent := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      500,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "blocker"},
+			NewEntry: &filer_pb.Entry{
+				Name:       "blocker",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 2},
+			},
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), blockerEvent, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock)
+		t.Fatalf("apply blocker event: %v", err)
+	}
+
+	// The event for the file predates the open below.
+	older := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      1000,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "file"},
+			NewEntry: &filer_pb.Entry{
+				Name:       "file",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 100},
+			},
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), older, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock)
+		t.Fatalf("apply subscriber event: %v", err)
+	}
+
+	// Open the file now: the lookup reaches the filer, which already serves
+	// the newer size-200 state.
+	inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false)
+	fh, status := wfs.AcquireHandle(inode, 0, 0, 0)
+	if status != fuse.OK {
+		wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock)
+		t.Fatalf("AcquireHandle status = %v, want OK", status)
+	}
+	if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 200 {
+		wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock)
+		t.Fatalf("opened handle file size = %d, want 200", size)
+	}
+
+	wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock)
+	wfs.metaCache.WaitForEntryInvalidations()
+
+	if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 200 {
+		t.Fatalf("open handle file size = %d, want 200 (event queued before the open must not roll it back)", size)
+	}
+}
+
+// The delivered-event cursor misses events already committed on the filer but
+// not yet delivered to the subscription. A pre-RPC filer self-ping reads the
+// clock those events are stamped with, so state fetched after the ping fences
+// them out.
+func TestFilerBarrierCoversUndeliveredEvents(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+	startFakeFiler(t, wfs, &fakeFilerServer{pingTsNs: 2000})
+
+	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},
+	})
+
+	// The copy/remote-cache pattern: barrier, then the operation's result.
+	// The event at TsNs 1500 is committed but not yet delivered, so only the
+	// filer clock (2000) can cover it.
+	baselineTsNs := wfs.filerBarrierTsNs()
+	if baselineTsNs != 2000 {
+		t.Fatalf("filerBarrierTsNs = %d, want 2000 from the filer ping", baselineTsNs)
+	}
+	fh.SetEntry(&filer_pb.Entry{
+		Name:       "file",
+		Attributes: &filer_pb.FuseAttributes{FileSize: 200},
+	})
+	fh.advanceLocalEntryTs(baselineTsNs)
+
+	late := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      1500,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "file"},
+			NewEntry: &filer_pb.Entry{
+				Name:       "file",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 100},
+			},
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), late, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply late event: %v", err)
+	}
+	wfs.metaCache.WaitForEntryInvalidations()
+
+	if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 200 {
+		t.Fatalf("open handle file size = %d, want 200 (undelivered-at-barrier event must not roll back)", size)
+	}
+}
+
+// An event buffered for a building directory must be covered by the open-time
+// cursor even if it never reaches a store write: aborting the build drops the
+// buffered events while their invalidations stay queued, so an open fenced
+// below the event would be rolled back.
+func TestAbortedBuildEventStillCoveredByOpenFence(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+	startFakeFiler(t, wfs, &fakeFilerServer{lookupSize: 200})
+
+	// Stall the invalidation worker on an unrelated handle's lock.
+	blockerInode := wfs.inodeToPath.Lookup(util.FullPath("/other/blocker"), time.Now().Unix(), false, false, 0, false)
+	blockerFh := wfs.fhMap.AcquireFileHandle(wfs, blockerInode, &filer_pb.Entry{
+		Name:       "blocker",
+		Attributes: &filer_pb.FuseAttributes{FileSize: 1},
+	})
+	blockerLock := wfs.fhLockTable.AcquireLock("test", blockerFh.fh, util.ExclusiveLock)
+	blockerEvent := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/other",
+		TsNs:      500,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "blocker"},
+			NewEntry: &filer_pb.Entry{
+				Name:       "blocker",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 2},
+			},
+			NewParentPath: "/other",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), blockerEvent, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock)
+		t.Fatalf("apply blocker event: %v", err)
+	}
+
+	if err := wfs.metaCache.BeginDirectoryBuild(context.Background(), util.FullPath("/dir")); err != nil {
+		wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock)
+		t.Fatalf("begin build: %v", err)
+	}
+	buffered := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      1000,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "file"},
+			NewEntry: &filer_pb.Entry{
+				Name:       "file",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 100},
+			},
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), buffered, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock)
+		t.Fatalf("apply buffered event: %v", err)
+	}
+	if got := wfs.metaCache.LatestEventTsNs(); got != 1000 {
+		wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock)
+		t.Fatalf("LatestEventTsNs = %d, want 1000 (buffered event must advance the cursor)", got)
+	}
+	if err := wfs.metaCache.AbortDirectoryBuild(context.Background(), util.FullPath("/dir")); err != nil {
+		wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock)
+		t.Fatalf("abort build: %v", err)
+	}
+
+	// Open after the abort: the lookup reaches the filer, which serves the
+	// newer size-200 state; the fence must cover the still-queued event.
+	inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false)
+	fh, status := wfs.AcquireHandle(inode, 0, 0, 0)
+	if status != fuse.OK {
+		wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock)
+		t.Fatalf("AcquireHandle status = %v, want OK", status)
+	}
+
+	wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock)
+	wfs.metaCache.WaitForEntryInvalidations()
+
+	if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 200 {
+		t.Fatalf("open handle file size = %d, want 200 (aborted-build event must not roll back the open)", size)
+	}
+}
+
+// During filer failover, WithFilerClient retries the callback against another
+// filer while the current-filer index still points at the failed one. The
+// barrier must ping through the callback's client, or it silently degrades to
+// the delivered-event cursor and reopens the undelivered-event rollback.
+func TestRemoteCacheBarrierFollowsFailover(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+	fake := &fakeFilerServer{pingTsNs: 2000, cacheSize: 200}
+	startFakeFiler(t, wfs, fake)
+	// First filer is unreachable; WithFilerClient fails over to the fake.
+	live := wfs.option.FilerAddresses[0]
+	wfs.option.FilerAddresses = []pb.ServerAddress{
+		pb.NewServerAddressWithGrpcPort("127.0.0.1:1", 1),
+		live,
+	}
+
+	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},
+	})
+
+	if err := fh.downloadRemoteEntry(fh.GetEntry()); err != nil {
+		t.Fatalf("downloadRemoteEntry: %v", err)
+	}
+	if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 200 {
+		t.Fatalf("downloaded file size = %d, want 200", size)
+	}
+
+	// An event committed before the download (TsNs 1500 < ping 2000) but
+	// delivered only now must not roll the handle back.
+	late := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      1500,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "file"},
+			NewEntry: &filer_pb.Entry{
+				Name:       "file",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 100},
+			},
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), late, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply late event: %v", err)
+	}
+	wfs.metaCache.WaitForEntryInvalidations()
+
+	if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 200 {
+		t.Fatalf("open handle file size = %d, want 200 (barrier must come from the failover filer)", size)
+	}
+}
+
+// An event applied while the open's lookup is in flight can already be
+// reflected in the returned entry while sitting above the pre-lookup cursor.
+// The open re-reads until the cursor is stable across the lookup so the fence
+// covers such events.
+func TestEventDuringOpenLookupIsFenced(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+	fake := &fakeFilerServer{
+		lookupSize:    200,
+		lookupStarted: make(chan struct{}),
+		lookupGate:    make(chan struct{}),
+	}
+	startFakeFiler(t, wfs, fake)
+
+	// Stall the invalidation worker on an unrelated handle's lock.
+	blockerInode := wfs.inodeToPath.Lookup(util.FullPath("/other/blocker"), time.Now().Unix(), false, false, 0, false)
+	blockerFh := wfs.fhMap.AcquireFileHandle(wfs, blockerInode, &filer_pb.Entry{
+		Name:       "blocker",
+		Attributes: &filer_pb.FuseAttributes{FileSize: 1},
+	})
+	blockerLock := wfs.fhLockTable.AcquireLock("test", blockerFh.fh, util.ExclusiveLock)
+	blockerReleased := false
+	releaseBlocker := func() {
+		if !blockerReleased {
+			blockerReleased = true
+			wfs.fhLockTable.ReleaseLock(blockerFh.fh, blockerLock)
+		}
+	}
+	defer releaseBlocker()
+	blockerEvent := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/other",
+		TsNs:      500,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "blocker"},
+			NewEntry: &filer_pb.Entry{
+				Name:       "blocker",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 2},
+			},
+			NewParentPath: "/other",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), blockerEvent, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply blocker event: %v", err)
+	}
+
+	inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false)
+	type openResult struct {
+		fh     *FileHandle
+		status fuse.Status
+	}
+	opened := make(chan openResult, 1)
+	go func() {
+		fh, status := wfs.AcquireHandle(inode, 0, 0, 0)
+		opened <- openResult{fh, status}
+	}()
+
+	// While the open's lookup is blocked in the filer, an event lands and its
+	// invalidation is queued behind the blocker.
+	<-fake.lookupStarted
+	during := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      1500,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "file"},
+			NewEntry: &filer_pb.Entry{
+				Name:       "file",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 100},
+			},
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), during, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply mid-lookup event: %v", err)
+	}
+	close(fake.lookupGate)
+
+	result := <-opened
+	if result.status != fuse.OK {
+		t.Fatalf("AcquireHandle status = %v, want OK", result.status)
+	}
+
+	releaseBlocker()
+	wfs.metaCache.WaitForEntryInvalidations()
+
+	if size := result.fh.GetEntry().GetEntry().Attributes.FileSize; size != 200 {
+		t.Fatalf("open handle file size = %d, want 200 (mid-lookup event must be fenced)", size)
+	}
+}
+
+// A pre-RPC ping cannot cover an event committed during the RPC itself. The
+// cache response carries a log timestamp stamped before the filer read the
+// entry, causally fencing everything the returned entry reflects.
+func TestCacheResponseLogTsFencesEventsCommittedDuringRPC(t *testing.T) {
+	wfs := newInvalidateTestWFS(t)
+	// The ping (100) predates the event (1500); only the response's log
+	// timestamp (2000) can cover it.
+	startFakeFiler(t, wfs, &fakeFilerServer{pingTsNs: 100, cacheSize: 200, cacheLogTsNs: 2000})
+
+	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},
+	})
+
+	if err := fh.downloadRemoteEntry(fh.GetEntry()); err != nil {
+		t.Fatalf("downloadRemoteEntry: %v", err)
+	}
+
+	late := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		TsNs:      1500,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "file"},
+			NewEntry: &filer_pb.Entry{
+				Name:       "file",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 100},
+			},
+			NewParentPath: "/dir",
+		},
+	}
+	if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), late, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply late event: %v", err)
+	}
+	wfs.metaCache.WaitForEntryInvalidations()
+
+	if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 200 {
+		t.Fatalf("open handle file size = %d, want 200 (response log ts must fence the mid-RPC event)", size)
+	}
+}

weed/mount/weedfs_metadata_flush.go

@@ -171,10 +171,12 @@ func (wfs *WFS) flushFileMetadata(fh *FileHandle) error {
 
 	wfs.mapPbIdFromLocalToFiler(request.Entry)
 
+	baselineTsNs := wfs.latestKnownFilerTsNs()
 	resp, err := wfs.streamCreateEntry(context.Background(), request)
 	if err != nil {
 		return err
 	}
+	fh.noteFilerAck(baselineTsNs, resp.GetMetadataEvent())
 
 	event := resp.GetMetadataEvent()
 	if event == nil {

weed/mount/weedfs_rename_test.go

@@ -30,7 +30,7 @@ func TestHandleRenameResponseLeavesUncachedTargetOutOfCache(t *testing.T) {
 		func(path util.FullPath) bool {
 			return inodeToPath.IsChildrenCached(path)
 		},
-		func(util.FullPath, *filer_pb.Entry) {},
+		func(util.FullPath, *filer_pb.Entry, int64) {},
 		nil,
 	)
 	defer mc.Shutdown()

weed/mount/wfs_save.go

@@ -27,6 +27,7 @@ func (wfs *WFS) saveEntry(path util.FullPath, entry *filer_pb.Entry) (code fuse.
 
 	glog.V(1).Infof("save entry: %v", request)
 
+	baselineTsNs := wfs.latestKnownFilerTsNs()
 	var resp *filer_pb.UpdateEntryResponse
 	err := retryMetadataFlushIf(context.Background(), func() error {
 		var callErr error
@@ -51,6 +52,14 @@ func (wfs *WFS) saveEntry(path util.FullPath, entry *filer_pb.Entry) (code fuse.
 		return fuseStatus
 	}
 
+	// The mutation is acknowledged; keep any open handle for this path ahead
+	// of subscription events the filer logged before it.
+	if inode, found := wfs.inodeToPath.GetInode(path); found {
+		if fh, fhFound := wfs.fhMap.FindFileHandle(inode); fhFound {
+			fh.noteFilerAck(baselineTsNs, resp.GetMetadataEvent())
+		}
+	}
+
 	event := resp.GetMetadataEvent()
 	if event == nil {
 		event = metadataUpdateEvent(parentDir, entry)

weed/pb/filer.proto

@@ -763,6 +763,9 @@ message CacheRemoteObjectToLocalClusterRequest {
 message CacheRemoteObjectToLocalClusterResponse {
     Entry entry = 1;
     SubscribeMetadataResponse metadata_event = 2;
+    // filer log position stamped before the entry read: every event at or
+    // below it is reflected in the returned entry
+    int64 log_ts_ns = 3;
 }
 
 /////////////////////////

weed/pb/filer_pb/filer.pb.go

@@ -5262,6 +5262,9 @@ type CacheRemoteObjectToLocalClusterResponse struct {
 	state         protoimpl.MessageState     `protogen:"open.v1"`
 	Entry         *Entry                     `protobuf:"bytes,1,opt,name=entry,proto3" json:"entry,omitempty"`
 	MetadataEvent *SubscribeMetadataResponse `protobuf:"bytes,2,opt,name=metadata_event,json=metadataEvent,proto3" json:"metadata_event,omitempty"`
+	// filer log position stamped before the entry read: every event at or
+	// below it is reflected in the returned entry
+	LogTsNs       int64 `protobuf:"varint,3,opt,name=log_ts_ns,json=logTsNs,proto3" json:"log_ts_ns,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -5310,6 +5313,13 @@ func (x *CacheRemoteObjectToLocalClusterResponse) GetMetadataEvent() *SubscribeM
 	return nil
 }
 
+func (x *CacheRemoteObjectToLocalClusterResponse) GetLogTsNs() int64 {
+	if x != nil {
+		return x.LogTsNs
+	}
+	return 0
+}
+
 // ///////////////////////
 // distributed lock management
 // ///////////////////////
@@ -7333,10 +7343,11 @@ const file_filer_proto_rawDesc = "" +
 	"\tdirectory\x18\x01 \x01(\tR\tdirectory\x12\x12\n" +
 	"\x04name\x18\x02 \x01(\tR\x04name\x12+\n" +
 	"\x11chunk_concurrency\x18\x03 \x01(\x05R\x10chunkConcurrency\x121\n" +
-	"\x14download_concurrency\x18\x04 \x01(\x05R\x13downloadConcurrency\"\x9c\x01\n" +
+	"\x14download_concurrency\x18\x04 \x01(\x05R\x13downloadConcurrency\"\xb8\x01\n" +
 	"'CacheRemoteObjectToLocalClusterResponse\x12%\n" +
 	"\x05entry\x18\x01 \x01(\v2\x0f.filer_pb.EntryR\x05entry\x12J\n" +
-	"\x0emetadata_event\x18\x02 \x01(\v2#.filer_pb.SubscribeMetadataResponseR\rmetadataEvent\"\x9b\x01\n" +
+	"\x0emetadata_event\x18\x02 \x01(\v2#.filer_pb.SubscribeMetadataResponseR\rmetadataEvent\x12\x1a\n" +
+	"\tlog_ts_ns\x18\x03 \x01(\x03R\alogTsNs\"\x9b\x01\n" +
 	"\vLockRequest\x12\x12\n" +
 	"\x04name\x18\x01 \x01(\tR\x04name\x12&\n" +
 	"\x0fseconds_to_lock\x18\x02 \x01(\x03R\rsecondsToLock\x12\x1f\n" +

weed/server/filer_grpc_server_remote.go

@@ -64,6 +64,11 @@ func (fs *FilerServer) CacheRemoteObjectToLocalCluster(ctx context.Context, req
 // doCacheRemoteObjectToLocalCluster performs the actual caching operation.
 // This is called from singleflight, so only one instance runs per object.
 func (fs *FilerServer) doCacheRemoteObjectToLocalCluster(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) {
+	// Log position fence, stamped before the entry read: metadata events are
+	// logged after their store write and carry this clock, so every event at
+	// or below this timestamp is reflected in the entry returned below.
+	logTsNs := time.Now().UnixNano()
+
 	// find the entry first to check if already cached
 	entry, err := fs.filer.FindEntry(ctx, util.JoinPath(req.Directory, req.Name))
 	if err == filer_pb.ErrNotFound {
@@ -73,7 +78,7 @@ func (fs *FilerServer) doCacheRemoteObjectToLocalCluster(ctx context.Context, re
 		return nil, fmt.Errorf("find entry %s/%s: %v", req.Directory, req.Name, err)
 	}
 
-	resp := &filer_pb.CacheRemoteObjectToLocalClusterResponse{}
+	resp := &filer_pb.CacheRemoteObjectToLocalClusterResponse{LogTsNs: logTsNs}
 
 	// Early return if not a remote-only object or already cached
 	if entry.Remote == nil || entry.Remote.RemoteSize == 0 {