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()