From edacb39b819e54e00d131952896b8c7a167a7a7a Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 22 Jul 2026 14:09:39 -0700 Subject: [PATCH] mount: re-invalidate open handles when replaying buffered build events MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An event touching a directory mid-build is buffered — its store write waits for build completion — while its invalidation runs immediately. That refresh resolves against the store, which can still hold the older listing snapshot, pinning the handle to pre-event state with no later refresh. Replay buffered events with entry invalidation enabled: by then the store is ordered, so the re-invalidation lands the final state. Events the snapshot filter skips need no re-invalidation since the listing state is at least as new. Also correct the resolution comment: the store misses for TTL-expired entries too, and on a read-through miss the event-entry fallback can still roll back a racing local flush, since neither write reaches the store there. --- weed/mount/meta_cache/meta_cache.go | 7 +- weed/mount/weedfs.go | 9 ++- .../weedfs_invalidate_open_handle_test.go | 65 +++++++++++++++++++ 3 files changed, 77 insertions(+), 4 deletions(-) diff --git a/weed/mount/meta_cache/meta_cache.go b/weed/mount/meta_cache/meta_cache.go index 731d7634f..afe9b115b 100644 --- a/weed/mount/meta_cache/meta_cache.go +++ b/weed/mount/meta_cache/meta_cache.go @@ -749,7 +749,12 @@ func (mc *MetaCache) completeDirectoryBuildNow(ctx context.Context, dirPath util if snapshotTsNs != 0 && event.TsNs != 0 && event.TsNs <= snapshotTsNs { continue } - if err := mc.applyMetadataResponseDirect(ctx, event, MetadataResponseApplyOptions{}, true); err != nil { + // Re-invalidate on replay: the invalidation enqueued when this event + // arrived resolved against the store mid-build, when the store could + // still hold the older listing snapshot (the event's own store write + // was deferred to this replay). Skipped events need no re-invalidation + // since the listing state is at least as new. + if err := mc.applyMetadataResponseDirect(ctx, event, MetadataResponseApplyOptions{InvalidateEntries: true}, true); err != nil { return err } } diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index f665ea5d0..82cbdae3b 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -701,9 +701,12 @@ func (wfs *WFS) invalidateOpenFileHandle(filePath util.FullPath, entry *filer_pb // snapshot by now: a local flush can install newer state while the event // sits in the queue, and the flush's own event is dedup-suppressed, so a // rollback would never heal. The apply loop has already ordered this event - // and any later state into the local store, so prefer the store's entry; - // it misses only for read-through directories, where the event entry is - // the freshest ordered information available. + // and any later state into the local store, so prefer the store's entry. + // The store misses for read-through directories and TTL-expired entries; + // falling back to the event entry there can still roll back a racing + // local flush (neither write reaches the store in a read-through + // directory), but it is the best ordered information available without a + // filer round-trip. if localEntry, findErr := wfs.metaCache.FindEntry(context.Background(), filePath); findErr == nil && localEntry != nil { fh.SetEntry(localEntry.ToProtoEntry()) return diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go index d1d871908..32ad95af2 100644 --- a/weed/mount/weedfs_invalidate_open_handle_test.go +++ b/weed/mount/weedfs_invalidate_open_handle_test.go @@ -6,6 +6,7 @@ import ( "testing" "time" + "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" @@ -176,3 +177,67 @@ func TestQueuedEventDoesNotRollBackNewerLocalState(t *testing.T) { 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, so that refresh can resolve to the older +// listing snapshot. The completion replay must re-invalidate so the handle +// lands on the event'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) + } + // The in-progress listing inserts the pre-event snapshot. + 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: 100, + }, + }); err != nil { + t.Fatalf("insert listing entry: %v", err) + } + + // Postdates the listing snapshot, so it is buffered; its immediate + // invalidation resolves to the older listing entry. + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + TsNs: 2000, + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "file"}, + NewEntry: &filer_pb.Entry{ + Name: "file", + Attributes: &filer_pb.FuseAttributes{FileSize: 300}, + }, + 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 (listing snapshot)", size) + } + + 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 (buffered event must re-invalidate)", size) + } +}