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_read.go

@@ -190,7 +190,9 @@ 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.
+		// 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 {
@@ -198,6 +200,9 @@ func (fh *FileHandle) downloadRemoteEntry(entry *LockedEntry) error {
 		}
 
 		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.

weed/mount/weedfs_filehandle.go

@@ -29,6 +29,21 @@ func (wfs *WFS) AcquireHandle(inode uint64, flags, uid, gid uint32) (fileHandle
 	var path util.FullPath
 	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

weed/mount/weedfs_invalidate_open_handle_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"net"
 	"path/filepath"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -345,9 +346,13 @@ func TestQueuedEventOlderThanFlushedStateIsIgnored(t *testing.T) {
 
 type fakeFilerServer struct {
 	filer_pb.UnimplementedSeaweedFilerServer
-	lookupSize uint64
-	pingTsNs   int64
-	cacheSize  uint64
+	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) {
@@ -357,10 +362,15 @@ func (s *fakeFilerServer) CacheRemoteObjectToLocalCluster(ctx context.Context, r
 			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,
@@ -783,3 +793,132 @@ func TestRemoteCacheBarrierFollowsFailover(t *testing.T) {
 		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/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 {