diff --git a/weed/mount/meta_cache/meta_cache.go b/weed/mount/meta_cache/meta_cache.go index be586343b..731d7634f 100644 --- a/weed/mount/meta_cache/meta_cache.go +++ b/weed/mount/meta_cache/meta_cache.go @@ -564,7 +564,7 @@ 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) } type metadataResponseSideEffects struct { @@ -975,15 +975,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}) newKey := util.NewFullPath(newDir, message.NewEntry.Name) invalidations = append(invalidations, metadataInvalidation{path: newKey, entry: message.NewEntry}) + } else { + invalidations = append(invalidations, metadataInvalidation{path: oldKey, entry: message.NewEntry}) } return invalidations } @@ -999,7 +1001,7 @@ func collectEntryInvalidations(resp *filer_pb.SubscribeMetadataResponse) []metad 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}) } return invalidations diff --git a/weed/mount/meta_cache/meta_cache_apply_test.go b/weed/mount/meta_cache/meta_cache_apply_test.go index aa97675f9..2ba85cbfa 100644 --- a/weed/mount/meta_cache/meta_cache_apply_test.go +++ b/weed/mount/meta_cache/meta_cache_apply_test.go @@ -409,6 +409,65 @@ 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", + 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 { + t.Fatalf("in-place update invalidations = %+v, want [{/dir/file.txt NewEntry}]", 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() diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index fd2ef7cc9..9c3395d0a 100644 --- a/weed/mount/weedfs.go +++ b/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" @@ -302,26 +303,7 @@ func NewSeaweedFileSystem(option *Option) *WFS { }, 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) - } - } - } - } + wfs.invalidateOpenFileHandle(filePath, entry) }, func(dirPath util.FullPath) { if wfs.inodeToPath.RecordDirectoryUpdate(dirPath, time.Now(), wfs.dirHotWindow, wfs.dirHotThreshold) { wfs.markDirectoryReadThrough(dirPath) @@ -694,6 +676,41 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, fuse.Status) return filer.FromPbEntry(dir, entry), fuse.OK } +// invalidateOpenFileHandle refreshes an open file handle from a metadata +// subscription event. The event entry is applied directly: a second lookup +// here can fail transiently or serve stale cached metadata, and since the +// subscription cursor has already advanced past the event, the handle would +// stay pinned to its old entry until an unrelated event arrives. A nil entry +// means the path no longer holds one (delete, rename away); the handle keeps +// its last entry so unlinked-but-open reads still work. +func (wfs *WFS) invalidateOpenFileHandle(filePath util.FullPath, entry *filer_pb.Entry) { + 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) + + fh.dirtyPages.Destroy() + fh.dirtyPages = newPageWriter(fh, wfs.option.ChunkSizeLimit) + + 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) { diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go new file mode 100644 index 000000000..8381c2f53 --- /dev/null +++ b/weed/mount/weedfs_invalidate_open_handle_test.go @@ -0,0 +1,121 @@ +package mount + +import ( + "context" + "path/filepath" + "testing" + "time" + + "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) + } +}