mount: fix four readdir pagination bugs (#10624)

* mount: size the direct listing slice from the batch, not the offset

The limit passed here is skipCount+batchSize when a client resumes a
fresh handle partway through a directory, so preallocating for it turns
the client's cookie into an allocation: a readdir at offset 3,000,000
reserves 24MB before the first entry arrives, and an offset near the
uint32 ceiling asks makeslice for ~4.29e9 elements.

* mount: stop replaying a directory that shrank past the resume offset

A client that opens a fresh handle and resumes at a cookie from an
earlier, larger listing gets a preload that cannot reach the entry before
that offset. The resume name was then left empty and the follow-up batch
listed from the directory's first child again, so the client was handed
every name a second time.

The stream always runs from the first child, so failing to reach that
entry means the directory is simply shorter than the offset. That is the
end of it.

* mount: page a directory from where the store reached

The batch loader treated a short batch as the end of the directory, but
the meta cache drops an expired child after the store has already spent
it against the limit, so a batch that filled up could still deliver fewer
entries than asked for. A directory with a handful of expired children
would stop listing early and hide every child behind them; a whole batch
of expired ones truncated the listing to nothing.

ListDirectoryEntries now reports the name the store itself reached. That
is both the sound end-of-directory signal -- the store returning nothing
-- and the right cursor, since resuming from the last visible name would
re-read the dropped children every round and never get past a batch that
was entirely expired.

* mount: drop the entries a directory walk has already passed

entryStreamOffset was only ever written by reset, so the stream a handle
holds grew for the life of the walk and a directory was retained whole
even though nothing could read the entries behind the client's position
again. A 10M-entry walk parked millions of entries per handle, and
NFS-Ganesha opens several on the same directory.

Offsets index into the stream from entryStreamOffset, so advancing the
two together keeps them lined up. One entry is held back because the next
batch resumes from the name immediately before the offset. Seeking back
behind what is still held now restarts the directory, which is what the
offset scheme can honestly support -- it previously returned nothing.

weed/mount/meta_cache/meta_cache.go

@@ -626,7 +626,12 @@ func (mc *MetaCache) deleteFolderChildrenForRebuild(ctx context.Context, dirPath
 	return nil
 }
 
-func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) error {
+// ListDirectoryEntries reports the last name the store reached, which is not
+// the last name handed to eachEntryFunc: an expired child is dropped here after
+// the store has already spent it against limit. A caller paginating by count
+// would read a short batch as the end of the directory, and one resuming from
+// the last name it saw would re-read the dropped ones forever.
+func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
 	mc.RLock()
 	defer mc.RUnlock()
 
@@ -635,17 +640,13 @@ func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.Full
 		glog.Warningf("unsynchronized dir: %v", dirPath)
 	}
 
-	_, err := mc.localStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *filer.Entry) (bool, error) {
+	return mc.localStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *filer.Entry) (bool, error) {
 		if entry.TtlSec > 0 && entry.Crtime.Add(time.Duration(entry.TtlSec)*time.Second).Before(time.Now()) {
 			return true, nil
 		}
 		mc.mapIdFromFilerToLocal(entry)
 		return eachEntryFunc(entry)
 	})
