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 <noreply@anthropic.com>

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)
+	}
+}

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)