From 239cfe64d3f759b1bfd5791a30788060fbdabcaa Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 22 Jul 2026 15:43:19 -0700 Subject: [PATCH] mount: fence new handles and cover undelivered events in the watermark MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A handle opened while an older event sat in the invalidation queue started with a zero watermark, so the delayed worker rolled its freshly looked-up entry back. The open now fences the handle with the event cursor captured before the lookup: the entry it installs reflects every event applied by then, so anything queued at or before that position is old news. Captured before, never after, so an event arriving mid-open cannot inflate the fence past the state the lookup returned — and the cursor now advances only after the store write, so the pairing is sound. Handles reusing an existing entry are not fenced, since that entry did not come from the lookup. The delivered-event cursor also misses events committed on the filer but not yet delivered to the subscription, which let a late-arriving older event roll back a server-side copy or an already-cached remote download. Those paths now take a barrier from a filer self-ping issued before the operation: events are stamped with that same clock, so the returned state is at least as new as anything at or below it. The ping is a single bounded attempt with the cursor as fallback. --- weed/mount/filehandle_read.go | 2 +- weed/mount/meta_cache/meta_cache.go | 5 +- weed/mount/weedfs.go | 21 +++ weed/mount/weedfs_file_copy_range.go | 2 +- weed/mount/weedfs_filehandle.go | 16 +- .../weedfs_invalidate_open_handle_test.go | 158 +++++++++++++++++- 6 files changed, 197 insertions(+), 7 deletions(-) diff --git a/weed/mount/filehandle_read.go b/weed/mount/filehandle_read.go index 1d40d0ef3..b173f39ff 100644 --- a/weed/mount/filehandle_read.go +++ b/weed/mount/filehandle_read.go @@ -188,7 +188,7 @@ func (fh *FileHandle) downloadRemoteEntry(entry *LockedEntry) error { } glog.V(4).Infof("download entry: %v", request) - baselineTsNs := fh.wfs.latestKnownFilerTsNs() + baselineTsNs := fh.wfs.filerBarrierTsNs() resp, err := client.CacheRemoteObjectToLocalCluster(context.Background(), request) if err != nil { return fmt.Errorf("CacheRemoteObjectToLocalCluster file %s: %v", fileFullPath, err) diff --git a/weed/mount/meta_cache/meta_cache.go b/weed/mount/meta_cache/meta_cache.go index 16a91505a..84473f0be 100644 --- a/weed/mount/meta_cache/meta_cache.go +++ b/weed/mount/meta_cache/meta_cache.go @@ -576,7 +576,6 @@ type metadataResponseSideEffects struct { } func (mc *MetaCache) applyMetadataResponseNow(ctx context.Context, resp *filer_pb.SubscribeMetadataResponse, options MetadataResponseApplyOptions) error { - mc.advanceLatestEventTs(resp.TsNs) if mc.shouldSkipDuplicateEvent(resp) { return nil } @@ -610,6 +609,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 } diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index a05e845c2..d53e90e0c 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -686,6 +686,27 @@ func (wfs *WFS) latestKnownFilerTsNs() int64 { return wfs.metaCache.LatestEventTsNs() } +// filerBarrierTsNs returns a timestamp at or below the filer's current log +// position. 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 — no WithFilerClient +// retry waves for an optimization. +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 { + resp, err := filer_pb.NewSeaweedFilerClient(conn).Ping(ctx, &filer_pb.PingRequest{}) + if err == nil && resp.StartTimeNs > baseline { + baseline = resp.StartTimeNs + } + return err + }, 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 diff --git a/weed/mount/weedfs_file_copy_range.go b/weed/mount/weedfs_file_copy_range.go index 1656d0e4f..fb2b29103 100644 --- a/weed/mount/weedfs_file_copy_range.go +++ b/weed/mount/weedfs_file_copy_range.go @@ -216,7 +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.latestKnownFilerTsNs() + baselineTsNs := wfs.filerBarrierTsNs() entry, outcome, err := performServerSideWholeFileCopy(cancel, wfs, copyRequest) switch outcome { case serverSideWholeFileCopyCommitted: diff --git a/weed/mount/weedfs_filehandle.go b/weed/mount/weedfs_filehandle.go index 7fe61d1db..d6a551f02 100644 --- a/weed/mount/weedfs_filehandle.go +++ b/weed/mount/weedfs_filehandle.go @@ -17,9 +17,18 @@ 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) if status == fuse.OK { if wormEnforced, _ := wfs.wormEnforcedForEntry(path, entry); wormEnforced && flags&fuse.O_ANYWRITE != 0 { return nil, fuse.EPERM @@ -39,6 +48,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). diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go index 30199f54c..04b0d82cf 100644 --- a/weed/mount/weedfs_invalidate_open_handle_test.go +++ b/weed/mount/weedfs_invalidate_open_handle_test.go @@ -343,11 +343,26 @@ func TestQueuedEventOlderThanFlushedStateIsIgnored(t *testing.T) { } } -type saveEntryTestServer struct { +type fakeFilerServer struct { filer_pb.UnimplementedSeaweedFilerServer + lookupSize uint64 + pingTsNs int64 } -func (s *saveEntryTestServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { +func (s *fakeFilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + 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, @@ -371,7 +386,7 @@ func TestSaveEntryKeepsOpenHandleAheadOfOlderEvents(t *testing.T) { } t.Cleanup(func() { _ = listener.Close() }) server := pb.NewGrpcServer() - filer_pb.RegisterSeaweedFilerServer(server, &saveEntryTestServer{}) + filer_pb.RegisterSeaweedFilerServer(server, &fakeFilerServer{}) go server.Serve(listener) t.Cleanup(server.Stop) @@ -490,3 +505,140 @@ func TestNilAckEventFallsBackToLatestSeenTs(t *testing.T) { 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) + } +}