diff --git a/weed/command/mount.go b/weed/command/mount.go
index e8945ddc8..5784e80da 100644
--- a/weed/command/mount.go
+++ b/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")
diff --git a/weed/mount/inode_to_path.go b/weed/mount/inode_to_path.go
index 295b1a804..3a6595835 100644
--- a/weed/mount/inode_to_path.go
+++ b/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)
 	}
diff --git a/weed/mount/inode_to_path_test.go b/weed/mount/inode_to_path_test.go
index 3699653a3..e76dd4583 100644
--- a/weed/mount/inode_to_path_test.go
+++ b/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")
diff --git a/weed/mount/meta_cache/meta_cache.go b/weed/mount/meta_cache/meta_cache.go
index 6ffa31cac..7e234a786 100644
--- a/weed/mount/meta_cache/meta_cache.go
+++ b/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)
diff --git a/weed/mount/meta_cache/meta_cache_apply_test.go b/weed/mount/meta_cache/meta_cache_apply_test.go
index 9d830d881..2f363815a 100644
--- a/weed/mount/meta_cache/meta_cache_apply_test.go
+++ b/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)
 	}
 
diff --git a/weed/mount/meta_cache/meta_cache_build_test.go b/weed/mount/meta_cache/meta_cache_build_test.go
index 458111cca..e54f9fbe2 100644
--- a/weed/mount/meta_cache/meta_cache_build_test.go
+++ b/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)
 	}
 
diff --git a/weed/mount/meta_cache/meta_cache_init.go b/weed/mount/meta_cache/meta_cache_init.go
index 3f7c6d902..e3d8be7b9 100644
--- a/weed/mount/meta_cache/meta_cache_init.go
+++ b/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)
 		}
diff --git a/weed/mount/meta_cache/meta_cache_purge_test.go b/weed/mount/meta_cache/meta_cache_purge_test.go
index ef6b885ee..c7833c12a 100644
--- a/weed/mount/meta_cache/meta_cache_purge_test.go
+++ b/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)
 	}
 
diff --git a/weed/mount/meta_cache/meta_cache_sections.go b/weed/mount/meta_cache/meta_cache_sections.go
new file mode 100644
index 000000000..4dbba4786
--- /dev/null
+++ b/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
+}
diff --git a/weed/mount/meta_cache/meta_cache_sections_test.go b/weed/mount/meta_cache/meta_cache_sections_test.go
new file mode 100644
index 000000000..63200bc21
--- /dev/null
+++ b/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")
+	}
+}
diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go
index b3060dfb8..e4858f593 100644
--- a/weed/mount/weedfs.go
+++ b/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.
diff --git a/weed/mount/weedfs_dir_read.go b/weed/mount/weedfs_dir_read.go
index 428d25ebc..556626507 100644
--- a/weed/mount/weedfs_dir_read.go
+++ b/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))
diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go
index eec95ded6..260580259 100644
--- a/weed/mount/weedfs_invalidate_open_handle_test.go
+++ b/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)
 	}