mount: invalidate hot directory listings by section (#10712)

* mount: invalidate hot directory listings by section

A cached directory used to be dropped whole when it saw 64 changes in
2s: with a continuous writer the listing cycled through wipe, direct
listing and full rebuild for as long as the writer kept going, and
every sibling lookup fell through to the filer in between.

Split each cached listing into name-range sections of 1024 entries. A
burst of foreign changes invalidates just the section it lands in;
entries stay served and events keep applying, and the next readdir
re-lists only that range from the filer, reconciled through the version
gate so it cannot roll back newer applied events. Lookups in an
invalidated section read through until then. The mount's own writes no
longer invalidate anything: they are ground truth for its cache.

* meta_cache: drop the version floor with a deleted or moved directory

The other teardown paths already clear both maps; a floor left behind
here would fence the listing of a directory re-created at the same
path.

* mount: harden section refresh

An unversioned listing (pre-upgrade filer) now only fills gaps instead
of reconciling: without a snapshot to order against, an overwrite or
the deletion sweep could roll back an event applied after the listing.

The section table can be rebuilt or re-split between the listing and
its apply, so the refresh only marks fresh or splits when the section
still covers the range it read. Splicing bounds from a stale range
into a rebuilt table could leave them unsorted.

Bound the wait: a readdir gives a refresh five seconds before serving
the maintained-but-unverified cache. Bound the size: a range grown
past four sections aborts the refresh and drops the directory cache,
re-tiling it with a full rebuild, with that request served direct.

Cover the filer-facing path with a listing server: paging with the
snapshot pinned across pages, the section cutoff, no calls for a
fresh section, and the overgrown-range abort.

* meta_cache: make the section table a self-contained state machine

Churn counting, freshness, stale-range scanning and the refresh
completion with its guard and split now live on dirSections itself,
free of the lock, the store and the apply loop, so they test directly
with synthetic clocks and tables. MetaCache keeps thin wrappers that
hold its mutex and find the directory's table.

* meta_cache: keep section internals out of the apply request

The request now carries the completed build's table and one refresh as
opaque values built by section code, and the boundary-derivation rule
moves out of the build loop into a collector next to the rest of the
section logic.

* mount: fence refreshed sections with a snapshot floor

A refresh versioned the entries it fetched and tombstoned the ones it
swept, but a name absent from both cache and listing kept the old
directory floor, so a delayed event between the two snapshots could
resurrect it into a section already marked fresh. The section now
carries its own floor, consulted next to the directory floor, covering
every name in the range, present or absent — which also retires the
refresh's per-entry version stamps and sweep tombstones.

An unversioned listing sets no floor and vouches for nothing: it may
still fill gaps, but the section stays stale and reads through until a
filer that stamps snapshots re-validates it.

A listing's reach is unknowable up front — a resumed handle can skip
far ahead, and shrunken sections let one batch span many — so a
readdir now re-validates every stale section from its start name to
the end of the directory instead of the next two.

* mount: fence tombstoned names with floors and gate the reconcile

A tombstone answered for its name before the floors were consulted, so
one at an old position let through events the newer listing floor
should have fenced; a build never hit this because it prunes
superseded tombstones, which a section refresh does not. The version
gate now raises a tombstone to the floors like any other record.

With no per-entry versions, only the section floor fences a
reconcile's work, so a range the rebuilt or re-split table no longer
has must not touch the store either: the range check moves ahead of
the mutations, under the same lock the floor install holds.

An unversioned refresh no longer retries: the section is remembered as
unverifiable and skipped by the stale scan, or every batch of every
readdir would re-list the same ranges against a filer that cannot
vouch for them.

* mount: clear beaten unversioned markers and skip refresh mid-build

An unversioned marker outliving the snapshot write that replaced its
content bypassed the section floor the same way an old tombstone did,
letting a delayed pre-snapshot event roll the entry back. The refresh
now clears the marker when its write wins; pinned local-only entries
are not replaced at all, keeping their content and marker.

A rebuild wipes and repopulates the store off the apply loop, so a
refresh reconciling meanwhile could sweep children the build had
already inserted and let it publish the directory incomplete. The
refresh now skips a building directory, as events (buffered) and
purges (skipped) already do; its staleness dies with the build's
fresh table.

* mount: clear the unversioned marker only after its replacement lands

Clearing before the insert meant a failed write left the old local
content claiming the listing floors, fencing the very events that were
still entitled to correct it.

* meta_cache: rename the section state machine to sectionList

dirSections named both the type and the map of them.

* mount: raise the default cacheDirMaxEntries to 100000

The low ceiling guarded against whole-listing rebuild churn: a big
cached directory under writes kept re-streaming everything. Sectioned
invalidation ended that — a burst now costs one range listing — so the
remaining cost of caching a large directory is its one-time build,
comparable to the single direct listing that read-through mode pays on
every enumeration instead.

* meta_cache: cover section border and edge cases

A bound-named entry belongs to the section starting at the bound: the
neighboring refresh's sweep stops before it, its own section's covers
it. Churn past everything the build saw lands in the tail section, a
rename spanning two sections invalidates both, and a listed entry at
the section's end name is cut off with the ones beyond it.

weed/command/mount.go

@@ -115,7 +115,7 @@ func init() {
 	mountOptions.cacheDirForWrite = cmdMount.Flag.String("cacheDirWrite", "", "buffer writes mostly for large files")
 	mountOptions.writeBufferSizeMB = cmdMount.Flag.Int64("writeBufferSizeMB", 0, "global cap on the per-mount write buffer (memory + swap) in MB, 0 means unlimited. Bounds /tmp growth when volume uploads stall")
 	mountOptions.cacheMetaTtlSec = cmdMount.Flag.Int("cacheMetaTtlSec", 60, "metadata cache validity seconds")
-	mountOptions.cacheDirMaxEntries = cmdMount.Flag.Int("cacheDirMaxEntries", 10000, "a directory with more children than this is not cached locally but read directly from the filer; 0 caches everything")
+	mountOptions.cacheDirMaxEntries = cmdMount.Flag.Int("cacheDirMaxEntries", 100000, "a directory with more children than this is not cached locally but read directly from the filer; 0 caches everything")
 	mountOptions.dataCenter = cmdMount.Flag.String("dataCenter", "", "prefer to write to the data center")
 	mountOptions.allowOthers = cmdMount.Flag.Bool("allowOthers", true, "allows other users to access the file system")
 	mountOptions.defaultPermissions = cmdMount.Flag.Bool("defaultPermissions", true, "enforce permissions by the operating system")

weed/mount/inode_to_path.go

@@ -34,8 +34,6 @@ type dirState struct {
 	cachedExpiresTime time.Time
 	lastAccess        time.Time
 	lastRefresh       time.Time
-	updateWindowStart time.Time
-	updateCount       int
 	subdirCount       int32 // tracked in-memory for POSIX directory nlink
 }
 
@@ -43,8 +41,6 @@ func (d *dirState) resetCacheState() {
 	d.isChildrenCached = false
 	d.readDirDirect = false
 	d.cachedExpiresTime = time.Time{}
-	d.updateCount = 0
-	d.updateWindowStart = time.Time{}
 }
 
 func (ie *InodeEntry) removeOnePath(p util.FullPath) bool {
@@ -245,8 +241,6 @@ func (i *InodeToPath) MarkChildrenCached(fullpath util.FullPath) {
 	now := time.Now()
 	d.lastAccess = now
 	d.lastRefresh = now
-	d.updateCount = 0
-	d.updateWindowStart = time.Time{}
 	if i.cacheMetaTtlSec > 0 {
 		d.cachedExpiresTime = now.Add(i.cacheMetaTtlSec)
 	}
@@ -376,43 +370,9 @@ func (i *InodeToPath) MarkDirectoryReadThrough(fullpath util.FullPath, now time.
 	d.cachedExpiresTime = time.Time{}
 	d.lastAccess = now
 	d.lastRefresh = time.Time{}
-	d.updateCount = 0
-	d.updateWindowStart = time.Time{}
 	return true
 }
 
-func (i *InodeToPath) RecordDirectoryUpdate(fullpath util.FullPath, now time.Time, window time.Duration, threshold int) bool {
-	if threshold <= 0 || window <= 0 {
-		return false
-	}
-	i.Lock()
-	defer i.Unlock()
-	inode, found := i.path2inode[fullpath]
-	if !found {
-		return false
-	}
-	d := i.dirStates[inode]
-	if d == nil || !d.isChildrenCached {
-		return false
-	}
-	if d.updateWindowStart.IsZero() || now.Sub(d.updateWindowStart) > window {
-		d.updateWindowStart = now
-		d.updateCount = 0
-	}
-	d.updateCount++
-	if d.updateCount >= threshold {
-		d.isChildrenCached = false
-		d.readDirDirect = true
-		d.cachedExpiresTime = time.Time{}
-		d.lastAccess = now
-		d.lastRefresh = time.Time{}
-		d.updateCount = 0
-		d.updateWindowStart = time.Time{}
-		return true
-	}
-	return false
-}
-
 func (i *InodeToPath) ShouldReadDirectoryDirect(fullpath util.FullPath) bool {
 	i.RLock()
 	defer i.RUnlock()
@@ -441,8 +401,6 @@ func (i *InodeToPath) MarkDirectoryRefreshed(fullpath util.FullPath, now time.Ti
 	d.lastRefresh = now
 	d.lastAccess = now
 	d.readDirDirect = false
-	d.updateCount = 0
-	d.updateWindowStart = time.Time{}
 	if i.cacheMetaTtlSec > 0 {
 		d.cachedExpiresTime = now.Add(i.cacheMetaTtlSec)
 	}

weed/mount/inode_to_path_test.go

@@ -126,26 +126,6 @@ func TestOnlyDirectoriesGetDirState(t *testing.T) {
 	}
 }
 
-func TestRecordDirectoryUpdateSwitchesDirectoryToReadThrough(t *testing.T) {
-	root := util.FullPath("/")
-	dir := util.FullPath("/data")
-
-	inodeToPath := NewInodeToPath(root, 60)
-	inodeToPath.Lookup(dir, time.Now().Unix(), true, false, 0, true)
-	inodeToPath.MarkChildrenCached(dir)
-
-	now := time.Now()
-	if !inodeToPath.RecordDirectoryUpdate(dir, now, time.Second, 1) {
-		t.Fatal("expected directory to switch to read-through mode")
-	}
-	if inodeToPath.IsChildrenCached(dir) {
-		t.Fatal("directory should no longer be marked cached")
-	}
-	if !inodeToPath.ShouldReadDirectoryDirect(dir) {
-		t.Fatal("directory should be served via direct reads after hot invalidation")
-	}
-}
-
 func TestMarkChildrenCachedClearsReadThroughMode(t *testing.T) {
 	root := util.FullPath("/")
 	dir := util.FullPath("/data")

weed/mount/meta_cache/meta_cache.go

@@ -54,6 +54,11 @@ type MetaCache struct {
 	// instead of a record per child.
 	dirVersionFloors map[util.FullPath]int64
 
+	// dirSections is each cached directory's listing split into name-range
+	// sections, so a churn burst invalidates one section instead of the
+	// whole listing. See meta_cache_sections.go.
+	dirSections map[util.FullPath]*sectionList
+
 	// Entry invalidations run on a worker, not inline on the apply loop:
 	// invalidateFunc takes the fh lock, which a flush can hold while waiting on
 	// the apply loop (flushMetadataToFiler -> applyLocalMetadataEvent), so inline
@@ -92,6 +97,7 @@ const (
 	metadataCompleteBuild
 	metadataAbortBuild
 	metadataPurgeDir
+	metadataSectionRefresh
 	metadataShutdown
 )
 
@@ -102,6 +108,8 @@ type metadataApplyRequest struct {
 	options      MetadataResponseApplyOptions
 	buildPath    util.FullPath
 	snapshotTsNs int64
+	sections     *sectionList    // the completed build's section table
+	refresh      *sectionRefresh // one section re-listing to reconcile
 	resetFn      func()
 	done         chan error
 }
@@ -124,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),
+		dirSections:          make(map[util.FullPath]*sectionList),
 		oversizedDirs:        make(map[util.FullPath]struct{}),
 	}
 	mc.invalidateWorker = util.NewAsyncBatchWorker(func(batch []EntryInvalidation) {
@@ -316,11 +325,12 @@ func (mc *MetaCache) BeginDirectoryBuild(ctx context.Context, dirPath util.FullP
 	})
 }
 
-func (mc *MetaCache) CompleteDirectoryBuild(ctx context.Context, dirPath util.FullPath, snapshotTsNs int64) error {
+func (mc *MetaCache) CompleteDirectoryBuild(ctx context.Context, dirPath util.FullPath, snapshotTsNs int64, sectionBounds []string) error {
 	return mc.enqueueAndWait(ctx, metadataApplyRequest{
 		kind:         metadataCompleteBuild,
 		buildPath:    dirPath,
 		snapshotTsNs: snapshotTsNs,
+		sections:     newSectionTable(sectionBounds),
 	})
 }
 
@@ -488,22 +498,19 @@ func (mc *MetaCache) entryVersionRecordLocked(ctx context.Context, fp util.FullP
 }
 
 // entryVersionBlocksLocked reports whether a write at tsNs is already
-// reflected at fp. A tombstone fences with no entry present. Otherwise the
-// path's version is its own record or, lacking one, its directory's listing
-// floor — which covers children the listing saw present and absent alike.
-// A plain record only counts while its entry exists: records linger after a
+// reflected at fp. The path's version is its own record or, lacking one, the
+// listing floors — which cover names a listing saw present and absent alike.
+// A tombstone fences with no entry present, and a later floor outranks it; a
+// plain record only counts while its entry exists: records linger after a
 // bulk folder wipe and must not fence a recreate.
 func (mc *MetaCache) entryVersionBlocksLocked(ctx context.Context, fp util.FullPath, tsNs int64) bool {
 	recordTsNs, tombstone, unversioned := mc.entryVersionRecordLocked(ctx, fp)
-	if tombstone {
-		return recordTsNs >= tsNs
-	}
 	if unversioned {
 		// Local content no log position describes: fence nothing, so any
 		// event can correct it.
 		return false
 	}
-	if !mc.entryExistsLocked(ctx, fp) {
+	if !tombstone && !mc.entryExistsLocked(ctx, fp) {
 		recordTsNs = 0
 	}
 	return mc.entryVersionFloorLocked(fp, recordTsNs) >= tsNs
@@ -520,12 +527,18 @@ func (mc *MetaCache) entryExistsLocked(ctx context.Context, fp util.FullPath) bo
 }
 
 // entryVersionFloorLocked raises a path's own version record to its
-// directory's listing floor: the listing covered every child at its snapshot,
-// so a child without a later record of its own is versioned at the snapshot.
+// directory's listing floor or its section's refresh floor: each listing
+// covered every name in its range at its snapshot, so a name without a later
+// record of its own is versioned at the snapshot.
 func (mc *MetaCache) entryVersionFloorLocked(fp util.FullPath, recordTsNs int64) int64 {
-	dir, _ := fp.DirAndName()
+	dir, name := fp.DirAndName()
 	if floor := mc.dirVersionFloors[util.FullPath(dir)]; floor > recordTsNs {
-		return floor
+		recordTsNs = floor
+	}
+	if sl := mc.dirSections[util.FullPath(dir)]; sl != nil {
+		if floor := sl.floorOf(name); floor > recordTsNs {
+			recordTsNs = floor
+		}
 	}
 	return recordTsNs
 }
@@ -592,6 +605,7 @@ func (mc *MetaCache) DeleteFolderChildren(ctx context.Context, fp util.FullPath)
 	mc.Lock()
 	defer mc.Unlock()
 	delete(mc.dirVersionFloors, fp)
+	delete(mc.dirSections, fp)
 	mc.deleteChildVersionRecordsLocked(ctx, fp)
 	return mc.localStore.DeleteFolderChildren(ctx, fp)
 }
@@ -790,11 +804,13 @@ func (mc *MetaCache) handleApplyRequest(req metadataApplyRequest) error {
 	case metadataBeginBuild:
 		return mc.beginDirectoryBuildNow(req.buildPath)
 	case metadataCompleteBuild:
-		return mc.completeDirectoryBuildNow(req.ctx, req.buildPath, req.snapshotTsNs)
+		return mc.completeDirectoryBuildNow(req.ctx, req.buildPath, req.snapshotTsNs, req.sections)
 	case metadataAbortBuild:
 		return mc.abortDirectoryBuildNow(req.buildPath)
 	case metadataPurgeDir:
 		return mc.purgeDirectoryChildrenNow(req.ctx, req.buildPath, req.resetFn)
+	case metadataSectionRefresh:
+		return mc.applySectionRefreshNow(req.ctx, req.buildPath, req.refresh)
 	case metadataShutdown:
 		return nil
 	default:
@@ -942,6 +958,17 @@ func (mc *MetaCache) applyMetadataResponseLocked(ctx context.Context, resp *file
 			newEntry = nil
 		}
 	}
+	// Only foreign churn counts toward section invalidation: this mount's own
+	// writes are ground truth for its cache.
+	if options.InvalidateEntries {
+		now := time.Now()
+		if oldPath != "" {
+			mc.noteSectionChangeLocked(oldPath, now)
+		}
+		if newEntry != nil && newEntry.FullPath != oldPath {
+			mc.noteSectionChangeLocked(newEntry.FullPath, now)
+		}
+	}
 	err := mc.atomicUpdateEntryFromFilerLocked(ctx, oldPath, newEntry, allowUncachedInsert, resp.TsNs)
 	if err == nil && hideNewPath {
 		if purgeErr := mc.purgeEntryLocked(ctx, newPath, message.NewEntry.IsDirectory); purgeErr != nil {
@@ -954,6 +981,8 @@ 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 {
+			delete(mc.dirVersionFloors, oldPath)
+			delete(mc.dirSections, oldPath)
 			if deleteErr := mc.localStore.DeleteFolderChildren(ctx, oldPath); deleteErr != nil {
 				glog.V(2).Infof("delete descendants of %s: %v", oldPath, deleteErr)
 			}
@@ -995,11 +1024,12 @@ func (mc *MetaCache) purgeDirectoryChildrenNow(ctx context.Context, dirPath util
 	mc.Lock()
 	defer mc.Unlock()
 	delete(mc.dirVersionFloors, dirPath)
+	delete(mc.dirSections, dirPath)
 	mc.deleteChildVersionRecordsLocked(ctx, dirPath)
 	return mc.localStore.DeleteFolderChildren(ctx, dirPath)
 }
 
-func (mc *MetaCache) completeDirectoryBuildNow(ctx context.Context, dirPath util.FullPath, snapshotTsNs int64) error {
+func (mc *MetaCache) completeDirectoryBuildNow(ctx context.Context, dirPath util.FullPath, snapshotTsNs int64, sections *sectionList) error {
 	state := mc.buildingDirs[dirPath]
 	delete(mc.buildingDirs, dirPath)
 
@@ -1012,6 +1042,7 @@ func (mc *MetaCache) completeDirectoryBuildNow(ctx context.Context, dirPath util
 	// touches it. An unversioned listing (pre-upgrade filer) instead clears the
 	// children's records, or a re-inserted entry would inherit a stale one.
 	mc.Lock()
+	mc.dirSections[dirPath] = sections
 	if snapshotTsNs != 0 {
 		mc.dirVersionFloors[dirPath] = snapshotTsNs
 		mc.pruneSupersededTombstonesLocked(ctx, dirPath, snapshotTsNs)

weed/mount/meta_cache/meta_cache_apply_test.go

@@ -621,7 +621,7 @@ func TestUnversionedRebuildClearsStaleVersions(t *testing.T) {
 	}}); err != nil {
 		t.Fatalf("batch insert: %v", err)
 	}
-	if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 0); err != nil {
+	if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 0, nil); err != nil {
 		t.Fatalf("complete unversioned build: %v", err)
 	}
 
@@ -717,7 +717,7 @@ func TestBuildCompletionPrunesSupersededTombstones(t *testing.T) {
 	if err := mc.BeginDirectoryBuild(context.Background(), util.FullPath("/dir")); err != nil {
 		t.Fatalf("begin build: %v", err)
 	}
-	if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 3000); err != nil {
+	if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 3000, nil); err != nil {
 		t.Fatalf("complete build: %v", err)
 	}
 
@@ -806,7 +806,7 @@ func TestBuildFloorVersionsChildrenWithoutPerChildRecords(t *testing.T) {
 	}}); err != nil {
 		t.Fatalf("batch insert: %v", err)
 	}
-	if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 2000); err != nil {
+	if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 2000, nil); err != nil {
 		t.Fatalf("complete build: %v", err)
 	}
 
@@ -889,7 +889,7 @@ func TestUnversionedWriteDoesNotInheritDirectoryFloor(t *testing.T) {
 	if err := mc.BeginDirectoryBuild(context.Background(), util.FullPath("/dir")); err != nil {
 		t.Fatalf("begin build: %v", err)
 	}
-	if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 2000); err != nil {
+	if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 2000, nil); err != nil {
 		t.Fatalf("complete build: %v", err)
 	}
 

weed/mount/meta_cache/meta_cache_build_test.go

@@ -208,7 +208,7 @@ func TestDirectoryNotificationsSuppressedDuringBuild(t *testing.T) {
 	}
 
 	// Complete the build — buffered events should be replayed
-	if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 150); err != nil {
+	if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 150, nil); err != nil {
 		t.Fatalf("complete build: %v", err)
 	}
 
@@ -274,7 +274,7 @@ func TestEmptyDirectoryBuildReplaysAllBufferedEvents(t *testing.T) {
 	}
 
 	// Complete with snapshotTsNs=0 — simulates empty directory listing
-	if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/empty"), 0); err != nil {
+	if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/empty"), 0, nil); err != nil {
 		t.Fatalf("complete build: %v", err)
 	}
 
@@ -350,7 +350,7 @@ func TestBuildCompletionSurvivesCallerCancellation(t *testing.T) {
 	// ctx.Done() first, but the operation itself still completes in the
 	// apply loop. Poll for the observable side effect instead of using
 	// a fixed sleep.
-	_ = mc.CompleteDirectoryBuild(cancelledCtx, util.FullPath("/dir"), 100)
+	_ = mc.CompleteDirectoryBuild(cancelledCtx, util.FullPath("/dir"), 100, nil)
 
 	// Poll until the build completes or a deadline elapses.
 	deadline := time.After(2 * time.Second)
@@ -445,7 +445,7 @@ func TestBufferedRenameUpdatesOtherDirectoryBeforeBuildCompletes(t *testing.T) {
 		t.Fatalf("new path should stay hidden until build completes: %+v", newEntry)
 	}
 
-	if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/dst"), 100); err != nil {
+	if err := mc.CompleteDirectoryBuild(context.Background(), util.FullPath("/dst"), 100, nil); err != nil {
 		t.Fatalf("complete build: %v", err)
 	}
 

weed/mount/meta_cache/meta_cache_init.go

@@ -132,9 +132,10 @@ func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerCl
 		}()
 
 		// reloadFromFiler wipes the cached children and reloads them from the filer.
-		reloadFromFiler := func() (entryCount int, snapshotTsNs int64, err error) {
+		reloadFromFiler := func() (entryCount int, snapshotTsNs int64, sections sectionBoundsCollector, err error) {
 			err = util.Retry("ReadDirAllEntries", func() error {
 				entryCount = 0
+				sections = sectionBoundsCollector{}
 				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)
@@ -149,6 +150,7 @@ func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerCl
 					if maxCacheableEntries > 0 && entryCount >= maxCacheableEntries {
 						return &DirectoryTooLargeError{Path: path}
 					}
+					sections.note(entry.Name())
 					batch = append(batch, entry)
 					entryCount++
 
@@ -171,10 +173,10 @@ func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerCl
 				}
 				return nil
 			})
-			return entryCount, snapshotTsNs, err
+			return entryCount, snapshotTsNs, sections, err
 		}
 
-		entryCount, snapshotTsNs, fetchErr := reloadFromFiler()
+		entryCount, snapshotTsNs, sections, fetchErr := reloadFromFiler()
 		if fetchErr != nil {
 			var tooLarge *DirectoryTooLargeError
 			if errors.As(fetchErr, &tooLarge) {
@@ -204,7 +206,7 @@ func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerCl
 					return nil, ctx.Err()
 				}
 			}
-			if entryCount, snapshotTsNs, fetchErr = reloadFromFiler(); fetchErr != nil {
+			if entryCount, snapshotTsNs, sections, fetchErr = reloadFromFiler(); fetchErr != nil {
 				cleanupBuild("failed")
 				return nil, fmt.Errorf("confirm empty list %s: %w", path, fetchErr)
 			}
@@ -213,7 +215,7 @@ func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerCl
 			}
 		}
 
-		if err := mc.CompleteDirectoryBuild(context.Background(), path, snapshotTsNs); err != nil {
+		if err := mc.CompleteDirectoryBuild(context.Background(), path, snapshotTsNs, sections.bounds); err != nil {
 			cleanupBuild("unreplayed")
 			return nil, fmt.Errorf("complete build for %s: %w", path, err)
 		}

weed/mount/meta_cache/meta_cache_purge_test.go

@@ -58,7 +58,7 @@ func TestPurgeSkippedWhileDirectoryBuilding(t *testing.T) {
 	var resetCalls int32
 	mc.PurgeDirectoryChildren(dir, func() { atomic.AddInt32(&resetCalls, 1) })
 
-	if err := mc.CompleteDirectoryBuild(context.Background(), dir, 0); err != nil {
+	if err := mc.CompleteDirectoryBuild(context.Background(), dir, 0, nil); err != nil {
 		t.Fatalf("complete build: %v", err)
 	}
 

weed/mount/meta_cache/meta_cache_sections.go

@@ -0,0 +1,422 @@
+package meta_cache
+
+import (
+	"context"
+	"errors"
+	"math"
+	"sort"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// A cached directory's listing is split into contiguous name-range sections so
+// a burst of remote changes invalidates one section, not the whole listing.
+// Events keep applying to a stale section; staleness only means the next
+// listing re-validates that range against the filer before serving it, and
+// lookups in it read through until then.
+const (
+	// dirSectionSize is the target entries per section, fixed when a listing
+	// is built and re-derived when a re-listed section has outgrown it.
+	dirSectionSize = 1024
+	// sectionHotThreshold remote changes within sectionHotWindow invalidate
+	// the section they land in.
+	sectionHotThreshold = 64
+	sectionHotWindow    = 2 * time.Second
+	// sectionRefreshTimeout bounds how long a readdir waits on re-validating
+	// a section before serving the maintained-but-unverified cache instead.
+	sectionRefreshTimeout = 5 * time.Second
+	// sectionRefreshMaxEntries is the most one refresh will carry; a section
+	// grown past it is cheaper to re-tile with a full directory rebuild.
+	sectionRefreshMaxEntries = 4 * dirSectionSize
+)
+
+// ErrRefreshRangeTooLarge reports a section that outgrew one refresh; the
+// caller should drop the directory cache so a full rebuild re-tiles it.
+var ErrRefreshRangeTooLarge = errors.New("section outgrew one refresh")
+
+// sectionList: bounds[i] is the first name of section i+1; section 0 starts at
+// the beginning of the namespace, the last section runs to the end. It is a
+// plain state machine — no locking, no store; MetaCache drives it under its
+// own mutex.
+type sectionList struct {
+	bounds   []string
+	sections []sectionState
+}
+
+type sectionState struct {
+	stale       bool
+	updateCount int
+	windowStart time.Time
+	// unverifiable marks a stale section whose filer stamps no listing
+	// snapshots: re-listing it can never vouch for it, so stop trying and
+	// leave its lookups reading through.
+	unverifiable bool
+	// floorTsNs is the section's own listing snapshot: a refresh at it covered
+	// every name in the range, present or absent, so it fences like the
+	// directory floor but for this range alone.
+	floorTsNs int64
+}
+
+func newSectionTable(bounds []string) *sectionList {
+	return &sectionList{bounds: bounds, sections: make([]sectionState, len(bounds)+1)}
+}
+
+// sectionBoundsCollector derives section boundaries from an ordered listing:
+// every dirSectionSize-th name starts a new section.
+type sectionBoundsCollector struct {
+	count  int
+	bounds []string
+}
+
+func (c *sectionBoundsCollector) note(name string) {
+	if c.count > 0 && c.count%dirSectionSize == 0 {
+		c.bounds = append(c.bounds, name)
+	}
+	c.count++
+}
+
+// sectionRefresh carries one section's re-listing to the apply loop.
+type sectionRefresh struct {
+	lo, hi       string
+	entries      []*filer.Entry
+	snapshotTsNs int64
+}
+
+func (sl *sectionList) sectionOf(name string) int {
+	idx := sort.SearchStrings(sl.bounds, name)
+	if idx < len(sl.bounds) && sl.bounds[idx] == name {
+		idx++
+	}
+	return idx
+}
+
+// sectionRange returns the [lo, hi) name range of section idx; "" is unbounded.
+func (sl *sectionList) sectionRange(idx int) (lo, hi string) {
+	if idx > 0 {
+		lo = sl.bounds[idx-1]
+	}
+	if idx < len(sl.bounds) {
+		hi = sl.bounds[idx]
+	}
+	return
+}
+
+// noteChange counts one change against the section it lands in, marking the
+// section stale when a burst crosses the threshold.
+func (sl *sectionList) noteChange(name string, now time.Time) {
+	s := &sl.sections[sl.sectionOf(name)]
+	if s.stale {
+		return
+	}
+	if s.windowStart.IsZero() || now.Sub(s.windowStart) > sectionHotWindow {
+		s.windowStart = now
+		s.updateCount = 0
+	}
+	s.updateCount++
+	if s.updateCount >= sectionHotThreshold {
+		s.stale = true
+	}
+}
+
+func (sl *sectionList) isFresh(name string) bool {
+	return !sl.sections[sl.sectionOf(name)].stale
+}
+
+// floorOf returns the refresh snapshot covering this name, or zero when its
+// section has never been re-listed. A stale section keeps fencing: what its
+// last listing established stays established.
+func (sl *sectionList) floorOf(name string) int64 {
+	return sl.sections[sl.sectionOf(name)].floorTsNs
+}
+
+type nameRange struct {
+	lo, hi string
+}
+
+// staleRangesAhead returns the invalidated ranges worth re-listing from the
+// section holding startName to the end of the directory.
+func (sl *sectionList) staleRangesAhead(startName string) (ranges []nameRange) {
+	for i := sl.sectionOf(startName); i < len(sl.sections); i++ {
+		if sl.sections[i].stale && !sl.sections[i].unverifiable {
+			lo, hi := sl.sectionRange(i)
+			ranges = append(ranges, nameRange{lo: lo, hi: hi})
+		}
+	}
+	return
+}
+
+// hasRange reports whether the table still has a section covering exactly
+// [lo, hi); a rebuild or re-split since a listing was taken retires the range
+// it described.
+func (sl *sectionList) hasRange(lo, hi string) bool {
+	curLo, curHi := sl.sectionRange(sl.sectionOf(lo))
+	return curLo == lo && curHi == hi
+}
+
+// completeRefresh marks the section covering exactly [lo, hi) fresh after a
+// re-listing that fetched names at snapshotTsNs, re-splitting a section that
+// outgrew twice its target size. The snapshot becomes the section's floor. A
+// range the table no longer has is ignored — splicing bounds from a stale
+// range could leave the table unsorted — and an unversioned listing vouches
+// for nothing: the section stays stale, remembered as not worth re-listing.
+func (sl *sectionList) completeRefresh(lo, hi string, names []string, snapshotTsNs int64) bool {
+	idx := sl.sectionOf(lo)
+	if curLo, curHi := sl.sectionRange(idx); curLo != lo || curHi != hi {
+		return false
+	}
+	if snapshotTsNs == 0 {
+		sl.sections[idx].unverifiable = true
+		return false
+	}
+	if len(names) > 2*dirSectionSize {
+		var newBounds []string
+		for i := dirSectionSize; i < len(names); i += dirSectionSize {
+			newBounds = append(newBounds, names[i])
+		}
+		bounds := make([]string, 0, len(sl.bounds)+len(newBounds))
+		bounds = append(bounds, sl.bounds[:idx]...)
+		bounds = append(bounds, newBounds...)
+		bounds = append(bounds, sl.bounds[idx:]...)
+		sections := make([]sectionState, 0, len(bounds)+1)
+		sections = append(sections, sl.sections[:idx]...)
+		for i := 0; i <= len(newBounds); i++ {
+			sections = append(sections, sectionState{floorTsNs: snapshotTsNs})
+		}
+		sections = append(sections, sl.sections[idx+1:]...)
+		sl.bounds, sl.sections = bounds, sections
+	} else {
+		sl.sections[idx] = sectionState{floorTsNs: snapshotTsNs}
+	}
+	return true
+}
+
+// noteSectionChangeLocked counts one remote change against the section of the
+// directory it lands in.
+func (mc *MetaCache) noteSectionChangeLocked(fp util.FullPath, now time.Time) {
+	dir, name := fp.DirAndName()
+	if sl := mc.dirSections[util.FullPath(dir)]; sl != nil {
+		sl.noteChange(name, now)
+	}
+}
+
+// IsNameFresh reports whether the cached listing still vouches for this name.
+// A directory without section state vouches for all of it.
+func (mc *MetaCache) IsNameFresh(fp util.FullPath) bool {
+	dir, name := fp.DirAndName()
+	mc.RLock()
+	defer mc.RUnlock()
+	sl := mc.dirSections[util.FullPath(dir)]
+	return sl == nil || sl.isFresh(name)
+}
+
+func (mc *MetaCache) staleRangesAhead(dirPath util.FullPath, startName string) []nameRange {
+	mc.RLock()
+	defer mc.RUnlock()
+	sl := mc.dirSections[dirPath]
+	if sl == nil {
+		return nil
+	}
+	return sl.staleRangesAhead(startName)
+}
+
+func (mc *MetaCache) rangeStale(dirPath util.FullPath, lo string) bool {
+	mc.RLock()
+	defer mc.RUnlock()
+	sl := mc.dirSections[dirPath]
+	return sl != nil && !sl.isFresh(lo)
+}
+
+// EnsureListingFresh re-validates every invalidated section from startName to
+// the end of the directory before a listing pages through them. A listing's
+// reach is unknowable up front — a resumed handle can skip far ahead, and
+// shrunken sections let one batch span many — so all of them are covered.
+func EnsureListingFresh(ctx context.Context, mc *MetaCache, client filer_pb.FilerClient, dirPath util.FullPath, startName string) error {
+	ranges := mc.staleRangesAhead(dirPath, startName)
+	if len(ranges) == 0 {
+		return nil
+	}
+	ctx, cancel := context.WithTimeout(ctx, sectionRefreshTimeout)
+	defer cancel()
+	for _, r := range ranges {
+		if err := mc.refreshSection(ctx, client, dirPath, r.lo, r.hi); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (mc *MetaCache) refreshSection(ctx context.Context, client filer_pb.FilerClient, dirPath util.FullPath, lo, hi string) error {
+	_, err, _ := mc.visitGroup.Do(string(dirPath)+"\x00section\x00"+lo, func() (interface{}, error) {
+		if !mc.rangeStale(dirPath, lo) {
+			return nil, nil
+		}
+		entries, snapshotTsNs, err := mc.listFilerRange(ctx, client, dirPath, lo, hi)
+		if err != nil {
+			return nil, err
+		}
+		return nil, mc.enqueueAndWait(ctx, metadataApplyRequest{
+			kind:      metadataSectionRefresh,
+			buildPath: dirPath,
+			refresh:   &sectionRefresh{lo: lo, hi: hi, entries: entries, snapshotTsNs: snapshotTsNs},
+		})
+	})
+	return err
+}
+
+// listFilerRange reads [lo, hi) from the filer at one snapshot, paging by
+// section-sized batches.
+func (mc *MetaCache) listFilerRange(ctx context.Context, client filer_pb.FilerClient, dirPath util.FullPath, lo, hi string) (entries []*filer.Entry, snapshotTsNs int64, err error) {
+	startFrom, includeStart := lo, lo != ""
+	for {
+		var page []*filer.Entry
+		var pageCount int
+		var last string
+		done := false
+		err = client.WithFilerClient(false, func(sc filer_pb.SeaweedFilerClient) error {
+			// reset in case a failover retry re-runs a partly streamed page
+			page, pageCount, last, done = nil, 0, "", false
+			ts, listErr := filer_pb.DoSeaweedListWithSnapshot(ctx, sc, dirPath, "", func(pbEntry *filer_pb.Entry, isLast bool) error {
+				pageCount++
+				last = pbEntry.Name
+				if hi != "" && pbEntry.Name >= hi {
+					done = true
+				}
+				if done {
+					return nil
+				}
+				if !mc.includeSystemEntries && IsHiddenSystemEntry(string(dirPath), pbEntry.Name) {
+					return nil
+				}
+				page = append(page, filer.FromPbEntry(string(dirPath), pbEntry))
+				return nil
+			}, startFrom, includeStart, dirSectionSize, snapshotTsNs)
+			if listErr != nil {
+				return listErr
+			}
+			if snapshotTsNs == 0 {
+				snapshotTsNs = ts
+			}
+			return nil
+		})
+		if err != nil {
+			return nil, 0, err
+		}
+		entries = append(entries, page...)
+		if len(entries) > sectionRefreshMaxEntries {
+			return nil, 0, ErrRefreshRangeTooLarge
+		}
+		if done || pageCount < dirSectionSize {
+			return entries, snapshotTsNs, nil
+		}
+		startFrom, includeStart = last, false
+	}
+}
+
+// applySectionRefreshNow reconciles one section against a filer listing of its
+// range, then marks it fresh. Runs on the apply loop; mutations go through the
+// version gate so the listing cannot roll back a newer applied event, and
+// pinned local-only entries (deferred creates not yet on the filer) survive.
+func (mc *MetaCache) applySectionRefreshNow(ctx context.Context, dirPath util.FullPath, r *sectionRefresh) error {
+	lo, hi, snapshotTsNs := r.lo, r.hi, r.snapshotTsNs
+
+	// A build wipes and repopulates the store off-loop; reconciling against
+	// it would sweep children the build already inserted and publish the
+	// directory incomplete. The staleness dies with the build's fresh table.
+	if mc.isBuildingDir(dirPath) {
+		return nil
+	}
+
+	mc.Lock()
+	defer mc.Unlock()
+
+	// With no per-entry versions, only the section floor fences this work; a
+	// range the rebuilt or re-split table no longer has gets no floor, so it
+	// must not touch the store either. The lock is held through the floor
+	// install below, so the check cannot go stale.
+	sl := mc.dirSections[dirPath]
+	if sl == nil || !sl.hasRange(lo, hi) {
+		return nil
+	}
+
+	fetchedNames := make([]string, 0, len(r.entries))
+	fetched := make(map[string]struct{}, len(r.entries))
+	for _, entry := range r.entries {
+		fetchedNames = append(fetchedNames, entry.Name())
+		fetched[entry.Name()] = struct{}{}
+		if snapshotTsNs == 0 {
+			// A pre-upgrade filer stamps no snapshot, leaving nothing to
+			// order against: only fill gaps, so a concurrently applied event
+			// can never be rolled back.
+			if mc.entryExistsLocked(ctx, entry.FullPath) {
+				continue
+			}
+			if _, tombstone := mc.getEntryVersionRecordLocked(ctx, entry.FullPath); tombstone {
+				continue
+			}
+			if err := mc.localStore.InsertEntry(ctx, entry); err != nil {
+				return err
+			}
+			mc.setEntryVersionLocked(ctx, entry.FullPath, 0)
+			continue
+		}
+		if mc.entryVersionBlocksLocked(ctx, entry.FullPath, snapshotTsNs) {
+			continue
+		}
+		// An unversioned marker would bypass the section floor, so it cannot
+		// outlive the snapshot write that replaces its content — but pinned
+		// local-only state stays authoritative and is not replaced at all.
+		_, _, unversioned := mc.entryVersionRecordLocked(ctx, entry.FullPath)
+		if unversioned {
+			if existing, findErr := mc.localStore.FindEntry(ctx, entry.FullPath); findErr == nil && existing != nil && mc.pinnedChildFn != nil && mc.pinnedChildFn(existing) {
+				continue
+			}
+		}
+		// no per-entry version: the section floor set below covers the range
+		if err := mc.localStore.InsertEntry(ctx, entry); err != nil {
+			return err
+		}
+		if unversioned {
+			// only once the write landed: if it fails, the old content keeps
+			// the marker, and with it the right to be corrected by any event
+			mc.clearEntryVersionLocked(ctx, entry.FullPath)
+		}
+	}
+
+	// Deletions need the snapshot as an ordering reference; without one a
+	// name created after the listing would be swept away.
+	if snapshotTsNs != 0 {
+		var vanished []*filer.Entry
+		if _, err := mc.localStore.ListDirectoryEntries(ctx, dirPath, lo, true, math.MaxInt64, func(entry *filer.Entry) (bool, error) {
+			if hi != "" && entry.Name() >= hi {
+				return false, nil
+			}
+			if _, found := fetched[entry.Name()]; !found {
+				vanished = append(vanished, entry)
+			}
+			return true, nil
+		}); err != nil {
+			return err
+		}
+		for _, entry := range vanished {
+			if mc.pinnedChildFn != nil && mc.pinnedChildFn(entry) {
+				continue
+			}
+			if mc.entryVersionBlocksLocked(ctx, entry.FullPath, snapshotTsNs) {
+				continue
+			}
+			if err := mc.localStore.DeleteEntry(ctx, entry.FullPath); err != nil {
+				return err
+			}
+			mc.clearEntryVersionLocked(ctx, entry.FullPath)
+		}
+	}
+
+	// The floor fences the whole range, absent names included; an unversioned
+	// listing sets none and the section stays stale, its lookups reading
+	// through, with no further re-listing attempts.
+	sl.completeRefresh(lo, hi, fetchedNames, snapshotTsNs)
+	return nil
+}

weed/mount/meta_cache/meta_cache_sections_test.go

@@ -0,0 +1,929 @@
+package meta_cache
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+func TestSectionOf(t *testing.T) {
+	sl := newSectionTable([]string{"g", "p"})
+	for name, want := range map[string]int{
+		"a": 0, "f": 0,
+		"g": 1, "h": 1, "o": 1,
+		"p": 2, "z": 2,
+	} {
+		if got := sl.sectionOf(name); got != want {
+			t.Errorf("sectionOf(%q) = %d, want %d", name, got, want)
+		}
+	}
+	if lo, hi := sl.sectionRange(0); lo != "" || hi != "g" {
+		t.Errorf("sectionRange(0) = %q..%q, want ..g", lo, hi)
+	}
+	if lo, hi := sl.sectionRange(1); lo != "g" || hi != "p" {
+		t.Errorf("sectionRange(1) = %q..%q, want g..p", lo, hi)
+	}
+	if lo, hi := sl.sectionRange(2); lo != "p" || hi != "" {
+		t.Errorf("sectionRange(2) = %q..%q, want p..", lo, hi)
+	}
+}
+
+func buildSectionedDir(t *testing.T, mc *MetaCache, dir util.FullPath, snapshotTsNs int64, bounds []string) {
+	t.Helper()
+	if err := mc.BeginDirectoryBuild(context.Background(), dir); err != nil {
+		t.Fatalf("begin build: %v", err)
+	}
+	if err := mc.CompleteDirectoryBuild(context.Background(), dir, snapshotTsNs, bounds); err != nil {
+		t.Fatalf("complete build: %v", err)
+	}
+}
+
+func applyChurn(t *testing.T, mc *MetaCache, dir, prefix string, n int, baseTsNs int64, options MetadataResponseApplyOptions) {
+	t.Helper()
+	for i := 0; i < n; i++ {
+		resp := &filer_pb.SubscribeMetadataResponse{
+			Directory: dir,
+			EventNotification: &filer_pb.EventNotification{
+				NewEntry: &filer_pb.Entry{
+					Name: fmt.Sprintf("%s%03d", prefix, i),
+					Attributes: &filer_pb.FuseAttributes{
+						Crtime:   1,
+						Mtime:    1,
+						FileMode: 0100644,
+						FileSize: 1,
+					},
+				},
+			},
+			TsNs: baseTsNs + int64(i),
+		}
+		if err := mc.ApplyMetadataResponse(context.Background(), resp, options); err != nil {
+			t.Fatalf("apply churn %d: %v", i, err)
+		}
+	}
+}
+
+func TestSectionBurstInvalidatesOnlyItsSection(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
+		"/":    true,
+		"/dir": true,
+	})
+	defer mc.Shutdown()
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, []string{"m"})
+
+	applyChurn(t, mc, "/dir", "a-", sectionHotThreshold, 2000, SubscriberMetadataResponseApplyOptions)
+
+	if mc.IsNameFresh(util.FullPath("/dir/a-000")) {
+		t.Fatal("burst section should be invalidated")
+	}
+	if !mc.IsNameFresh(util.FullPath("/dir/zzz")) {
+		t.Fatal("the other section must stay fresh")
+	}
+}
+
+func TestLocalChangesDoNotInvalidateSections(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
+		"/":    true,
+		"/dir": true,
+	})
+	defer mc.Shutdown()
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, nil)
+
+	applyChurn(t, mc, "/dir", "a-", 2*sectionHotThreshold, 2000, LocalMetadataResponseApplyOptions)
+
+	if !mc.IsNameFresh(util.FullPath("/dir/a-000")) {
+		t.Fatal("this mount's own writes must not invalidate its cache")
+	}
+}
+
+func TestSectionRefreshReconciles(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
+		"/":    true,
+		"/dir": true,
+	})
+	defer mc.Shutdown()
+	mc.SetPinnedChildFn(func(entry *filer.Entry) bool {
+		return entry.Name() == "b-pinned"
+	})
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, []string{"m"})
+
+	for _, name := range []string{"b-keep", "b-vanish", "b-pinned"} {
+		if err := mc.InsertEntry(context.Background(), &filer.Entry{
+			FullPath: util.FullPath("/dir/" + name),
+			Attr:     filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(1, 0), Mode: 0100644, FileSize: 1},
+		}, 0); err != nil {
+			t.Fatalf("seed %s: %v", name, err)
+		}
+	}
+	// applied ahead of the refresh snapshot; the listing must not roll it back
+	if err := mc.ApplyMetadataResponse(context.Background(), &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{
+				Name:       "b-newer",
+				Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 9, FileMode: 0100644, FileSize: 9},
+			},
+		},
+		TsNs: 9000,
+	}, SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply newer event: %v", err)
+	}
+
+	applyChurn(t, mc, "/dir", "a-", sectionHotThreshold, 2000, SubscriberMetadataResponseApplyOptions)
+	if mc.IsNameFresh(util.FullPath("/dir/b-keep")) {
+		t.Fatal("section should be stale before the refresh")
+	}
+
+	if err := mc.enqueueAndWait(context.Background(), metadataApplyRequest{
+		kind:      metadataSectionRefresh,
+		buildPath: util.FullPath("/dir"),
+		refresh: &sectionRefresh{
+			hi: "m",
+			entries: []*filer.Entry{
+				{
+					FullPath: util.FullPath("/dir/b-keep"),
+					Attr:     filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(2, 0), Mode: 0100644, FileSize: 7},
+				},
+				{
+					FullPath: util.FullPath("/dir/b-new"),
+					Attr:     filer.Attr{Crtime: time.Unix(2, 0), Mtime: time.Unix(2, 0), Mode: 0100644, FileSize: 3},
+				},
+			},
+			snapshotTsNs: 5000,
+		},
+	}); err != nil {
+		t.Fatalf("refresh: %v", err)
+	}
+
+	if !mc.IsNameFresh(util.FullPath("/dir/b-keep")) {
+		t.Fatal("section should be fresh after the refresh")
+	}
+	entry, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/b-keep"))
+	if err != nil || entry.FileSize != 7 {
+		t.Fatalf("b-keep = %+v, %v; want size 7", entry, err)
+	}
+	if entry, _, err = mc.FindEntry(context.Background(), util.FullPath("/dir/b-new")); err != nil || entry.FileSize != 3 {
+		t.Fatalf("b-new = %+v, %v; want size 3", entry, err)
+	}
+	if _, _, err = mc.FindEntry(context.Background(), util.FullPath("/dir/b-vanish")); err != filer_pb.ErrNotFound {
+		t.Fatalf("b-vanish error = %v, want not found", err)
+	}
+	if _, _, err = mc.FindEntry(context.Background(), util.FullPath("/dir/a-000")); err != filer_pb.ErrNotFound {
+		t.Fatalf("churned a-000 error = %v, want not found", err)
+	}
+	if entry, _, err = mc.FindEntry(context.Background(), util.FullPath("/dir/b-newer")); err != nil || entry.FileSize != 9 {
+		t.Fatalf("b-newer = %+v, %v; want size 9 kept", entry, err)
+	}
+	if entry, _, err = mc.FindEntry(context.Background(), util.FullPath("/dir/b-pinned")); err != nil || entry == nil {
+		t.Fatalf("pinned local-only entry must survive a refresh: %v", err)
+	}
+
+	// a redelivered create older than the snapshot must not resurrect b-vanish
+	if err := mc.ApplyMetadataResponse(context.Background(), &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{
+				Name:       "b-vanish",
+				Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 1, FileMode: 0100644, FileSize: 1},
+			},
+		},
+		TsNs: 4000,
+	}, SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply stale event: %v", err)
+	}
+	if _, _, err = mc.FindEntry(context.Background(), util.FullPath("/dir/b-vanish")); err != filer_pb.ErrNotFound {
+		t.Fatalf("b-vanish resurrected by a pre-snapshot event: %v", err)
+	}
+}
+
+func TestSectionRefreshSplitsOvergrownSection(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
+		"/":    true,
+		"/dir": true,
+	})
+	defer mc.Shutdown()
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, nil)
+
+	count := 2*dirSectionSize + 1
+	entries := make([]*filer.Entry, 0, count)
+	for i := 0; i < count; i++ {
+		entries = append(entries, &filer.Entry{
+			FullPath: util.FullPath(fmt.Sprintf("/dir/f-%05d", i)),
+			Attr:     filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(1, 0), Mode: 0100644, FileSize: 1},
+		})
+	}
+	if err := mc.enqueueAndWait(context.Background(), metadataApplyRequest{
+		kind:      metadataSectionRefresh,
+		buildPath: util.FullPath("/dir"),
+		refresh:   &sectionRefresh{entries: entries, snapshotTsNs: 5000},
+	}); err != nil {
+		t.Fatalf("refresh: %v", err)
+	}
+
+	mc.RLock()
+	sl := mc.dirSections[util.FullPath("/dir")]
+	bounds := append([]string(nil), sl.bounds...)
+	sections := len(sl.sections)
+	mc.RUnlock()
+	want := []string{fmt.Sprintf("f-%05d", dirSectionSize), fmt.Sprintf("f-%05d", 2*dirSectionSize)}
+	if len(bounds) != len(want) || bounds[0] != want[0] || bounds[1] != want[1] {
+		t.Fatalf("bounds = %v, want %v", bounds, want)
+	}
+	if sections != len(bounds)+1 {
+		t.Fatalf("sections = %d, want %d", sections, len(bounds)+1)
+	}
+}
+
+type sectionFilerServer struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	mu       sync.Mutex
+	names    []string // sorted; all under one directory
+	snapshot int64
+	requests []*filer_pb.ListEntriesRequest
+}
+
+func (s *sectionFilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream filer_pb.SeaweedFiler_ListEntriesServer) error {
+	s.mu.Lock()
+	s.requests = append(s.requests, req)
+	names := s.names
+	snapshot := s.snapshot
+	s.mu.Unlock()
+
+	sent := uint32(0)
+	first := true
+	for _, name := range names {
+		if name < req.StartFromFileName || (name == req.StartFromFileName && !req.InclusiveStartFrom) {
+			continue
+		}
+		resp := &filer_pb.ListEntriesResponse{Entry: &filer_pb.Entry{
+			Name:       name,
+			Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 1, FileMode: 0100644, FileSize: 1},
+		}}
+		if first {
+			resp.SnapshotTsNs = snapshot
+			first = false
+		}
+		if err := stream.Send(resp); err != nil {
+			return err
+		}
+		sent++
+		if req.Limit > 0 && sent >= req.Limit {
+			break
+		}
+	}
+	return nil
+}
+
+func (s *sectionFilerServer) listRequests() []*filer_pb.ListEntriesRequest {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return append([]*filer_pb.ListEntriesRequest(nil), s.requests...)
+}
+
+type sectionTestFilerClient struct {
+	addr string
+}
+
+func (c *sectionTestFilerClient) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
+	conn, err := grpc.NewClient(c.addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+	return fn(filer_pb.NewSeaweedFilerClient(conn))
+}
+
+func (c *sectionTestFilerClient) AdjustedUrl(location *filer_pb.Location) string { return location.Url }
+
+func (c *sectionTestFilerClient) GetDataCenter() string { return "" }
+
+func startSectionFilerServer(t *testing.T, s *sectionFilerServer) filer_pb.FilerClient {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	srv := grpc.NewServer()
+	filer_pb.RegisterSeaweedFilerServer(srv, s)
+	go srv.Serve(listener)
+	t.Cleanup(srv.Stop)
+	return &sectionTestFilerClient{addr: listener.Addr().String()}
+}
+
+func TestEnsureListingFreshRefreshesFromFiler(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
+		"/":    true,
+		"/dir": true,
+	})
+	defer mc.Shutdown()
+
+	server := &sectionFilerServer{snapshot: 5000}
+	for i := 0; i < 1500; i++ {
+		server.names = append(server.names, fmt.Sprintf("b-%04d", i))
+	}
+	// at and beyond the section's end, must not be applied
+	server.names = append(server.names, "m", "z-1")
+	client := startSectionFilerServer(t, server)
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, []string{"m"})
+	applyChurn(t, mc, "/dir", "a-", sectionHotThreshold, 2000, SubscriberMetadataResponseApplyOptions)
+	if mc.IsNameFresh(util.FullPath("/dir/a-000")) {
+		t.Fatal("section should be stale before the refresh")
+	}
+
+	if err := EnsureListingFresh(context.Background(), mc, client, util.FullPath("/dir"), ""); err != nil {
+		t.Fatalf("refresh: %v", err)
+	}
+
+	if !mc.IsNameFresh(util.FullPath("/dir/a-000")) {
+		t.Fatal("section should be fresh after the refresh")
+	}
+	for _, name := range []string{"b-0000", "b-1499"} {
+		if entry, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/"+name)); err != nil || entry == nil {
+			t.Fatalf("%s missing after refresh: %v", name, err)
+		}
+	}
+	if _, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/a-000")); err != filer_pb.ErrNotFound {
+		t.Fatalf("churned a-000 error = %v, want not found", err)
+	}
+	for _, name := range []string{"m", "z-1"} {
+		if _, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/"+name)); err != filer_pb.ErrNotFound {
+			t.Fatalf("%s at or beyond the section's end was applied: %v", name, err)
+		}
+	}
+
+	requests := server.listRequests()
+	if len(requests) < 2 {
+		t.Fatalf("multi-page refresh made %d requests, want at least 2", len(requests))
+	}
+	if requests[0].SnapshotTsNs != 0 || requests[1].SnapshotTsNs != 5000 {
+		t.Fatalf("snapshot not pinned across pages: %d then %d", requests[0].SnapshotTsNs, requests[1].SnapshotTsNs)
+	}
+
+	// a fresh section costs no filer calls
+	if err := EnsureListingFresh(context.Background(), mc, client, util.FullPath("/dir"), ""); err != nil {
+		t.Fatalf("second refresh: %v", err)
+	}
+	if got := len(server.listRequests()); got != len(requests) {
+		t.Fatalf("fresh section still listed the filer: %d requests, was %d", got, len(requests))
+	}
+}
+
+func TestEnsureListingFreshGivesUpOnOvergrownRange(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
+		"/":    true,
+		"/dir": true,
+	})
+	defer mc.Shutdown()
+
+	server := &sectionFilerServer{snapshot: 5000}
+	for i := 0; i <= sectionRefreshMaxEntries; i++ {
+		server.names = append(server.names, fmt.Sprintf("c-%05d", i))
+	}
+	client := startSectionFilerServer(t, server)
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, nil)
+	applyChurn(t, mc, "/dir", "x-", sectionHotThreshold, 2000, SubscriberMetadataResponseApplyOptions)
+
+	err := EnsureListingFresh(context.Background(), mc, client, util.FullPath("/dir"), "")
+	if !errors.Is(err, ErrRefreshRangeTooLarge) {
+		t.Fatalf("refresh error = %v, want ErrRefreshRangeTooLarge", err)
+	}
+	if mc.IsNameFresh(util.FullPath("/dir/x-000")) {
+		t.Fatal("an aborted refresh must leave the section stale")
+	}
+}
+
+func TestSectionNoteChangeWindowExpiry(t *testing.T) {
+	sl := newSectionTable(nil)
+	t0 := time.Unix(100, 0)
+	for i := 0; i < sectionHotThreshold-1; i++ {
+		sl.noteChange(fmt.Sprintf("f-%03d", i), t0)
+	}
+	if !sl.isFresh("f-000") {
+		t.Fatal("below the threshold the section must stay fresh")
+	}
+
+	// the burst never completed inside one window, so the count restarts
+	t1 := t0.Add(sectionHotWindow + time.Second)
+	for i := 0; i < sectionHotThreshold-1; i++ {
+		sl.noteChange(fmt.Sprintf("f-%03d", i), t1)
+	}
+	if !sl.isFresh("f-000") {
+		t.Fatal("an expired window must not carry its count forward")
+	}
+
+	sl.noteChange("f-999", t1)
+	if sl.isFresh("f-000") {
+		t.Fatal("a full burst within one window must invalidate the section")
+	}
+}
+
+func TestSectionCompleteRefreshGuardsChangedTable(t *testing.T) {
+	sl := newSectionTable([]string{"g", "p"})
+	sl.sections[1].stale = true
+
+	if sl.completeRefresh("g", "q", []string{"h"}, 5000) {
+		t.Fatal("a range the table no longer has must be ignored")
+	}
+	if sl.isFresh("h") {
+		t.Fatal("an ignored refresh must not mark anything fresh")
+	}
+	if !sl.completeRefresh("g", "p", []string{"h"}, 5000) {
+		t.Fatal("the matching range must be accepted")
+	}
+	if !sl.isFresh("h") {
+		t.Fatal("section should be fresh after the refresh")
+	}
+}
+
+func TestSectionCompleteRefreshSplitsMiddleSection(t *testing.T) {
+	sl := newSectionTable([]string{"g", "p"})
+	sl.sections[1].stale = true
+
+	names := make([]string, 0, 2*dirSectionSize+1)
+	for i := 0; i <= 2*dirSectionSize; i++ {
+		names = append(names, fmt.Sprintf("g-%05d", i))
+	}
+	if !sl.completeRefresh("g", "p", names, 5000) {
+		t.Fatal("refresh of the middle section must be accepted")
+	}
+
+	wantBounds := []string{"g", names[dirSectionSize], names[2*dirSectionSize], "p"}
+	if len(sl.bounds) != len(wantBounds) {
+		t.Fatalf("bounds = %v, want %v", sl.bounds, wantBounds)
+	}
+	for i, b := range wantBounds {
+		if sl.bounds[i] != b {
+			t.Fatalf("bounds = %v, want %v", sl.bounds, wantBounds)
+		}
+	}
+	if len(sl.sections) != len(sl.bounds)+1 {
+		t.Fatalf("sections = %d, want %d", len(sl.sections), len(sl.bounds)+1)
+	}
+	for _, name := range []string{"a", "g-00000", names[dirSectionSize], "z"} {
+		if !sl.isFresh(name) {
+			t.Fatalf("%q should be fresh after the split", name)
+		}
+	}
+	if got := sl.sectionOf(names[dirSectionSize+1]); got != 2 {
+		t.Fatalf("sectionOf(%q) = %d, want 2", names[dirSectionSize+1], got)
+	}
+}
+
+func TestSectionFloorFencesAbsentNames(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
+		"/":    true,
+		"/dir": true,
+	})
+	defer mc.Shutdown()
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, []string{"m"})
+	applyChurn(t, mc, "/dir", "a-", sectionHotThreshold, 2000, SubscriberMetadataResponseApplyOptions)
+
+	if err := mc.enqueueAndWait(context.Background(), metadataApplyRequest{
+		kind:      metadataSectionRefresh,
+		buildPath: util.FullPath("/dir"),
+		refresh:   &sectionRefresh{hi: "m", snapshotTsNs: 5000},
+	}); err != nil {
+		t.Fatalf("refresh: %v", err)
+	}
+
+	// a delayed create between the directory floor and the refresh snapshot,
+	// for a name neither the cache nor the listing ever held
+	ghost := func(tsNs int64) {
+		t.Helper()
+		if err := mc.ApplyMetadataResponse(context.Background(), &filer_pb.SubscribeMetadataResponse{
+			Directory: "/dir",
+			EventNotification: &filer_pb.EventNotification{
+				NewEntry: &filer_pb.Entry{
+					Name:       "b-ghost",
+					Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 1, FileMode: 0100644, FileSize: 1},
+				},
+			},
+			TsNs: tsNs,
+		}, SubscriberMetadataResponseApplyOptions); err != nil {
+			t.Fatalf("apply ghost event: %v", err)
+		}
+	}
+	ghost(4500)
+	if _, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/b-ghost")); err != filer_pb.ErrNotFound {
+		t.Fatalf("pre-snapshot event resurrected an absent name: %v", err)
+	}
+	ghost(6000)
+	if _, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/b-ghost")); err != nil {
+		t.Fatalf("post-snapshot event must apply: %v", err)
+	}
+}
+
+func TestUnversionedRefreshStaysStale(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
+		"/":    true,
+		"/dir": true,
+	})
+	defer mc.Shutdown()
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, nil)
+	applyChurn(t, mc, "/dir", "a-", sectionHotThreshold, 2000, SubscriberMetadataResponseApplyOptions)
+
+	if err := mc.enqueueAndWait(context.Background(), metadataApplyRequest{
+		kind:      metadataSectionRefresh,
+		buildPath: util.FullPath("/dir"),
+		refresh: &sectionRefresh{entries: []*filer.Entry{{
+			FullPath: util.FullPath("/dir/b-gap"),
+			Attr:     filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(1, 0), Mode: 0100644, FileSize: 1},
+		}}},
+	}); err != nil {
+		t.Fatalf("refresh: %v", err)
+	}
+
+	if mc.IsNameFresh(util.FullPath("/dir/a-000")) {
+		t.Fatal("an unversioned refresh vouches for nothing and must stay stale")
+	}
+	if _, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/b-gap")); err != nil {
+		t.Fatalf("gap fill should still land: %v", err)
+	}
+	if ranges := mc.staleRangesAhead(util.FullPath("/dir"), ""); len(ranges) != 0 {
+		t.Fatalf("an unverifiable section must not be re-listed, got %d ranges", len(ranges))
+	}
+}
+
+func TestEnsureListingFreshCoversAllStaleSectionsAhead(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
+		"/":    true,
+		"/dir": true,
+	})
+	defer mc.Shutdown()
+
+	server := &sectionFilerServer{snapshot: 5000, names: []string{"b-1", "n-1"}}
+	client := startSectionFilerServer(t, server)
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, []string{"m"})
+	applyChurn(t, mc, "/dir", "a-", sectionHotThreshold, 2000, SubscriberMetadataResponseApplyOptions)
+	applyChurn(t, mc, "/dir", "n-", sectionHotThreshold, 3000, SubscriberMetadataResponseApplyOptions)
+
+	if err := EnsureListingFresh(context.Background(), mc, client, util.FullPath("/dir"), ""); err != nil {
+		t.Fatalf("refresh: %v", err)
+	}
+	if !mc.IsNameFresh(util.FullPath("/dir/a-000")) || !mc.IsNameFresh(util.FullPath("/dir/n-000")) {
+		t.Fatal("one call must re-validate every stale section ahead of the start")
+	}
+}
+
+func TestSectionCompleteRefreshSetsFloors(t *testing.T) {
+	sl := newSectionTable([]string{"g", "p"})
+	sl.sections[1].stale = true
+
+	names := make([]string, 0, 2*dirSectionSize+1)
+	for i := 0; i <= 2*dirSectionSize; i++ {
+		names = append(names, fmt.Sprintf("g-%05d", i))
+	}
+	if !sl.completeRefresh("g", "p", names, 7000) {
+		t.Fatal("refresh must be accepted")
+	}
+	for _, name := range []string{"g", "g-99999", names[dirSectionSize+1]} {
+		if got := sl.floorOf(name); got != 7000 {
+			t.Fatalf("floorOf(%q) = %d, want 7000", name, got)
+		}
+	}
+	if got := sl.floorOf("a"); got != 0 {
+		t.Fatalf("floorOf(a) = %d, want 0 for a never-refreshed section", got)
+	}
+}
+
+func TestSectionFloorOutranksOlderTombstone(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
+		"/":    true,
+		"/dir": true,
+	})
+	defer mc.Shutdown()
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, nil)
+
+	event := func(tsNs int64, oldEntry, newEntry *filer_pb.Entry) {
+		t.Helper()
+		if err := mc.ApplyMetadataResponse(context.Background(), &filer_pb.SubscribeMetadataResponse{
+			Directory:         "/dir",
+			EventNotification: &filer_pb.EventNotification{OldEntry: oldEntry, NewEntry: newEntry},
+			TsNs:              tsNs,
+		}, SubscriberMetadataResponseApplyOptions); err != nil {
+			t.Fatalf("apply event at %d: %v", tsNs, err)
+		}
+	}
+	pbFile := &filer_pb.Entry{
+		Name:       "b-x",
+		Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 1, FileMode: 0100644, FileSize: 1},
+	}
+	event(1500, nil, pbFile)                       // create
+	event(2000, &filer_pb.Entry{Name: "b-x"}, nil) // delete, leaves a tombstone at 2000
+
+	applyChurn(t, mc, "/dir", "a-", sectionHotThreshold, 2500, SubscriberMetadataResponseApplyOptions)
+	if err := mc.enqueueAndWait(context.Background(), metadataApplyRequest{
+		kind:      metadataSectionRefresh,
+		buildPath: util.FullPath("/dir"),
+		refresh:   &sectionRefresh{snapshotTsNs: 5000},
+	}); err != nil {
+		t.Fatalf("refresh: %v", err)
+	}
+
+	// the floor at 5000 outranks the tombstone at 2000
+	event(4000, nil, pbFile)
+	if _, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/b-x")); err != filer_pb.ErrNotFound {
+		t.Fatalf("pre-floor event applied over an older tombstone: %v", err)
+	}
+	event(6000, nil, pbFile)
+	if _, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/b-x")); err != nil {
+		t.Fatalf("post-floor event must apply: %v", err)
+	}
+}
+
+func TestMismatchedRefreshLeavesStoreAlone(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
+		"/":    true,
+		"/dir": true,
+	})
+	defer mc.Shutdown()
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, []string{"m"})
+	if err := mc.InsertEntry(context.Background(), &filer.Entry{
+		FullPath: util.FullPath("/dir/b-keep"),
+		Attr:     filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(1, 0), Mode: 0100644, FileSize: 1},
+	}, 0); err != nil {
+		t.Fatalf("seed: %v", err)
+	}
+	applyChurn(t, mc, "/dir", "a-", sectionHotThreshold, 2000, SubscriberMetadataResponseApplyOptions)
+
+	// a range the table does not have: the reconcile must not touch the store
+	if err := mc.enqueueAndWait(context.Background(), metadataApplyRequest{
+		kind:      metadataSectionRefresh,
+		buildPath: util.FullPath("/dir"),
+		refresh: &sectionRefresh{hi: "q", snapshotTsNs: 5000, entries: []*filer.Entry{{
+			FullPath: util.FullPath("/dir/b-new"),
+			Attr:     filer.Attr{Crtime: time.Unix(2, 0), Mtime: time.Unix(2, 0), Mode: 0100644, FileSize: 3},
+		}}},
+	}); err != nil {
+		t.Fatalf("refresh: %v", err)
+	}
+
+	if _, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/b-new")); err != filer_pb.ErrNotFound {
+		t.Fatalf("mismatched refresh inserted an entry: %v", err)
+	}
+	if _, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/b-keep")); err != nil {
+		t.Fatalf("mismatched refresh swept an entry: %v", err)
+	}
+	if mc.IsNameFresh(util.FullPath("/dir/a-000")) {
+		t.Fatal("mismatched refresh must not mark anything fresh")
+	}
+}
+
+func TestRefreshClearsUnversionedMarker(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
+		"/":    true,
+		"/dir": true,
+	})
+	defer mc.Shutdown()
+	mc.SetPinnedChildFn(func(entry *filer.Entry) bool {
+		return entry.Name() == "b-pin"
+	})
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, nil)
+	for _, name := range []string{"b-local", "b-pin"} {
+		if err := mc.InsertEntry(context.Background(), &filer.Entry{
+			FullPath: util.FullPath("/dir/" + name),
+			Attr:     filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(1, 0), Mode: 0100644, FileSize: 1},
+		}, 0); err != nil { // versionTsNs 0 leaves the unversioned marker
+			t.Fatalf("seed %s: %v", name, err)
+		}
+	}
+	applyChurn(t, mc, "/dir", "a-", sectionHotThreshold, 2000, SubscriberMetadataResponseApplyOptions)
+
+	refreshed := func(name string) *filer.Entry {
+		return &filer.Entry{
+			FullPath: util.FullPath("/dir/" + name),
+			Attr:     filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(2, 0), Mode: 0100644, FileSize: 7},
+		}
+	}
+	if err := mc.enqueueAndWait(context.Background(), metadataApplyRequest{
+		kind:      metadataSectionRefresh,
+		buildPath: util.FullPath("/dir"),
+		refresh:   &sectionRefresh{snapshotTsNs: 5000, entries: []*filer.Entry{refreshed("b-local"), refreshed("b-pin")}},
+	}); err != nil {
+		t.Fatalf("refresh: %v", err)
+	}
+
+	entry, versionTsNs, err := mc.FindEntry(context.Background(), util.FullPath("/dir/b-local"))
+	if err != nil || entry.FileSize != 7 {
+		t.Fatalf("b-local = %+v, %v; want the snapshot's size 7", entry, err)
+	}
+	if versionTsNs != 5000 {
+		t.Fatalf("b-local version = %d, want the floor 5000 once the marker is cleared", versionTsNs)
+	}
+
+	// a delayed event below the floor must not roll the snapshot write back
+	if err := mc.ApplyMetadataResponse(context.Background(), &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{
+				Name:       "b-local",
+				Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 3, FileMode: 0100644, FileSize: 9},
+			},
+		},
+		TsNs: 4000,
+	}, SubscriberMetadataResponseApplyOptions); err != nil {
+		t.Fatalf("apply delayed event: %v", err)
+	}
+	if entry, _, err = mc.FindEntry(context.Background(), util.FullPath("/dir/b-local")); err != nil || entry.FileSize != 7 {
+		t.Fatalf("b-local rolled back to %+v, %v", entry, err)
+	}
+
+	// pinned local-only state is neither replaced nor unmarked
+	entry, versionTsNs, err = mc.FindEntry(context.Background(), util.FullPath("/dir/b-pin"))
+	if err != nil || entry.FileSize != 1 {
+		t.Fatalf("b-pin = %+v, %v; want local size 1 kept", entry, err)
+	}
+	if versionTsNs != 0 {
+		t.Fatalf("b-pin version = %d, want 0 while pinned", versionTsNs)
+	}
+}
+
+func TestRefreshSkipsBuildingDirectory(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
+		"/":    true,
+		"/dir": true,
+	})
+	defer mc.Shutdown()
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, nil)
+	applyChurn(t, mc, "/dir", "a-", sectionHotThreshold, 2000, SubscriberMetadataResponseApplyOptions)
+
+	// a rebuild is streaming: its inserts must survive a refresh's sweep
+	if err := mc.BeginDirectoryBuild(context.Background(), util.FullPath("/dir")); err != nil {
+		t.Fatalf("begin build: %v", err)
+	}
+	if err := mc.InsertEntry(context.Background(), &filer.Entry{
+		FullPath: util.FullPath("/dir/b-built"),
+		Attr:     filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(1, 0), Mode: 0100644, FileSize: 1},
+	}, 0); err != nil {
+		t.Fatalf("insert mid-build: %v", err)
+	}
+
+	if err := mc.enqueueAndWait(context.Background(), metadataApplyRequest{
+		kind:      metadataSectionRefresh,
+		buildPath: util.FullPath("/dir"),
+		refresh:   &sectionRefresh{snapshotTsNs: 5000},
+	}); err != nil {
+		t.Fatalf("refresh: %v", err)
+	}
+
+	if _, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/b-built")); err != nil {
+		t.Fatalf("refresh swept a mid-build insert: %v", err)
+	}
+	if mc.IsNameFresh(util.FullPath("/dir/a-000")) {
+		t.Fatal("a skipped refresh must not mark the section fresh")
+	}
+	if err := mc.AbortDirectoryBuild(context.Background(), util.FullPath("/dir")); err != nil {
+		t.Fatalf("abort build: %v", err)
+	}
+}
+
+// TestSectionBorderEntries pins the border semantics: a bound-named entry
+// belongs to the section starting at the bound, a refresh sweeps [lo, hi)
+// inclusive of lo and exclusive of hi, and the neighboring sections' entries
+// come through untouched.
+func TestSectionBorderEntries(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
+		"/":    true,
+		"/dir": true,
+	})
+	defer mc.Shutdown()
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, []string{"g", "p"})
+	for _, name := range []string{"g", "h", "p"} {
+		if err := mc.InsertEntry(context.Background(), &filer.Entry{
+			FullPath: util.FullPath("/dir/" + name),
+			Attr:     filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(1, 0), Mode: 0100644, FileSize: 1},
+		}, 0); err != nil {
+			t.Fatalf("seed %s: %v", name, err)
+		}
+	}
+	applyChurn(t, mc, "/dir", "a-", sectionHotThreshold, 2000, SubscriberMetadataResponseApplyOptions)
+	applyChurn(t, mc, "/dir", "h-", sectionHotThreshold, 3000, SubscriberMetadataResponseApplyOptions)
+
+	// the bound name is part of the section starting at it
+	if mc.IsNameFresh(util.FullPath("/dir/a-000")) || mc.IsNameFresh(util.FullPath("/dir/g")) {
+		t.Fatal("both churned sections should be stale")
+	}
+
+	// refreshing [ , g) sweeps its churn but must not reach the bound entry
+	if err := mc.enqueueAndWait(context.Background(), metadataApplyRequest{
+		kind:      metadataSectionRefresh,
+		buildPath: util.FullPath("/dir"),
+		refresh:   &sectionRefresh{hi: "g", snapshotTsNs: 5000},
+	}); err != nil {
+		t.Fatalf("refresh section 0: %v", err)
+	}
+	if _, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/a-000")); err != filer_pb.ErrNotFound {
+		t.Fatalf("churned a-000 should be swept: %v", err)
+	}
+	for _, name := range []string{"g", "h", "p"} {
+		if _, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/"+name)); err != nil {
+			t.Fatalf("%s swept by the neighboring refresh: %v", name, err)
+		}
+	}
+	if !mc.IsNameFresh(util.FullPath("/dir/a-000")) || mc.IsNameFresh(util.FullPath("/dir/g")) {
+		t.Fatal("only the refreshed section should be fresh")
+	}
+
+	// refreshing [g, p) covers the bound entry itself and stops before p
+	if err := mc.enqueueAndWait(context.Background(), metadataApplyRequest{
+		kind:      metadataSectionRefresh,
+		buildPath: util.FullPath("/dir"),
+		refresh: &sectionRefresh{lo: "g", hi: "p", snapshotTsNs: 5001, entries: []*filer.Entry{{
+			FullPath: util.FullPath("/dir/h"),
+			Attr:     filer.Attr{Crtime: time.Unix(1, 0), Mtime: time.Unix(2, 0), Mode: 0100644, FileSize: 7},
+		}}},
+	}); err != nil {
+		t.Fatalf("refresh section 1: %v", err)
+	}
+	if _, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/g")); err != filer_pb.ErrNotFound {
+		t.Fatalf("vanished bound entry should be swept by its own section: %v", err)
+	}
+	if _, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/h-000")); err != filer_pb.ErrNotFound {
+		t.Fatalf("churned h-000 should be swept: %v", err)
+	}
+	if entry, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/h")); err != nil || entry.FileSize != 7 {
+		t.Fatalf("h = %+v, %v; want size 7", entry, err)
+	}
+	if _, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/p")); err != nil {
+		t.Fatalf("the next bound's entry is outside the range and must survive: %v", err)
+	}
+}
+
+func TestChurnBeyondBuiltEntriesLandsInEdgeSection(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
+		"/":    true,
+		"/dir": true,
+	})
+	defer mc.Shutdown()
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, []string{"m"})
+
+	// names sorting past everything the build saw land in the last section
+	applyChurn(t, mc, "/dir", "z-", sectionHotThreshold, 2000, SubscriberMetadataResponseApplyOptions)
+
+	if mc.IsNameFresh(util.FullPath("/dir/z-000")) {
+		t.Fatal("the tail section should be invalidated")
+	}
+	if !mc.IsNameFresh(util.FullPath("/dir/a")) {
+		t.Fatal("the head section must stay fresh")
+	}
+}
+
+func TestRenameAcrossSectionsCountsBoth(t *testing.T) {
+	mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{
+		"/":    true,
+		"/dir": true,
+	})
+	defer mc.Shutdown()
+
+	buildSectionedDir(t, mc, util.FullPath("/dir"), 1000, []string{"m"})
+
+	for i := 0; i < sectionHotThreshold; i++ {
+		resp := &filer_pb.SubscribeMetadataResponse{
+			Directory: "/dir",
+			EventNotification: &filer_pb.EventNotification{
+				OldEntry: &filer_pb.Entry{Name: fmt.Sprintf("a-%03d", i)},
+				NewEntry: &filer_pb.Entry{
+					Name:       fmt.Sprintf("n-%03d", i),
+					Attributes: &filer_pb.FuseAttributes{Crtime: 1, Mtime: 1, FileMode: 0100644, FileSize: 1},
+				},
+				NewParentPath: "/dir",
+			},
+			TsNs: 2000 + int64(i),
+		}
+		if err := mc.ApplyMetadataResponse(context.Background(), resp, SubscriberMetadataResponseApplyOptions); err != nil {
+			t.Fatalf("apply rename %d: %v", i, err)
+		}
+	}
+
+	if mc.IsNameFresh(util.FullPath("/dir/a-000")) {
+		t.Fatal("the vacated side's section should be invalidated")
+	}
+	if mc.IsNameFresh(util.FullPath("/dir/n-000")) {
+		t.Fatal("the landing side's section should be invalidated")
+	}
+}

