From 5456f9d695ee1751b2f0810c4a2bb404b918086b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 24 Jun 2026 14:25:23 -0700 Subject: [PATCH] mount: confirm an empty directory rebuild before caching it (#10092) A directory rebuild wiped the cached children, listed the filer once, and published the directory authoritatively cached over whatever came back. A transient empty listing -- a momentary list-stream glitch that ends as a clean EOF with no entries -- then stranded a populated directory cached over an empty store, hiding every file in it until some unrelated event happened to rebuild it: stat returns ENOENT and readdir returns nothing though the files are safe on the filer, and nothing re-triggers a build. Re-read the directory when the listing comes back empty before trusting it. The first re-read is immediate, since the likely transient clears on a fresh stream; later attempts space out. A genuinely empty directory still lists empty every time and caches as before, so only empty listings pay the extra read. Co-authored-by: Claude Opus 4.8 --- .../mount/meta_cache/meta_cache_build_test.go | 76 ++++++++++++++++ weed/mount/meta_cache/meta_cache_init.go | 89 +++++++++++++------ 2 files changed, 136 insertions(+), 29 deletions(-) diff --git a/weed/mount/meta_cache/meta_cache_build_test.go b/weed/mount/meta_cache/meta_cache_build_test.go index ead21dae2..64ce96aca 100644 --- a/weed/mount/meta_cache/meta_cache_build_test.go +++ b/weed/mount/meta_cache/meta_cache_build_test.go @@ -557,3 +557,79 @@ func TestEnsureVisitedDropsUnpinnedStaleEntry(t *testing.T) { t.Fatalf("unpinned stale entry survived rebuild = %+v, %v; want nil, %v", entry, err, filer_pb.ErrNotFound) } } + +// sequencedListClient returns a different ListEntries result per call (repeating +// the last), modelling a filer that lists empty transiently then the real entries. +type sequencedListClient struct { + filer_pb.SeaweedFilerClient + mu sync.Mutex + perCall [][]*filer_pb.ListEntriesResponse + calls int +} + +func (c *sequencedListClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) { + c.mu.Lock() + idx := c.calls + if idx >= len(c.perCall) { + idx = len(c.perCall) - 1 + } + c.calls++ + resp := c.perCall[idx] + c.mu.Unlock() + return &buildListStream{responses: resp}, nil +} + +// TestEnsureVisitedConfirmsTransientEmptyListing: a rebuild whose first filer +// listing comes back empty must re-read and cache the real entries, not strand +// the directory cached over an empty store (the ConcurrentReadWrite ENOENT flake). +func TestEnsureVisitedConfirmsTransientEmptyListing(t *testing.T) { + mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{"/": true}) + defer mc.Shutdown() + + entry := &filer_pb.Entry{ + Name: "keep.txt", + Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 1, FileMode: 0100644, FileSize: 7}, + } + accessor := &buildFilerAccessor{client: &sequencedListClient{ + perCall: [][]*filer_pb.ListEntriesResponse{ + {}, // first read: transient empty + {{Entry: entry, SnapshotTsNs: 100}}, // confirm read: the real entry + }, + }} + + if err := EnsureVisited(mc, accessor, util.FullPath("/dir")); err != nil { + t.Fatalf("ensure visited: %v", err) + } + if !mc.IsDirectoryCached(util.FullPath("/dir")) { + t.Fatal("/dir should be cached after build completes") + } + if _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/keep.txt")); err != nil { + t.Fatalf("/dir/keep.txt stranded after transient empty listing: %v", err) + } +} + +// TestEnsureVisitedCachesGenuinelyEmptyDirectory: a really-empty directory lists +// empty on every confirm and must still end up cached. +func TestEnsureVisitedCachesGenuinelyEmptyDirectory(t *testing.T) { + mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{"/": true}) + defer mc.Shutdown() + + client := &sequencedListClient{ + perCall: [][]*filer_pb.ListEntriesResponse{{}}, // always empty + } + accessor := &buildFilerAccessor{client: client} + + if err := EnsureVisited(mc, accessor, util.FullPath("/empty")); err != nil { + t.Fatalf("ensure visited: %v", err) + } + if !mc.IsDirectoryCached(util.FullPath("/empty")) { + t.Fatal("/empty should be cached even though it has no entries") + } + // The empty result must have been confirmed, not trusted on the first read. + client.mu.Lock() + calls := client.calls + client.mu.Unlock() + if calls != emptyRebuildConfirmations+1 { + t.Fatalf("list calls = %d, want %d (initial + confirmations)", calls, emptyRebuildConfirmations+1) + } +} diff --git a/weed/mount/meta_cache/meta_cache_init.go b/weed/mount/meta_cache/meta_cache_init.go index 1ce01b79a..a7df7f523 100644 --- a/weed/mount/meta_cache/meta_cache_init.go +++ b/weed/mount/meta_cache/meta_cache_init.go @@ -3,6 +3,7 @@ package meta_cache import ( "context" "fmt" + "time" "golang.org/x/sync/errgroup" @@ -54,6 +55,11 @@ func EnsureVisited(mc *MetaCache, client filer_pb.FilerClient, dirPath util.Full // (fewer disk syncs). Larger values reduce I/O overhead but increase memory and latency. const batchInsertSize = 100 +const ( + emptyRebuildConfirmations = 2 + emptyRebuildConfirmDelay = 50 * time.Millisecond +) + func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerClient, path util.FullPath) error { // Use singleflight to deduplicate concurrent requests for the same path _, err, _ := mc.visitGroup.Do(string(path), func() (interface{}, error) { @@ -95,51 +101,76 @@ func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerCl } }() - // Collect entries in batches for efficient LevelDB writes - var batch []*filer.Entry - var snapshotTsNs int64 - - fetchErr := util.Retry("ReadDirAllEntries", func() error { - batch = nil // Reset batch on retry, allow GC of previous entries - if err := mc.deleteFolderChildrenForRebuild(ctx, path); err != nil { - return fmt.Errorf("clear existing entries for %s: %w", path, err) - } - var err error - snapshotTsNs, err = filer_pb.ReadDirAllEntriesWithSnapshot(ctx, client, path, "", func(pbEntry *filer_pb.Entry, isLast bool) error { - entry := filer.FromPbEntry(string(path), pbEntry) - if !mc.includeSystemEntries && IsHiddenSystemEntry(string(path), entry.Name()) { - return nil + // reloadFromFiler wipes the cached children and reloads them from the filer. + reloadFromFiler := func() (entryCount int, snapshotTsNs int64, err error) { + err = util.Retry("ReadDirAllEntries", func() error { + entryCount = 0 + var batch []*filer.Entry // reset on retry, allow GC of previous entries + if err := mc.deleteFolderChildrenForRebuild(ctx, path); err != nil { + return fmt.Errorf("clear existing entries for %s: %w", path, err) } + var listErr error + snapshotTsNs, listErr = filer_pb.ReadDirAllEntriesWithSnapshot(ctx, client, path, "", func(pbEntry *filer_pb.Entry, isLast bool) error { + entry := filer.FromPbEntry(string(path), pbEntry) + if !mc.includeSystemEntries && IsHiddenSystemEntry(string(path), entry.Name()) { + return nil + } - batch = append(batch, entry) + batch = append(batch, entry) + entryCount++ - // Flush batch when it reaches the threshold - // Don't rely on isLast here - hidden entries may cause early return - if len(batch) >= batchInsertSize { - // No lock needed - LevelDB Write() is thread-safe + // flush by size, not isLast: hidden entries can return early + if len(batch) >= batchInsertSize { + if err := mc.doBatchInsertEntries(ctx, batch); err != nil { + return fmt.Errorf("batch insert for %s: %w", path, err) + } + batch = make([]*filer.Entry, 0, batchInsertSize) + } + return nil + }) + if listErr != nil { + return listErr + } + if len(batch) > 0 { if err := mc.doBatchInsertEntries(ctx, batch); err != nil { - return fmt.Errorf("batch insert for %s: %w", path, err) + return fmt.Errorf("batch insert remaining for %s: %w", path, err) } - // Create new slice to allow GC of flushed entries - batch = make([]*filer.Entry, 0, batchInsertSize) } return nil }) - return err - }) + return entryCount, snapshotTsNs, err + } + entryCount, snapshotTsNs, fetchErr := reloadFromFiler() if fetchErr != nil { cleanupBuild("failed") return nil, fmt.Errorf("list %s: %w", path, fetchErr) } - // Flush any remaining entries in the batch - if len(batch) > 0 { - if err := mc.doBatchInsertEntries(ctx, batch); err != nil { - cleanupBuild("incomplete") - return nil, fmt.Errorf("batch insert remaining for %s: %w", path, err) + // A transient empty listing would strand a populated directory cached over + // an empty store; re-read to confirm before trusting it. First re-read is + // immediate (a clean-EOF stream glitch clears at once), later ones space out. + // On cancellation the deferred cleanup aborts the build. + for attempt := 0; entryCount == 0 && attempt < emptyRebuildConfirmations; attempt++ { + if ctx.Err() != nil { + return nil, ctx.Err() + } + if attempt > 0 { + select { + case <-time.After(emptyRebuildConfirmDelay): + case <-ctx.Done(): + return nil, ctx.Err() + } + } + if entryCount, snapshotTsNs, fetchErr = reloadFromFiler(); fetchErr != nil { + cleanupBuild("failed") + return nil, fmt.Errorf("confirm empty list %s: %w", path, fetchErr) + } + if entryCount > 0 { + glog.Warningf("rebuild of %s saw a transient empty listing, recovered %d entries on confirmation", path, entryCount) } } + if err := mc.CompleteDirectoryBuild(context.Background(), path, snapshotTsNs); err != nil { cleanupBuild("unreplayed") return nil, fmt.Errorf("complete build for %s: %w", path, err)