mount: keep a directory listing that has been walked all the way through

Every enumeration re-read the whole directory out of the local store and
decoded it again, so reopening a folder cost exactly what opening it did.
That is most of what a readdir spends: about 90% of the bytes and half
the allocations of a 200k walk.

A walk that covers a directory from its first child to its last now
leaves the decoded entries behind, and the next walk pages through them
in memory. Nothing is cached from a walk that stopped early or started
partway in, so a client reading one page of a huge directory does not pay
for the rest of it.

What makes this safe is that it is not a timer. The cache holds only what
the meta cache holds, so it is void whenever that is: every write goes
through one of four wrappers, each dropping the directory it touched
under the same lock that made the write, and a rebuild drops it at begin
and complete. A walk publishes only if its build is still the live one
for that directory, so a write landing halfway through cannot leave a
stale listing behind.

Expiry is the one thing that changes with no write at all, so it is
applied when serving rather than invalidated for -- a listing must not go
stale just because a child aged out while it sat there.

Only a listing that leaves out chunk lists is cached or served, since
those are the entries the readdir path decodes; an entry from the cache
can never reach a caller that wanted chunks.

weed/mount/meta_cache/listing_cache.go

@@ -0,0 +1,190 @@
+package meta_cache
+
+import (
+	"sync"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// DefaultListingCacheEntries is how many directory children the mount will hold
+// decoded in memory across every cached listing. A chunkless entry runs a few
+// hundred bytes, so this is tens of megabytes.
+const DefaultListingCacheEntries = 200000
+
+// listingCache keeps directories that have been walked all the way through, so
+// walking one again costs no store reads and no decoding. It holds only what the
+// meta cache already holds, and is void the moment that is: every store write
+// goes through one of a handful of methods, and each drops the directory it
+// touched.
+//
+// The entries are shared with whoever listed them, and are never written to
+// after publication.
+type listingCache struct {
+	mu       sync.Mutex
+	maxTotal int
+	total    int
+	listings map[util.FullPath]*cachedListing
+	// builds are the walks in progress. A walk publishes only if its build is
+	// still the live one for that directory when it finishes, so a write landing
+	// halfway through cannot leave a stale listing behind.
+	builds map[util.FullPath]*listingBuild
+}
+
+type cachedListing struct {
+	entries    []*filer.Entry
+	lastAccess time.Time
+}
+
+type listingBuild struct {
+	entries []*filer.Entry
+	// nextStart is the name the next page has to begin at for this build to be
+	// a continuation rather than a different walk.
+	nextStart string
+}
+
+func newListingCache(maxTotal int) *listingCache {
+	return &listingCache{
+		maxTotal: maxTotal,
+		listings: make(map[util.FullPath]*cachedListing),
+		builds:   make(map[util.FullPath]*listingBuild),
+	}
+}
+
+func (lc *listingCache) enabled() bool { return lc != nil && lc.maxTotal > 0 }
+
+// lookup returns a directory's complete listing.
+func (lc *listingCache) lookup(dirPath util.FullPath) ([]*filer.Entry, bool) {
+	if !lc.enabled() {
+		return nil, false
+	}
+	lc.mu.Lock()
+	defer lc.mu.Unlock()
+	cached, found := lc.listings[dirPath]
+	if !found {
+		return nil, false
+	}
+	cached.lastAccess = time.Now()
+	return cached.entries, true
+}
+
+// invalidate drops a directory's listing and kills any walk building one. Called
+// for every write the meta cache makes, so it has to stay cheap on a miss.
+func (lc *listingCache) invalidate(dirPath util.FullPath) {
+	if !lc.enabled() {
+		return
+	}
+	lc.mu.Lock()
+	defer lc.mu.Unlock()
+	lc.dropLocked(dirPath)
+	delete(lc.builds, dirPath)
+}
+
+// invalidateChild drops the listing of the directory holding path.
+func (lc *listingCache) invalidateChild(path util.FullPath) {
+	if !lc.enabled() {
+		return
+	}
+	dir, _ := path.DirAndName()
+	lc.invalidate(util.FullPath(dir))
+}
+
+func (lc *listingCache) invalidateAll() {
+	if !lc.enabled() {
+		return
+	}
+	lc.mu.Lock()
+	defer lc.mu.Unlock()
+	lc.listings = make(map[util.FullPath]*cachedListing)
+	lc.builds = make(map[util.FullPath]*listingBuild)
+	lc.total = 0
+}
+
+func (lc *listingCache) dropLocked(dirPath util.FullPath) {
+	if cached, found := lc.listings[dirPath]; found {
+		lc.total -= len(cached.entries)
+		delete(lc.listings, dirPath)
+	}
+}
+
+// beginOrContinue reports the build a page belongs to, or nil when this page is
+// not the next step of a walk from the directory's first child. Only a walk that
+// covers every child in order can be published.
+func (lc *listingCache) beginOrContinue(dirPath util.FullPath, startFileName string, includeStartFile bool) *listingBuild {
+	if !lc.enabled() || includeStartFile {
+		return nil
+	}
+	lc.mu.Lock()
+	defer lc.mu.Unlock()
+	if startFileName == "" {
+		build := &listingBuild{}
+		lc.builds[dirPath] = build
+		return build
+	}
+	build, found := lc.builds[dirPath]
+	if !found || build.nextStart != startFileName {
+		// A seek, a second walk at a different position, or a page that follows
+		// one this build never saw. Neither walk can be trusted to be complete.
+		delete(lc.builds, dirPath)
+		return nil
+	}
+	return build
+}
+
+// carry keeps a build alive for the next page.
+func (lc *listingCache) carry(dirPath util.FullPath, build *listingBuild, nextStart string) {
+	if build == nil {
+		return
+	}
+	lc.mu.Lock()
+	defer lc.mu.Unlock()
+	if lc.builds[dirPath] != build {
+		return // invalidated mid-walk
+	}
+	build.nextStart = nextStart
+}
+
+// publish installs a completed walk, unless a write invalidated it on the way or
+// it does not fit. Dropping is always safe: the next walk reads the store.
+func (lc *listingCache) publish(dirPath util.FullPath, build *listingBuild) {
+	if build == nil {
+		return
+	}
+	lc.mu.Lock()
+	defer lc.mu.Unlock()
+	if lc.builds[dirPath] != build {
+		return
+	}
+	delete(lc.builds, dirPath)
+	if len(build.entries) > lc.maxTotal {
+		return
+	}
+	lc.dropLocked(dirPath)
+	lc.evictLocked(len(build.entries))
+	lc.listings[dirPath] = &cachedListing{entries: build.entries, lastAccess: time.Now()}
+	lc.total += len(build.entries)
+}
+
+// evictLocked makes room for want entries, oldest use first.
+func (lc *listingCache) evictLocked(want int) {
+	for lc.total+want > lc.maxTotal && len(lc.listings) > 0 {
+		var oldestPath util.FullPath
+		var oldest time.Time
+		for path, cached := range lc.listings {
+			if oldest.IsZero() || cached.lastAccess.Before(oldest) {
+				oldestPath, oldest = path, cached.lastAccess
+			}
+		}
+		lc.dropLocked(oldestPath)
+	}
+}
+
+func (lc *listingCache) size() (dirs, entries int) {
+	if !lc.enabled() {
+		return 0, 0
+	}
+	lc.mu.Lock()
+	defer lc.mu.Unlock()
+	return len(lc.listings), lc.total
+}

weed/mount/meta_cache/listing_cache_test.go

@@ -0,0 +1,330 @@
+package meta_cache
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+func newListingTestCache(t *testing.T, maxEntries int) *MetaCache {
+	t.Helper()
+	uidGidMapper, err := NewUidGidMapper("", "")
+	if err != nil {
+		t.Fatalf("uid/gid mapper: %v", err)
+	}
+	cached := map[util.FullPath]bool{}
+	mc := NewMetaCacheWithListingCache(t.TempDir(), uidGidMapper, util.FullPath("/"), false,
+		func(p util.FullPath) { cached[p] = true },
+		func(p util.FullPath) bool { return cached[p] },
+		func(EntryInvalidation) {}, nil, maxEntries)
+	t.Cleanup(mc.Shutdown)
+	return mc
+}
+
+func seedDir(t *testing.T, mc *MetaCache, dir util.FullPath, names ...string) {
+	t.Helper()
+	now := time.Now()
+	ctx := context.Background()
+	if err := mc.InsertEntry(ctx, &filer.Entry{
+		FullPath: dir, Attr: filer.Attr{Mode: os.ModeDir | 0o755, Mtime: now, Crtime: now},
+	}, 0); err != nil {
+		t.Fatalf("insert dir: %v", err)
+	}
+	for _, name := range names {
+		if err := mc.InsertEntry(ctx, &filer.Entry{
+			FullPath: dir.Child(name), Attr: filer.Attr{Mode: 0o644, Mtime: now, Crtime: now, FileSize: 1},
+		}, 0); err != nil {
+			t.Fatalf("insert %s: %v", name, err)
+		}
+	}
+	mc.markCachedFn(util.FullPath("/"))
+	mc.markCachedFn(dir)
+}
+
+// walk lists a directory to exhaustion in pages, returning the names seen.
+func walk(t *testing.T, mc *MetaCache, ctx context.Context, dir util.FullPath, pageSize int64) []string {
+	t.Helper()
+	var seen []string
+	start := ""
+	for round := 0; round < 100; round++ {
+		count := 0
+		last, err := mc.ListDirectoryEntries(ctx, dir, start, false, pageSize, func(entry *filer.Entry) (bool, error) {
+			seen = append(seen, entry.Name())
+			count++
+			return true, nil
+		})
+		if err != nil {
+			t.Fatalf("list: %v", err)
+		}
+		if last == "" || last == start {
+			break
+		}
+		start = last
+		if int64(count) < pageSize && count == 0 {
+			break
+		}
+	}
+	return seen
+}
+
+func listingCtx() context.Context { return filer_pb.WithChunksOmitted(context.Background()) }
+
+func TestListingCacheServesAfterFullWalk(t *testing.T) {
+	mc := newListingTestCache(t, DefaultListingCacheEntries)
+	dir := util.FullPath("/d")
+	seedDir(t, mc, dir, "a", "b", "c")
+
+	if got := walk(t, mc, listingCtx(), dir, 100); len(got) != 3 {
+		t.Fatalf("first walk listed %v, want 3 entries", got)
+	}
+	dirs, entries := mc.ListingCacheSize()
+	if dirs != 1 || entries != 3 {
+		t.Fatalf("cache holds %d dirs / %d entries, want 1/3", dirs, entries)
+	}
+	if got := walk(t, mc, listingCtx(), dir, 100); len(got) != 3 {
+		t.Fatalf("second walk listed %v, want the same 3", got)
+	}
+}
+
+func TestListingCacheNotUsedWithoutChunksOmitted(t *testing.T) {
+	mc := newListingTestCache(t, DefaultListingCacheEntries)
+	dir := util.FullPath("/d")
+	seedDir(t, mc, dir, "a", "b")
+
+	// A listing that wants chunks must neither fill nor read the cache, since
+	// the cached entries were decoded without them.
+	if got := walk(t, mc, context.Background(), dir, 100); len(got) != 2 {
+		t.Fatalf("listed %v, want 2", got)
+	}
+	if dirs, _ := mc.ListingCacheSize(); dirs != 0 {
+		t.Fatalf("cache holds %d dirs, want none", dirs)
+	}
+}
+
+func TestListingCachePartialWalkIsNotCached(t *testing.T) {
+	mc := newListingTestCache(t, DefaultListingCacheEntries)
+	dir := util.FullPath("/d")
+	seedDir(t, mc, dir, "a", "b", "c", "d")
+
+	// Caller stops after the first entry: the rest of the directory was never
+	// seen, so nothing may be published.
+	if _, err := mc.ListDirectoryEntries(listingCtx(), dir, "", false, 100, func(entry *filer.Entry) (bool, error) {
+		return false, nil
+	}); err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if dirs, entries := mc.ListingCacheSize(); dirs != 0 {
+		t.Fatalf("cache holds %d dirs / %d entries after a stopped walk, want none", dirs, entries)
+	}
+}
+
+func TestListingCacheSeekDoesNotBuild(t *testing.T) {
+	mc := newListingTestCache(t, DefaultListingCacheEntries)
+	dir := util.FullPath("/d")
+	seedDir(t, mc, dir, "a", "b", "c")
+
+	// Starting partway through is not a walk of the whole directory.
+	if _, err := mc.ListDirectoryEntries(listingCtx(), dir, "a", false, 100, func(entry *filer.Entry) (bool, error) {
+		return true, nil
+	}); err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if dirs, _ := mc.ListingCacheSize(); dirs != 0 {
+		t.Fatalf("cache holds %d dirs after a mid-directory listing, want none", dirs)
+	}
+}
+
+func TestListingCacheInvalidation(t *testing.T) {
+	dir := util.FullPath("/d")
+	now := time.Now()
+	cases := []struct {
+		name   string
+		mutate func(t *testing.T, mc *MetaCache)
+	}{
+		{"insert a child", func(t *testing.T, mc *MetaCache) {
+			if err := mc.InsertEntry(context.Background(), &filer.Entry{
+				FullPath: dir.Child("new"), Attr: filer.Attr{Mode: 0o644, Mtime: now, Crtime: now},
+			}, 0); err != nil {
+				t.Fatal(err)
+			}
+		}},
+		{"update a child", func(t *testing.T, mc *MetaCache) {
+			if err := mc.UpdateEntry(context.Background(), &filer.Entry{
+				FullPath: dir.Child("a"), Attr: filer.Attr{Mode: 0o644, Mtime: now, Crtime: now, FileSize: 99},
+			}); err != nil {
+				t.Fatal(err)
+			}
+		}},
+		{"delete a child", func(t *testing.T, mc *MetaCache) {
+			if err := mc.DeleteEntry(context.Background(), dir.Child("a")); err != nil {
+				t.Fatal(err)
+			}
+		}},
+		{"delete the children", func(t *testing.T, mc *MetaCache) {
+			if err := mc.DeleteFolderChildren(context.Background(), dir); err != nil {
+				t.Fatal(err)
+			}
+		}},
+		{"rename into the directory", func(t *testing.T, mc *MetaCache) {
+			if err := mc.AtomicUpdateEntryFromFiler(context.Background(), "", &filer.Entry{
+				FullPath: dir.Child("moved"), Attr: filer.Attr{Mode: 0o644, Mtime: now, Crtime: now},
+			}); err != nil {
+				t.Fatal(err)
+			}
+		}},
+		{"purge the children", func(t *testing.T, mc *MetaCache) {
+			mc.PurgeDirectoryChildren(dir, func() {})
+		}},
+		{"distrust everything", func(t *testing.T, mc *MetaCache) {
+			mc.InvalidateAllListings()
+		}},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			mc := newListingTestCache(t, DefaultListingCacheEntries)
+			seedDir(t, mc, dir, "a", "b", "c")
+			walk(t, mc, listingCtx(), dir, 100)
+			if dirs, _ := mc.ListingCacheSize(); dirs != 1 {
+				t.Fatalf("expected the walk to cache the directory, got %d dirs", dirs)
+			}
+			tc.mutate(t, mc)
+			if dirs, entries := mc.ListingCacheSize(); dirs != 0 {
+				t.Errorf("cache still holds %d dirs / %d entries after %s", dirs, entries, tc.name)
+			}
+		})
+	}
+}
+
+func TestListingCacheSeesAWriteThatLandsMidWalk(t *testing.T) {
+	mc := newListingTestCache(t, DefaultListingCacheEntries)
+	dir := util.FullPath("/d")
+	seedDir(t, mc, dir, "a", "b", "c", "d")
+	ctx := listingCtx()
+
+	// First page of a two-page walk.
+	last, err := mc.ListDirectoryEntries(ctx, dir, "", false, 2, func(entry *filer.Entry) (bool, error) {
+		return true, nil
+	})
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+
+	// A write lands before the walk finishes.
+	if err := mc.InsertEntry(context.Background(), &filer.Entry{
+		FullPath: dir.Child("e"), Attr: filer.Attr{Mode: 0o644, Mtime: time.Now(), Crtime: time.Now()},
+	}, 0); err != nil {
+		t.Fatal(err)
+	}
+
+	// Finishing the walk must not publish what it started collecting.
+	if _, err := mc.ListDirectoryEntries(ctx, dir, last, false, 100, func(entry *filer.Entry) (bool, error) {
+		return true, nil
+	}); err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if dirs, entries := mc.ListingCacheSize(); dirs != 0 {
+		t.Fatalf("published %d dirs / %d entries from a walk a write interrupted", dirs, entries)
+	}
+
+	// A fresh walk sees all five.
+	if got := walk(t, mc, ctx, dir, 100); len(got) != 5 {
+		t.Fatalf("walk listed %v, want 5 entries", got)
+	}
+}
+
+func TestListingCacheAppliesExpiryOnServe(t *testing.T) {
+	mc := newListingTestCache(t, DefaultListingCacheEntries)
+	dir := util.FullPath("/d")
+	now := time.Now()
+	ctx := context.Background()
+	if err := mc.InsertEntry(ctx, &filer.Entry{
+		FullPath: dir, Attr: filer.Attr{Mode: os.ModeDir | 0o755, Mtime: now, Crtime: now},
+	}, 0); err != nil {
+		t.Fatal(err)
+	}
+	// "b" expires two seconds from now; nothing will write when it does.
+	for _, e := range []struct {
+		name   string
+		ttlSec int32
+		crtime time.Time
+	}{{"a", 0, now}, {"b", 2, now}, {"c", 0, now}} {
+		if err := mc.InsertEntry(ctx, &filer.Entry{
+			FullPath: dir.Child(e.name),
+			Attr:     filer.Attr{Mode: 0o644, Mtime: now, Crtime: e.crtime, TtlSec: e.ttlSec},
+		}, 0); err != nil {
+			t.Fatal(err)
+		}
+	}
+	mc.markCachedFn(util.FullPath("/"))
+	mc.markCachedFn(dir)
+
+	if got := walk(t, mc, listingCtx(), dir, 100); len(got) != 3 {
+		t.Fatalf("first walk listed %v, want 3", got)
+	}
+
+	// Age "b" past its TTL in place, the way the clock would, and confirm the
+	// cached listing stops reporting it without anything invalidating.
+	entry, _, err := mc.FindEntry(ctx, dir.Child("b"))
+	if err != nil {
+		t.Fatalf("find: %v", err)
+	}
+	entry.Crtime = now.Add(-time.Hour)
+	cachedEntries, found := mc.listings.lookup(dir)
+	if !found {
+		t.Fatal("expected the directory to be cached")
+	}
+	for _, e := range cachedEntries {
+		if e.Name() == "b" {
+			e.Crtime = now.Add(-time.Hour)
+		}
+	}
+
+	got := walk(t, mc, listingCtx(), dir, 100)
+	for _, name := range got {
+		if name == "b" {
+			t.Fatalf("expired entry still listed: %v", got)
+		}
+	}
+	if len(got) != 2 {
+		t.Fatalf("listed %v, want a and c", got)
+	}
+}
+
+func TestListingCacheEvictsToStayWithinBudget(t *testing.T) {
+	mc := newListingTestCache(t, 6)
+	ctx := listingCtx()
+	for d := 0; d < 4; d++ {
+		dir := util.FullPath(fmt.Sprintf("/d%d", d))
+		seedDir(t, mc, dir, "a", "b", "c")
+		if got := walk(t, mc, ctx, dir, 100); len(got) != 3 {
+			t.Fatalf("%s listed %v, want 3", dir, got)
+		}
+		_, entries := mc.ListingCacheSize()
+		if entries > 6 {
+			t.Fatalf("cache holds %d entries, over the 6 budget", entries)
+		}
+	}
+	dirs, entries := mc.ListingCacheSize()
+	if dirs != 2 || entries != 6 {
+		t.Fatalf("cache holds %d dirs / %d entries, want 2/6 after eviction", dirs, entries)
+	}
+}
+
+func TestListingCacheDisabled(t *testing.T) {
+	mc := newListingTestCache(t, 0)
+	dir := util.FullPath("/d")
+	seedDir(t, mc, dir, "a", "b")
+	if got := walk(t, mc, listingCtx(), dir, 100); len(got) != 2 {
+		t.Fatalf("listed %v, want 2", got)
+	}
+	if dirs, entries := mc.ListingCacheSize(); dirs != 0 || entries != 0 {
+		t.Fatalf("disabled cache holds %d dirs / %d entries", dirs, entries)
+	}
+}