weed/mount/weedfs.go

@@ -179,8 +179,6 @@ type WFS struct {
 	dirMtimeMap           map[uint64]time.Time // inode -> mtime/ctime, in-memory overlay for dirs
 	entryValidSec         uint64               // kernel FUSE entry cache TTL in seconds
 	attrValidSec          uint64               // kernel FUSE attr cache TTL in seconds
-	dirHotWindow          time.Duration
-	dirHotThreshold       int
 	dirIdleEvict          time.Duration
 
 	// openMtimeCache maps inode -> [mtime_sec, mtime_ns] from the last Open.
@@ -222,11 +220,7 @@ type WFS struct {
 	lockClient *cluster.LockClient
 }
 
-const (
-	defaultDirHotWindow    = 2 * time.Second
-	defaultDirHotThreshold = 64
-	defaultDirIdleEvict    = 10 * time.Minute
-)
+const defaultDirIdleEvict = 10 * time.Minute
 
 func NewSeaweedFileSystem(option *Option) *WFS {
 	// Only create FilerClient for direct volume access modes
@@ -251,8 +245,6 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 		)
 	}
 
-	dirHotWindow := defaultDirHotWindow
-	dirHotThreshold := defaultDirHotThreshold
 	dirIdleEvict := defaultDirIdleEvict
 	if option.DirIdleEvictSec != 0 {
 		dirIdleEvict = time.Duration(option.DirIdleEvictSec) * time.Second
@@ -281,8 +273,6 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 		dirMtimeMap:       make(map[uint64]time.Time, 1024),
 		entryValidSec:     1,
 		attrValidSec:      1,
-		dirHotWindow:      dirHotWindow,
-		dirHotThreshold:   dirHotThreshold,
 		dirIdleEvict:      dirIdleEvict,
 	}
 