-	if err != nil {
-		return err
-	}
-	return err
 }
 
 func (mc *MetaCache) Shutdown() {

weed/mount/weedfs_dir_read.go

@@ -33,11 +33,15 @@ type DirectoryHandle struct {
 	isFinished        bool
 	entryStream       []*filer.Entry
 	entryStreamOffset uint64
-	snapshotTsNs      int64 // snapshot timestamp for consistent readdir in direct mode
+	// lastListedName is how far the store itself reached, which runs ahead of
+	// the last visible entry whenever children are dropped as expired.
+	lastListedName string
+	snapshotTsNs   int64 // snapshot timestamp for consistent readdir in direct mode
 }
 
 func (dh *DirectoryHandle) reset() {
 	dh.isFinished = false
+	dh.lastListedName = ""
 	dh.snapshotTsNs = 0
 	// Nil out pointers to allow garbage collection of old entries,
 	// then reuse the slice's capacity to avoid re-allocations.
@@ -161,6 +165,11 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode
 
 	if input.Offset == 0 {
 		dh.reset()
+	} else if input.Offset < dh.entryStreamOffset {
+		// Seeking back before what the handle still holds. Start the directory
+		// again rather than reporting nothing; the preload below refills up to
+		// the requested offset.
+		dh.reset()
 	} else if dh.isFinished && input.Offset >= dh.entryStreamOffset {
 		entryCurrentIndex := input.Offset - dh.entryStreamOffset
 		if uint64(len(dh.entryStream)) <= entryCurrentIndex {
@@ -241,6 +250,19 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode
 
 	// Read from cache first, then load next batch if needed
 	if input.Offset >= dh.entryStreamOffset {
+		// Drop what the client has walked past. Offsets are indexes into the
+		// stream from entryStreamOffset, so advancing the two together keeps
+		// them lined up; one entry is kept back because the next batch resumes
+		// from the name immediately before the offset.
+		if trim := int(input.Offset-dh.entryStreamOffset) - 1; trim > 0 && trim <= len(dh.entryStream) {
+			copy(dh.entryStream, dh.entryStream[trim:])
+			for i := len(dh.entryStream) - trim; i < len(dh.entryStream); i++ {
+				dh.entryStream[i] = nil
+			}
+			dh.entryStream = dh.entryStream[:len(dh.entryStream)-trim]
+			dh.entryStreamOffset += uint64(trim)
+		}
+
 		// Handle case: new handle with non-zero offset but empty cache
 		// This happens when NFS-Ganesha opens multiple directory handles
 		if len(dh.entryStream) == 0 && input.Offset > dh.entryStreamOffset {
@@ -252,10 +274,11 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode
 			}
 
 			// Load entries from beginning to fill cache up to the requested offset
-			loadErr := wfs.metaCache.ListDirectoryEntries(readdirContext, dirPath, "", false, skipCount+int64(batchSize), func(entry *filer.Entry) (bool, error) {
+			storeLastName, loadErr := wfs.metaCache.ListDirectoryEntries(readdirContext, dirPath, "", false, skipCount+int64(batchSize), func(entry *filer.Entry) (bool, error) {
 				dh.entryStream = append(dh.entryStream, entry)
 				return true, nil
 			})
+			dh.lastListedName = storeLastName
 			if loadErr != nil {
 				glog.Errorf("list meta cache: %v", loadErr)
 				return fuse.EIO
@@ -266,6 +289,13 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode
 			entryPreviousIndex := (input.Offset - dh.entryStreamOffset) - 1
 			if uint64(len(dh.entryStream)) > entryPreviousIndex {
 				lastEntryName = dh.entryStream[entryPreviousIndex].Name()
+			} else {
+				// The stream runs from the directory's first child, so failing to
+				// reach the entry before this offset means the directory has since
+				// shrunk past it. Listing on from an empty name would replay the
+				// directory from the start and hand the client every name twice.
+				dh.isFinished = true
+				return fuse.OK
 			}
 		}
 
@@ -286,13 +316,18 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode
 			return fuse.EIO
 		}
 
-		// Batch loading: fetch batchSize entries starting from lastEntryName
-		loadedCount := 0
+		// Page from where the store itself reached, not from the last entry the
+		// sink saw. An expired child is counted against the batch and then
+		// dropped, so resuming from the last visible name would re-read it every
+		// round and never get past a batch that was entirely expired.
+		if dh.lastListedName > lastEntryName {
+			lastEntryName = dh.lastListedName
+		}
+
 		bufferFull := false
-		loadErr := wfs.metaCache.ListDirectoryEntries(readdirContext, dirPath, lastEntryName, false, int64(batchSize), func(entry *filer.Entry) (bool, error) {
+		storeLastName, loadErr := wfs.metaCache.ListDirectoryEntries(readdirContext, dirPath, lastEntryName, false, int64(batchSize), func(entry *filer.Entry) (bool, error) {
 			currentIndex := int64(len(dh.entryStream))
 			dh.entryStream = append(dh.entryStream, entry)
-			loadedCount++
 			if !processEachEntryFn(entry, currentIndex) {
 				bufferFull = true
 				return false, nil
@@ -303,10 +338,12 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode
 			glog.Errorf("list meta cache: %v", loadErr)
 			return fuse.EIO
 		}
+		dh.lastListedName = storeLastName
 
-		// Mark finished only when loading completed normally (not buffer full)
-		// and we got fewer entries than requested
-		if !bufferFull && loadedCount < batchSize {
+		// The store reaching nothing is the only sound end-of-directory signal:
+		// a batch can come back short because entries expired, not because the
+		// directory ran out.
+		if !bufferFull && storeLastName == "" {
 			dh.isFinished = true
 		}
 	}
@@ -335,6 +372,11 @@ func (wfs *WFS) readDirectoryDirect(input *fuse.ReadIn, out DirEntrySink, dh *Di
 			entryPreviousIndex := (input.Offset - dh.entryStreamOffset) - 1
 			if uint64(len(dh.entryStream)) > entryPreviousIndex {
 				lastEntryName = dh.entryStream[entryPreviousIndex].Name()
+			} else {
+				// See the cached path: the directory shrank past this offset, and
+				// resuming from an empty name would replay it from the start.
+				dh.isFinished = true
+				return fuse.OK
 			}
 		}
 
@@ -379,7 +421,14 @@ func (wfs *WFS) readDirectoryDirect(input *fuse.ReadIn, out DirEntrySink, dh *Di
 }
 
 func loadDirectoryEntriesDirect(ctx context.Context, client filer_pb.FilerClient, uidGidMapper *meta_cache.UidGidMapper, dirPath util.FullPath, startFileName string, includeStart bool, limit uint32, snapshotTsNs int64, includeSystemEntries bool) ([]*filer.Entry, int64, error) {
-	entries := make([]*filer.Entry, 0, limit)
+	// limit can be a client-supplied resume offset rather than a batch size, so
+	// preallocating for it would size the slice from where the caller happened to
+	// seek. Reserve a batch and let append find the rest.
+	prealloc := limit
+	if prealloc > batchSize {
+		prealloc = batchSize
+	}
+	entries := make([]*filer.Entry, 0, prealloc)
 	var actualSnapshotTsNs int64
 	err := client.WithFilerClient(false, func(sc filer_pb.SeaweedFilerClient) error {
 		var innerErr error

weed/mount/weedfs_dir_read_chunks_test.go

@@ -27,7 +27,7 @@ func TestListDirectoryEntriesOmitsChunks(t *testing.T) {
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			var seen int
-			err := wfs.metaCache.ListDirectoryEntries(tc.ctx, dir, "", false, 100, func(entry *filer.Entry) (bool, error) {
+			_, err := wfs.metaCache.ListDirectoryEntries(tc.ctx, dir, "", false, 100, func(entry *filer.Entry) (bool, error) {
 				seen++
 				if got := len(entry.Chunks) > 0; got != tc.wantChunks {
 					t.Errorf("%s: has chunks = %v, want %v", entry.Name(), got, tc.wantChunks)

weed/mount/weedfs_dir_read_pagination_test.go

@@ -0,0 +1,246 @@
+package mount
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/go-fuse/v2/fuse"
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/mount/meta_cache"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+// pagingSink collects names, accepting at most limit per round the way a kernel
+// reply buffer does.
+type pagingSink struct {
+	limit   int
+	names   []string
+	lastOff uint64
+	round   int
+}
+
+func (s *pagingSink) AddEntry(entry fuse.DirEntry) bool {
+	if s.round >= s.limit {
+		return false
+	}
+	s.round++
+	s.names = append(s.names, entry.Name)
+	s.lastOff = entry.Off
+	return true
+}
+
+func (s *pagingSink) AddEntryPlus(entry fuse.DirEntry) *fuse.EntryOut { return nil }
+func (s *pagingSink) TakesLookupRef() bool                            { return true }
+
+func newPagingWFS(tb testing.TB, dir util.FullPath, names []string, ttlSec int32) *WFS {
+	tb.Helper()
+	uidGidMapper, err := meta_cache.NewUidGidMapper("", "")
+	if err != nil {
+		tb.Fatalf("uid/gid mapper: %v", err)
+	}
+	root := util.FullPath("/")
+	wfs := &WFS{
+		option: &Option{
+			ChunkSizeLimit: 1024, ConcurrentReaders: 1, VolumeServerAccess: "filerProxy",
+			FilerAddresses:     []pb.ServerAddress{pb.NewServerAddressWithGrpcPort("127.0.0.1:1", 1)},
+			GrpcDialOption:     grpc.WithTransportCredentials(insecure.NewCredentials()),
+			FilerMountRootPath: "/", MountUid: 99, MountGid: 100, MountMode: 0o777,
+			MountMtime: time.Now(), MountCtime: time.Now(), UidGidMapper: uidGidMapper,
+		},
+		signature: 1, inodeToPath: NewInodeToPath(root, 0),
+		fhMap: NewFileHandleToInode(), dhMap: NewDirectoryHandleToInode(),
+		fhLockTable: util.NewLockTable[FileHandleId](), hardLinkLockTable: util.NewLockTable[string](),
+	}
+	wfs.metaCache = meta_cache.NewMetaCache(
+		filepath.Join(tb.TempDir(), "meta"), uidGidMapper, root, false,
+		func(p util.FullPath) { wfs.inodeToPath.MarkChildrenCached(p) },
+		func(p util.FullPath) bool { return wfs.inodeToPath.IsChildrenCached(p) },
+		func(meta_cache.EntryInvalidation) {}, nil,
+	)
+	tb.Cleanup(wfs.metaCache.Shutdown)
+
+	now := time.Now()
+	ctx := context.Background()
+	if err := wfs.metaCache.InsertEntry(ctx, &filer.Entry{
+		FullPath: dir,
+		Attr:     filer.Attr{Mode: os.ModeDir | 0o755, Mtime: now, Crtime: now},
+	}, 0); err != nil {
+		tb.Fatalf("insert dir: %v", err)
+	}
+	for _, name := range names {
+		child := dir.Child(name)
+		attr := filer.Attr{Mode: 0o644, Mtime: now, Crtime: now, FileSize: 1, Inode: child.AsInode(now.Unix())}
+		if ttlSec > 0 {
+			// Expired well in the past, so the meta cache drops it after the
+			// store has already counted it against the limit.
+			attr.TtlSec = ttlSec
+			attr.Crtime = now.Add(-time.Duration(ttlSec+60) * time.Second)
+		}
+		if err := wfs.metaCache.InsertEntry(ctx, &filer.Entry{FullPath: child, Attr: attr}, 0); err != nil {
+			tb.Fatalf("insert %s: %v", name, err)
+		}
+	}
+	wfs.inodeToPath.MarkChildrenCached(root)
+	wfs.inodeToPath.Lookup(dir, now.Unix(), true, false, 0, true)
+	wfs.inodeToPath.MarkChildrenCached(dir)
+	return wfs
+}
+
+// TestReadDirResumePastShrunkDirectory covers a client resuming a fresh handle
+// at a cookie from an earlier, larger listing. The directory has since shrunk
+// past that offset, and the readdir used to fall back to listing from the first
+// child again, handing back names the client had already consumed.
+func TestReadDirResumePastShrunkDirectory(t *testing.T) {
+	dir := util.FullPath("/d")
+	var names []string
+	for i := 0; i < 100; i++ {
+		names = append(names, fmt.Sprintf("f%03d", i))
+	}
+	wfs := newPagingWFS(t, dir, names, 0)
+	dirInode, _ := wfs.inodeToPath.GetInode(dir)
+
+	dhid, _ := wfs.AcquireDirectoryHandle()
+	defer wfs.ReleaseDirectoryHandle(dhid)
+
+	sink := &pagingSink{limit: 4096}
+	status := wfs.doReadDirectory(&fuse.ReadIn{
+		InHeader: fuse.InHeader{NodeId: dirInode},
+		Fh:       uint64(dhid),
+		Offset:   152, // past the end of a directory that now holds 100
+		Size:     1 << 20,
+	}, sink, false)
+	if status != fuse.OK {
+		t.Fatalf("readdir: %v", status)
+	}
+	if len(sink.names) != 0 {
+		t.Errorf("resuming past the end returned %d entries (first %q), want none", len(sink.names), sink.names[0])
+	}
+}
+
+// TestReadDirWithExpiredEntries covers a batch the store fills to the limit but
+// whose entries the meta cache then drops as expired. The post-filter count used
+// to be read as end-of-directory, silently hiding every later child.
+func TestReadDirWithExpiredEntries(t *testing.T) {
+	dir := util.FullPath("/d")
+	// One full store batch of entries that all expire, then live ones behind
+	// them. Nothing survives the first batch, which is the case that latched.
+	var names []string
+	for i := 0; i < batchSize; i++ {
+		names = append(names, fmt.Sprintf("a%05d", i))
+	}
+	wfs := newPagingWFS(t, dir, names, 30)
+
+	live := []string{"z001", "z002", "z003"}
+	now := time.Now()
+	for _, name := range live {
+		child := dir.Child(name)
+		if err := wfs.metaCache.InsertEntry(context.Background(), &filer.Entry{
+			FullPath: child,
+			Attr:     filer.Attr{Mode: 0o644, Mtime: now, Crtime: now, FileSize: 1, Inode: child.AsInode(now.Unix())},
+		}, 0); err != nil {
+			t.Fatalf("insert %s: %v", name, err)
+		}
+	}
+
+	dirInode, _ := wfs.inodeToPath.GetInode(dir)
+	dhid, _ := wfs.AcquireDirectoryHandle()
+	defer wfs.ReleaseDirectoryHandle(dhid)
+
+	sink := &pagingSink{limit: 4096}
+	var offset uint64
+	for round := 0; round < 20; round++ {
+		sink.round = 0
+		status := wfs.doReadDirectory(&fuse.ReadIn{
+			InHeader: fuse.InHeader{NodeId: dirInode},
+			Fh:       uint64(dhid),
+			Offset:   offset,
+			Size:     1 << 20,
+		}, sink, false)
+		if status != fuse.OK {
+			t.Fatalf("readdir: %v", status)
+		}
+		if sink.lastOff <= offset {
+			break
+		}
+		offset = sink.lastOff
+	}
+
+	var seen []string
+	for _, n := range sink.names {
+		if n != "." && n != ".." {
+			seen = append(seen, n)
+		}
+	}
+	if len(seen) != len(live) {
+		t.Fatalf("listed %d entries %v, want the %d live ones %v", len(seen), seen, len(live), live)
+	}
+}
+
+// TestReadDirTrimsConsumedEntries checks that a walk does not accumulate the
+// whole directory in the handle. Offsets index into the stream from
+// entryStreamOffset, so the two have to advance together or the listing
+// silently misaligns.
+func TestReadDirTrimsConsumedEntries(t *testing.T) {
+	dir := util.FullPath("/d")
+	const total = 5000
+	var names []string
+	for i := 0; i < total; i++ {
+		names = append(names, fmt.Sprintf("f%05d", i))
+	}
+	wfs := newPagingWFS(t, dir, names, 0)
+	dirInode, _ := wfs.inodeToPath.GetInode(dir)
+
+	dhid, dh := wfs.AcquireDirectoryHandle()
+	defer wfs.ReleaseDirectoryHandle(dhid)
+
+	// A small sink forces many rounds, which is when the stream would grow.
+	sink := &pagingSink{limit: 64}
+	var offset uint64
+	var seen []string
+	peak := 0
+	for round := 0; round < 500; round++ {
+		sink.round = 0
+		before := len(sink.names)
+		status := wfs.doReadDirectory(&fuse.ReadIn{
+			InHeader: fuse.InHeader{NodeId: dirInode},
+			Fh:       uint64(dhid),
+			Offset:   offset,
+			Size:     1 << 20,
+		}, sink, false)
+		if status != fuse.OK {
+			t.Fatalf("readdir: %v", status)
+		}
+		if n := len(dh.entryStream); n > peak {
+			peak = n
+		}
+		if len(sink.names) == before || sink.lastOff <= offset {
+			break
+		}
+		offset = sink.lastOff
+	}
+	for _, n := range sink.names {
+		if n != "." && n != ".." {
+			seen = append(seen, n)
+		}
+	}
+
+	if len(seen) != total {
+		t.Fatalf("listed %d entries, want %d", len(seen), total)
+	}
+	for i, name := range seen {
+		if want := fmt.Sprintf("f%05d", i); name != want {
+			t.Fatalf("entry %d is %q, want %q -- trimming misaligned the offsets", i, name, want)
+		}
+	}
+	// Without trimming the handle ends up holding every entry.
+	if peak >= total {
+		t.Errorf("handle held %d entries at peak, want well under the %d in the directory", peak, total)
+	}
+}