From 58cf7e9a899aca53d2214f9f3c295fc69b5e53fc Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 6 Mar 2026 00:45:23 -0800 Subject: [PATCH] mount: add refresh event buffering to MetaCache Add infrastructure to buffer subscription events during directory cache refresh. When a directory is being refreshed from the filer (BeginRefresh), events from the subscription handler, local deletes, and local creates are buffered. When the refresh completes (CommitRefresh), the filer snapshot atomically replaces the cached entries, then buffered events are replayed to ensure no mutations are lost. This fixes a race condition where concurrent deletes or creates could be overwritten by a stale filer snapshot during directory refresh, causing ghost entries or lost files. Fixes #8442 --- weed/mount/meta_cache/meta_cache.go | 106 +++++++++++++++++++++++++++- 1 file changed, 104 insertions(+), 2 deletions(-) diff --git a/weed/mount/meta_cache/meta_cache.go b/weed/mount/meta_cache/meta_cache.go index e08ba5c2d..21bc9a591 100644 --- a/weed/mount/meta_cache/meta_cache.go +++ b/weed/mount/meta_cache/meta_cache.go @@ -18,6 +18,20 @@ import ( // need to have logic similar to FilerStoreWrapper // e.g. fill fileId field for chunks +// bufferedEvent represents a subscription event captured during a directory refresh. +type bufferedEvent struct { + oldPath util.FullPath + newEntry *filer.Entry +} + +// refreshState tracks events that arrive while a directory is being refreshed +// from the filer. After the refresh snapshot is applied, buffered events are +// replayed so that creates, deletes, and updates that raced with the snapshot +// are not lost. +type refreshState struct { + events []bufferedEvent +} + type MetaCache struct { root util.FullPath localStore filer.VirtualFilerStore @@ -29,6 +43,7 @@ type MetaCache struct { invalidateFunc func(fullpath util.FullPath, entry *filer_pb.Entry) onDirectoryUpdate func(dir util.FullPath) visitGroup singleflight.Group // deduplicates concurrent EnsureVisited calls for the same path + refreshing map[util.FullPath]*refreshState } func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPath, @@ -45,6 +60,7 @@ func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPat invalidateFunc: func(fullpath util.FullPath, entry *filer_pb.Entry) { invalidateFunc(fullpath, entry) }, + refreshing: make(map[util.FullPath]*refreshState), } } @@ -69,6 +85,11 @@ func openMetaStore(dbFolder string) (*leveldb.LevelDBStore, filer.VirtualFilerSt func (mc *MetaCache) InsertEntry(ctx context.Context, entry *filer.Entry) error { mc.Lock() defer mc.Unlock() + // Buffer the insert if the parent directory is being refreshed + dir, _ := entry.DirAndName() + if state := mc.isRefreshingDir(util.FullPath(dir)); state != nil { + state.events = append(state.events, bufferedEvent{newEntry: entry}) + } return mc.doInsertEntry(ctx, entry) } @@ -86,6 +107,29 @@ func (mc *MetaCache) AtomicUpdateEntryFromFiler(ctx context.Context, oldPath uti mc.Lock() defer mc.Unlock() + // If the affected directory is being refreshed, buffer the event + // instead of applying it. It will be replayed after the snapshot commits. + if oldPath != "" { + dir, _ := oldPath.DirAndName() + if state := mc.isRefreshingDir(util.FullPath(dir)); state != nil { + state.events = append(state.events, bufferedEvent{oldPath, newEntry}) + return nil + } + } + if newEntry != nil { + newDir, _ := newEntry.DirAndName() + if state := mc.isRefreshingDir(util.FullPath(newDir)); state != nil { + state.events = append(state.events, bufferedEvent{oldPath, newEntry}) + return nil + } + } + + return mc.doAtomicUpdateEntryFromFiler(ctx, oldPath, newEntry) +} + +// doAtomicUpdateEntryFromFiler is the core logic for applying a filer event +// to the local cache. Caller must hold mc.Lock(). +func (mc *MetaCache) doAtomicUpdateEntryFromFiler(ctx context.Context, oldPath util.FullPath, newEntry *filer.Entry) error { entry, err := mc.localStore.FindEntry(ctx, oldPath) if err != nil && err != filer_pb.ErrNotFound { glog.Errorf("Metacache: find entry error: %v", err) @@ -104,8 +148,6 @@ func (mc *MetaCache) AtomicUpdateEntryFromFiler(ctx context.Context, oldPath uti } } } - } else { - // println("unknown old directory:", oldDir) } if newEntry != nil { @@ -143,6 +185,14 @@ func (mc *MetaCache) FindEntry(ctx context.Context, fp util.FullPath) (entry *fi func (mc *MetaCache) DeleteEntry(ctx context.Context, fp util.FullPath) (err error) { mc.Lock() defer mc.Unlock() + // Buffer the delete if the parent directory is being refreshed + dir, _ := fp.DirAndName() + if state := mc.isRefreshingDir(util.FullPath(dir)); state != nil { + state.events = append(state.events, bufferedEvent{oldPath: fp}) + } + // Always apply the delete directly as well, so the entry is removed + // immediately for the current node's view. CommitRefresh's snapshot + // may re-insert it, but the buffered event replay will re-delete it. return mc.localStore.DeleteEntry(ctx, fp) } func (mc *MetaCache) DeleteFolderChildren(ctx context.Context, fp util.FullPath) (err error) { @@ -151,6 +201,58 @@ func (mc *MetaCache) DeleteFolderChildren(ctx context.Context, fp util.FullPath) return mc.localStore.DeleteFolderChildren(ctx, fp) } +// BeginRefresh starts buffering subscription events for dirPath. +// While a refresh is active, AtomicUpdateEntryFromFiler will buffer events +// instead of applying them, so they can be replayed after the snapshot is committed. +func (mc *MetaCache) BeginRefresh(dirPath util.FullPath) { + mc.Lock() + defer mc.Unlock() + mc.refreshing[dirPath] = &refreshState{} +} + +// CommitRefresh atomically replaces a directory's cached entries with the +// filer snapshot, then replays any subscription events that were buffered +// during the refresh. This ensures no creates, deletes, or updates are lost. +func (mc *MetaCache) CommitRefresh(ctx context.Context, dirPath util.FullPath, entries []*filer.Entry) error { + mc.Lock() + defer mc.Unlock() + + // Clear stale entries and insert the fresh snapshot + if err := mc.localStore.DeleteFolderChildren(ctx, dirPath); err != nil { + return err + } + if len(entries) > 0 { + if err := mc.leveldbStore.BatchInsertEntries(ctx, entries); err != nil { + return err + } + } + + // Replay buffered events so mutations that raced with the snapshot are applied + state := mc.refreshing[dirPath] + delete(mc.refreshing, dirPath) + if state != nil { + for _, ev := range state.events { + if err := mc.doAtomicUpdateEntryFromFiler(ctx, ev.oldPath, ev.newEntry); err != nil { + glog.Warningf("replay buffered event for %s: %v", dirPath, err) + } + } + } + return nil +} + +// CancelRefresh discards the refresh state without replaying buffered events. +func (mc *MetaCache) CancelRefresh(dirPath util.FullPath) { + mc.Lock() + defer mc.Unlock() + delete(mc.refreshing, dirPath) +} + +// isRefreshing returns the refresh state for the directory containing fp, +// or nil if no refresh is active. Caller must hold mc.Lock(). +func (mc *MetaCache) isRefreshingDir(dirPath util.FullPath) *refreshState { + return mc.refreshing[dirPath] +} + func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) error { mc.RLock() defer mc.RUnlock()