@@ -318,11 +308,7 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 			wfs.inodeToPath.MarkChildrenCached(path)
 		}, func(path util.FullPath) bool {
 			return wfs.inodeToPath.IsChildrenCached(path)
-		}, wfs.onEntryInvalidation, func(dirPath util.FullPath) {
-			if wfs.inodeToPath.RecordDirectoryUpdate(dirPath, time.Now(), wfs.dirHotWindow, wfs.dirHotThreshold) {
-				wfs.markDirectoryReadThrough(dirPath)
-			}
-		})
+		}, wfs.onEntryInvalidation, nil)
 	wfs.metaCache.SetPinnedChildFn(wfs.isLocalOnlyEntry)
 	grace.OnInterrupt(func() {
 		// grace calls os.Exit(0) after all hooks, so WaitForAsyncFlush
@@ -614,7 +600,7 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, entryVersion,
 	dir, _ := fullpath.DirAndName()
 	dirPath := util.FullPath(dir)
 
-	if wfs.metaCache.IsDirectoryCached(dirPath) {
+	if wfs.metaCache.IsDirectoryCached(dirPath) && wfs.metaCache.IsNameFresh(fullpath) {
 		cachedEntry, cachedVersionTsNs, cacheErr := wfs.metaCache.FindEntry(context.Background(), fullpath)
 		if cacheErr != nil && cacheErr != filer_pb.ErrNotFound {
 			glog.Errorf("lookupEntry: cache lookup for %s failed: %v", fullpath, cacheErr)
@@ -626,9 +612,9 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, entryVersion,
 			return cachedEntry, entryVersion{tsNs: cachedVersionTsNs}, fuse.OK
 		}
 		// Re-check: the directory may have been evicted from cache between
-		// our IsDirectoryCached check and FindEntry (e.g. markDirectoryReadThrough).
+		// our IsDirectoryCached check and FindEntry.
 		// If it's no longer cached, fall through to the filer lookup below.
-		if wfs.metaCache.IsDirectoryCached(dirPath) {
+		if wfs.metaCache.IsDirectoryCached(dirPath) && wfs.metaCache.IsNameFresh(fullpath) {
 			// Authoritative ENOENT only if inodeToPath also has no record.
 			// If the kernel still tracks this inode, the three layers
 			// disagree; trust the filer over the local cache (the
@@ -689,8 +675,7 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, entryVersion,
 						glog.V(4).Infof("lookupEntry found deferred entry in local cache %s", fullpath)
 						return localEntry, entryVersion{tsNs: localVersionTsNs}, fuse.OK
 					}
-					// Creating many files at once can push the directory past
-					// the hot threshold and evict it, which drops the local
+					// Cache eviction (idle, kernel Forget) can drop the local
 					// placeholder a deferred create left behind. The handle
 					// still holding the unflushed entry is authoritative for
 					// it, so read it from there rather than reporting a file
@@ -988,18 +973,6 @@ func (wfs *WFS) ClearCacheDir() {
 	os.RemoveAll(wfs.option.getUniqueCacheDirForRead())
 }
 
-// markDirectoryReadThrough drops a hot directory's cached listing. Only safe
-// from the apply loop (onDirectoryUpdate), where it serializes with a build's
-// markCachedFn; off-loop callers must use purgeDirectoryCache.
-func (wfs *WFS) markDirectoryReadThrough(dirPath util.FullPath) {
-	if !wfs.inodeToPath.MarkDirectoryReadThrough(dirPath, time.Now()) {
-		return
-	}
-	if err := wfs.metaCache.DeleteFolderChildren(context.Background(), dirPath); err != nil {
-		glog.V(2).Infof("clear dir cache %s: %v", dirPath, err)
-	}
-}
-
 // purgeDirectoryCache drops a directory's cached listing from off the apply loop
 // (idle eviction, kernel Forget, copy-range fallback), routing through it so a
 // stale wipe can't strand a concurrently-rebuilt directory cached-but-empty.

weed/mount/weedfs_dir_read.go

@@ -284,6 +284,16 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode
 				return fuse.EIO
 			}
 
+			// Serve the maintained-but-unverified cache if the filer is unreachable.
+			if err := meta_cache.EnsureListingFresh(context.Background(), wfs.metaCache, wfs, dirPath, ""); err != nil {
+				if errors.Is(err, meta_cache.ErrRefreshRangeTooLarge) {
+					// re-tile with a full rebuild; serve this request direct
+					wfs.purgeDirectoryCache(dirPath)
+					return wfs.readDirectoryDirect(input, out, dh, dirPath, processEachEntryFn)
+				}
+				glog.V(1).Infof("refresh %s sections: %v", dirPath, err)
+			}
+
 			// Load entries from beginning to fill cache up to the requested offset
 			storeLastName, loadErr := wfs.metaCache.ListDirectoryEntries(readdirContext, dirPath, "", false, skipCount+int64(batchSize), func(entry *filer.Entry) (bool, error) {
 				dh.entryStream = append(dh.entryStream, entry)
@@ -341,6 +351,16 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode
 			lastEntryName = dh.lastListedName
 		}
 
+		// Serve the maintained-but-unverified cache if the filer is unreachable.
+		if err := meta_cache.EnsureListingFresh(context.Background(), wfs.metaCache, wfs, dirPath, lastEntryName); err != nil {
+			if errors.Is(err, meta_cache.ErrRefreshRangeTooLarge) {
+				// re-tile with a full rebuild; serve this request direct
+				wfs.purgeDirectoryCache(dirPath)
+				return wfs.readDirectoryDirect(input, out, dh, dirPath, processEachEntryFn)
+			}
+			glog.V(1).Infof("refresh %s sections: %v", dirPath, err)
+		}
+
 		bufferFull := false
 		storeLastName, loadErr := wfs.metaCache.ListDirectoryEntries(readdirContext, dirPath, lastEntryName, false, int64(batchSize), func(entry *filer.Entry) (bool, error) {
 			currentIndex := int64(len(dh.entryStream))

weed/mount/weedfs_invalidate_open_handle_test.go

@@ -336,7 +336,7 @@ func TestBufferedBuildEventReinvalidatesOnCompletion(t *testing.T) {
 		t.Fatalf("insert listing entry: %v", err)
 	}
 
-	if err := wfs.metaCache.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 1000); err != nil {
+	if err := wfs.metaCache.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 1000, nil); err != nil {
 		t.Fatalf("complete build: %v", err)
 	}
 	wfs.metaCache.WaitForEntryInvalidations()
@@ -846,7 +846,7 @@ func TestFloorProtectsSnapshotStateFromDelayedEvents(t *testing.T) {
 	}, 2000); err != nil {
 		t.Fatalf("insert listing entry: %v", err)
 	}
-	if err := wfs.metaCache.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 2000); err != nil {
+	if err := wfs.metaCache.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 2000, nil); err != nil {
 		t.Fatalf("complete build: %v", err)
 	}
 
@@ -1131,7 +1131,7 @@ func TestAbsenceFloorBlocksGhostCreate(t *testing.T) {
 	}, 0); err != nil {
 		t.Fatalf("insert listing entry: %v", err)
 	}
-	if err := wfs.metaCache.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 2000); err != nil {
+	if err := wfs.metaCache.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 2000, nil); err != nil {
 		t.Fatalf("complete build: %v", err)
 	}
 
@@ -1281,7 +1281,7 @@ func TestNewerAbsenceFloorOverridesOlderTombstone(t *testing.T) {
 	if err := wfs.metaCache.BeginDirectoryBuild(context.Background(), util.FullPath("/dir")); err != nil {
 		t.Fatalf("begin build: %v", err)
 	}
-	if err := wfs.metaCache.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 3000); err != nil {
+	if err := wfs.metaCache.CompleteDirectoryBuild(context.Background(), util.FullPath("/dir"), 3000, nil); err != nil {
 		t.Fatalf("complete build: %v", err)
 	}