weed/mount/meta_cache/meta_cache.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"math"
 	"os"
+	"sort"
 	"sync"
 	"time"
 
@@ -43,6 +44,11 @@ type MetaCache struct {
 	dedupRing            dedupRingBuffer
 	includeSystemEntries bool
 
+	// listings holds directories walked all the way through, so walking one
+	// again reads and decodes nothing. Every store write drops the directory it
+	// touched; see the storeX wrappers.
+	listings *listingCache
+
 	// dirVersionFloors is each cached directory's listing snapshot: the
 	// version of every child the listing covered, present or absent, unless
 	// a later event gave that child its own record. One map write per build
@@ -103,6 +109,13 @@ type metadataApplyRequest struct {
 
 func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPath, includeSystemEntries bool,
 	markCachedFn func(path util.FullPath), isCachedFn func(path util.FullPath) bool, invalidateFunc func(EntryInvalidation), onDirectoryUpdate func(dir util.FullPath)) *MetaCache {
+	return NewMetaCacheWithListingCache(dbFolder, uidGidMapper, root, includeSystemEntries, markCachedFn, isCachedFn, invalidateFunc, onDirectoryUpdate, DefaultListingCacheEntries)
+}
+
+// NewMetaCacheWithListingCache sizes the directory listing cache; zero disables it.
+func NewMetaCacheWithListingCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPath, includeSystemEntries bool,
+	markCachedFn func(path util.FullPath), isCachedFn func(path util.FullPath) bool, invalidateFunc func(EntryInvalidation), onDirectoryUpdate func(dir util.FullPath),
+	listingCacheEntries int) *MetaCache {
 	leveldbStore, virtualStore := openMetaStore(dbFolder)
 	mc := &MetaCache{
 		root:                 root,
@@ -119,6 +132,7 @@ func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPat
 		buildingDirs:         make(map[util.FullPath]*directoryBuildState),
 		dedupRing:            newDedupRingBuffer(),
 		dirVersionFloors:     make(map[util.FullPath]int64),
+		listings:             newListingCache(listingCacheEntries),
 	}
 	mc.invalidateWorker = util.NewAsyncBatchWorker(func(batch []EntryInvalidation) {
 		for _, invalidation := range batch {
@@ -157,7 +171,7 @@ func (mc *MetaCache) InsertEntry(ctx context.Context, entry *filer.Entry, versio
 }
 
 func (mc *MetaCache) doInsertEntry(ctx context.Context, entry *filer.Entry, versionTsNs int64) error {
-	if err := mc.localStore.InsertEntry(ctx, entry); err != nil {
+	if err := mc.storeInsertEntry(ctx, entry); err != nil {
 		return err
 	}
 	mc.setEntryVersionLocked(ctx, entry.FullPath, versionTsNs)
@@ -187,7 +201,7 @@ func (mc *MetaCache) atomicUpdateEntryFromFilerLocked(ctx context.Context, oldPa
 	if entry != nil && vacatingOldPath {
 		ctx = context.WithValue(ctx, "OP", "MV")
 		glog.V(3).Infof("DeleteEntry %s", oldPath)
-		if err := mc.localStore.DeleteEntry(ctx, oldPath); err != nil {
+		if err := mc.storeDeleteEntry(ctx, oldPath); err != nil {
 			return err
 		}
 	}
@@ -209,7 +223,7 @@ func (mc *MetaCache) atomicUpdateEntryFromFilerLocked(ctx context.Context, oldPa
 		newDir, _ := newEntry.DirAndName()
 		if allowUncachedInsert || mc.isCachedFn(util.FullPath(newDir)) {
 			glog.V(3).Infof("InsertEntry %s/%s", newDir, newEntry.Name())
-			if err := mc.localStore.InsertEntry(ctx, newEntry); err != nil {
+			if err := mc.storeInsertEntry(ctx, newEntry); err != nil {
 				return err
 			}
 			mc.setEntryVersionLocked(ctx, newEntry.FullPath, versionTsNs)
@@ -230,11 +244,11 @@ func (mc *MetaCache) purgeEntryLocked(ctx context.Context, fullpath util.FullPat
 	if fullpath == "" {
 		return nil
 	}
-	if err := mc.localStore.DeleteEntry(ctx, fullpath); err != nil {
+	if err := mc.storeDeleteEntry(ctx, fullpath); err != nil {
 		return err
 	}
 	if isDirectory {
-		if err := mc.localStore.DeleteFolderChildren(ctx, fullpath); err != nil {
+		if err := mc.storeDeleteFolderChildren(ctx, fullpath); err != nil {
 			return err
 		}
 	}
@@ -303,7 +317,10 @@ func (mc *MetaCache) applyMetadataResponseEnqueue(ctx context.Context, resp *fil
 	}
 }
 
+// BeginDirectoryBuild drops the cached listing: a rebuild replaces the whole
+// child set, and nothing may be served from the old one meanwhile.
 func (mc *MetaCache) BeginDirectoryBuild(ctx context.Context, dirPath util.FullPath) error {
+	mc.listings.invalidate(dirPath)
 	return mc.enqueueAndWait(ctx, metadataApplyRequest{
 		kind:      metadataBeginBuild,
 		buildPath: dirPath,
@@ -311,6 +328,7 @@ func (mc *MetaCache) BeginDirectoryBuild(ctx context.Context, dirPath util.FullP
 }
 
 func (mc *MetaCache) CompleteDirectoryBuild(ctx context.Context, dirPath util.FullPath, snapshotTsNs int64) error {
+	mc.listings.invalidate(dirPath)
 	return mc.enqueueAndWait(ctx, metadataApplyRequest{
 		kind:         metadataCompleteBuild,
 		buildPath:    dirPath,
@@ -319,6 +337,7 @@ func (mc *MetaCache) CompleteDirectoryBuild(ctx context.Context, dirPath util.Fu
 }
 
 func (mc *MetaCache) AbortDirectoryBuild(ctx context.Context, dirPath util.FullPath) error {
+	mc.listings.invalidate(dirPath)
 	return mc.enqueueAndWait(ctx, metadataApplyRequest{
 		kind:      metadataAbortBuild,
 		buildPath: dirPath,
@@ -330,6 +349,7 @@ func (mc *MetaCache) AbortDirectoryBuild(ctx context.Context, dirPath util.FullP
 // like kernel Forget don't block; see purgeDirectoryChildrenNow for why off-loop
 // callers must route through here rather than wiping the store directly.
 func (mc *MetaCache) PurgeDirectoryChildren(dirPath util.FullPath, resetFn func()) {
+	mc.listings.invalidate(dirPath)
 	_ = mc.enqueueApplyRequest(metadataApplyRequest{
 		ctx:       context.Background(),
 		kind:      metadataPurgeDir,
@@ -342,7 +362,7 @@ func (mc *MetaCache) PurgeDirectoryChildren(dirPath util.FullPath, resetFn func(
 func (mc *MetaCache) UpdateEntry(ctx context.Context, entry *filer.Entry) error {
 	mc.Lock()
 	defer mc.Unlock()
-	if err := mc.localStore.UpdateEntry(ctx, entry); err != nil {
+	if err := mc.storeUpdateEntry(ctx, entry); err != nil {
 		return err
 	}
 	mc.markEntryUnversionedLocked(ctx, entry.FullPath)
@@ -365,7 +385,7 @@ func (mc *MetaCache) TouchDirMtimeCtime(ctx context.Context, dirPath util.FullPa
 	}
 	entry.Attr.Mtime = now
 	entry.Attr.Ctime = now
-	if err := mc.localStore.UpdateEntry(ctx, entry); err != nil {
+	if err := mc.storeUpdateEntry(ctx, entry); err != nil {
 		return err
 	}
 	mc.markEntryUnversionedLocked(ctx, dirPath)
@@ -576,7 +596,7 @@ func (mc *MetaCache) deleteChildVersionRecordsLocked(ctx context.Context, dirPat
 func (mc *MetaCache) DeleteEntry(ctx context.Context, fp util.FullPath) (err error) {
 	mc.Lock()
 	defer mc.Unlock()
-	if err = mc.localStore.DeleteEntry(ctx, fp); err != nil {
+	if err = mc.storeDeleteEntry(ctx, fp); err != nil {
 		return err
 	}
 	mc.clearEntryVersionLocked(ctx, fp)
@@ -587,7 +607,7 @@ func (mc *MetaCache) DeleteFolderChildren(ctx context.Context, fp util.FullPath)
 	defer mc.Unlock()
 	delete(mc.dirVersionFloors, fp)
 	mc.deleteChildVersionRecordsLocked(ctx, fp)
-	return mc.localStore.DeleteFolderChildren(ctx, fp)
+	return mc.storeDeleteFolderChildren(ctx, fp)
 }
 
 // SetPinnedChildFn installs a predicate reporting whether a child holds
@@ -606,7 +626,7 @@ func (mc *MetaCache) deleteFolderChildrenForRebuild(ctx context.Context, dirPath
 	mc.Lock()
 	defer mc.Unlock()
 	if mc.pinnedChildFn == nil {
-		return mc.localStore.DeleteFolderChildren(ctx, dirPath)
+		return mc.storeDeleteFolderChildren(ctx, dirPath)
 	}
 	var pinned []*filer.Entry
 	if _, err := mc.localStore.ListDirectoryEntries(ctx, dirPath, "", true, math.MaxInt64, func(entry *filer.Entry) (bool, error) {
@@ -617,7 +637,7 @@ func (mc *MetaCache) deleteFolderChildrenForRebuild(ctx context.Context, dirPath
 	}); err != nil {
 		return err
 	}
-	if err := mc.localStore.DeleteFolderChildren(ctx, dirPath); err != nil {
+	if err := mc.storeDeleteFolderChildren(ctx, dirPath); err != nil {
 		return err
 	}
 	if len(pinned) > 0 {
@@ -631,6 +651,10 @@ func (mc *MetaCache) deleteFolderChildrenForRebuild(ctx context.Context, dirPath
 // the store has already spent it against limit. A caller paginating by count
 // would read a short batch as the end of the directory, and one resuming from
 // the last name it saw would re-read the dropped ones forever.
+//
+// A directory already walked all the way through is served from memory. Only a
+// listing that leaves out chunk lists is cached or served, so an entry from the
+// cache can never be handed to a caller that wanted chunks.
 func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
 	mc.RLock()
 	defer mc.RUnlock()
@@ -638,17 +662,128 @@ func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.Full
 	if !mc.isCachedFn(dirPath) {
 		// if this request comes after renaming, it should be fine
 		glog.Warningf("unsynchronized dir: %v", dirPath)
+		mc.listings.invalidate(dirPath)
+		return mc.listFromStore(ctx, dirPath, startFileName, includeStartFile, limit, eachEntryFunc, nil)
 	}
 
+	cacheable := filer_pb.ChunksOmitted(ctx)
+	if cacheable {
+		if entries, found := mc.listings.lookup(dirPath); found {
+			return serveCachedListing(entries, startFileName, includeStartFile, limit, eachEntryFunc)
+		}
+	}
+
+	var build *listingBuild
+	if cacheable {
+		build = mc.listings.beginOrContinue(dirPath, startFileName, includeStartFile)
+	}
+
+	storeCount := int64(0)
+	stoppedEarly := false
+	wrapped := func(entry *filer.Entry) (bool, error) {
+		ok, err := eachEntryFunc(entry)
+		if err != nil || !ok {
+			stoppedEarly = true
+		}
+		return ok, err
+	}
+	lastFileName, err = mc.listFromStore(ctx, dirPath, startFileName, includeStartFile, limit, wrapped, func(entry *filer.Entry) {
+		storeCount++
+		if build != nil {
+			build.entries = append(build.entries, entry)
+		}
+	})
+	if err != nil {
+		mc.listings.invalidate(dirPath)
+		return lastFileName, err
+	}
+	if build != nil {
+		// A short page means the store ran out only if the caller did not stop
+		// it first. A caller whose buffer filled has seen part of a page, and
+		// the rest of the directory is still out there.
+		if storeCount < limit && !stoppedEarly {
+			mc.listings.publish(dirPath, build)
+		} else {
+			mc.listings.carry(dirPath, build, lastFileName)
+		}
+	}
+	return lastFileName, nil
+}
+
+// listFromStore reads a page, dropping expired children and mapping ids. observe,
+// when set, sees every entry the store produced, expired ones included, since
+// those are what limit was spent on.
+func (mc *MetaCache) listFromStore(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc, observe func(*filer.Entry)) (string, error) {
 	return mc.localStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *filer.Entry) (bool, error) {
-		if entry.TtlSec > 0 && entry.Crtime.Add(time.Duration(entry.TtlSec)*time.Second).Before(time.Now()) {
+		mc.mapIdFromFilerToLocal(entry)
+		if observe != nil {
+			observe(entry)
+		}
+		if isTtlExpired(entry) {
 			return true, nil
 		}
-		mc.mapIdFromFilerToLocal(entry)
 		return eachEntryFunc(entry)
 	})
 }
 
+// serveCachedListing walks an in-memory listing the way the store would: limit
+// counts every child it steps over, expired or not, and lastFileName is the last
+// one stepped over rather than the last one reported. Expiry is applied here
+// rather than at cache time, so a listing does not go stale as children age out.
+func serveCachedListing(entries []*filer.Entry, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
+	start := sort.Search(len(entries), func(i int) bool {
+		if includeStartFile {
+			return entries[i].Name() >= startFileName
+		}
+		return entries[i].Name() > startFileName
+	})
+	for i := start; i < len(entries) && int64(i-start) < limit; i++ {
+		entry := entries[i]
+		lastFileName = entry.Name()
+		if isTtlExpired(entry) {
+			continue
+		}
+		if ok, err := eachEntryFunc(entry); err != nil || !ok {
+			return lastFileName, err
+		}
+	}
+	return lastFileName, nil
+}
+
+// Every write to the local store goes through these, so a directory's cached
+// listing cannot outlive a change to it. Reaching past them to mc.localStore
+// for a write would leave the listing stale.
+func (mc *MetaCache) storeInsertEntry(ctx context.Context, entry *filer.Entry) error {
+	mc.listings.invalidateChild(entry.FullPath)
+	return mc.localStore.InsertEntry(ctx, entry)
+}
+
+func (mc *MetaCache) storeUpdateEntry(ctx context.Context, entry *filer.Entry) error {
+	mc.listings.invalidateChild(entry.FullPath)
+	return mc.localStore.UpdateEntry(ctx, entry)
+}
+
+func (mc *MetaCache) storeDeleteEntry(ctx context.Context, fp util.FullPath) error {
+	mc.listings.invalidateChild(fp)
+	return mc.localStore.DeleteEntry(ctx, fp)
+}
+
+func (mc *MetaCache) storeDeleteFolderChildren(ctx context.Context, fp util.FullPath) error {
+	mc.listings.invalidate(fp)
+	return mc.localStore.DeleteFolderChildren(ctx, fp)
+}
+
+// InvalidateAllListings distrusts every cached listing, for when a subscription
+// gap means the mount cannot know what it missed.
+func (mc *MetaCache) InvalidateAllListings() {
+	mc.listings.invalidateAll()
+}
+
+// ListingCacheSize reports the cached directories and the children they hold.
+func (mc *MetaCache) ListingCacheSize() (dirs, entries int) {
+	return mc.listings.size()
+}
+
 func (mc *MetaCache) Shutdown() {
 	done := make(chan error, 1)
 
@@ -935,7 +1070,7 @@ func (mc *MetaCache) applyMetadataResponseLocked(ctx context.Context, resp *file
 		isDelete := message.NewEntry == nil
 		isMove := message.NewEntry != nil && (message.NewParentPath != resp.Directory || message.NewEntry.Name != message.OldEntry.Name)
 		if isDelete || isMove {
-			if deleteErr := mc.localStore.DeleteFolderChildren(ctx, oldPath); deleteErr != nil {
+			if deleteErr := mc.storeDeleteFolderChildren(ctx, oldPath); deleteErr != nil {
 				glog.V(2).Infof("delete descendants of %s: %v", oldPath, deleteErr)
 			}
 		}
@@ -977,7 +1112,7 @@ func (mc *MetaCache) purgeDirectoryChildrenNow(ctx context.Context, dirPath util
 	defer mc.Unlock()
 	delete(mc.dirVersionFloors, dirPath)
 	mc.deleteChildVersionRecordsLocked(ctx, dirPath)
-	return mc.localStore.DeleteFolderChildren(ctx, dirPath)
+	return mc.storeDeleteFolderChildren(ctx, dirPath)
 }
 
 func (mc *MetaCache) completeDirectoryBuildNow(ctx context.Context, dirPath util.FullPath, snapshotTsNs int64) error {