mount: drop consumed entries when reading a directory through (#10802)

The cached readdir trims the head of the handle's entry stream as the
client walks past it; the read-through path never did, so a directory
too large to cache -- the only kind that takes that path -- was held
whole in the handle for the length of the walk. Hoist the trim to cover
both paths.

weed/mount/weedfs_dir_read.go

@@ -53,6 +53,26 @@ func (dh *DirectoryHandle) reset() {
 	dh.entryStreamOffset = directoryStreamBaseOffset
 }
 
+// dropConsumed releases the entries the client has already 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.
+func (dh *DirectoryHandle) dropConsumed(offset uint64) {
+	if offset < dh.entryStreamOffset {
+		return
+	}
+	trim := int(offset-dh.entryStreamOffset) - 1
+	if trim <= 0 || trim > len(dh.entryStream) {
+		return
+	}
+	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)
+}
+
 type DirectoryHandleToInode struct {
 	sync.Mutex
 	dir2inode map[DirectoryHandleId]*DirectoryHandle
@@ -251,25 +271,17 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode
 
 	var lastEntryName string
 
+	// Both the cached and the direct walk page through the same stream, and a
+	// directory read through is precisely one too big to hold, so drop the
+	// consumed head before either of them appends to it.
+	dh.dropConsumed(input.Offset)
+
 	if wfs.inodeToPath.ShouldReadDirectoryDirect(dirPath) {
 		return wfs.readDirectoryDirect(input, out, dh, dirPath, processEachEntryFn)
 	}
 
 	// 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 {

weed/mount/weedfs_dir_read_pagination_test.go

@@ -12,6 +12,7 @@ import (
 	"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/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
@@ -244,3 +245,94 @@ func TestReadDirTrimsConsumedEntries(t *testing.T) {
 		t.Errorf("handle held %d entries at peak, want well under the %d in the directory", peak, total)
 	}
 }
+
+// listingFilerServer serves a fixed, sorted child list the way the filer
+// paginates one.
+type listingFilerServer struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	names []string
+}
+
+func (s *listingFilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream filer_pb.SeaweedFiler_ListEntriesServer) error {
+	var sent uint32
+	for _, name := range s.names {
+		if name < req.StartFromFileName || (name == req.StartFromFileName && !req.InclusiveStartFrom) {
+			continue
+		}
+		if req.Limit > 0 && sent >= req.Limit {
+			break
+		}
+		if err := stream.Send(&filer_pb.ListEntriesResponse{Entry: &filer_pb.Entry{
+			Name:       name,
+			Attributes: &filer_pb.FuseAttributes{FileMode: 0o644, FileSize: 1},
+		}}); err != nil {
+			return err
+		}
+		sent++
+	}
+	return nil
+}
+
+// TestReadDirDirectTrimsConsumedEntries is the same check for a directory read
+// through to the filer. That path is taken precisely for directories too big to
+// cache, so holding the whole walk in the handle is where a listing turns into
+// gigabytes of resident memory.
+func TestReadDirDirectTrimsConsumedEntries(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, nil, 0)
+	startFakeFiler(t, wfs, &listingFilerServer{names: names})
+	dirInode, _ := wfs.inodeToPath.GetInode(dir)
+	if !wfs.inodeToPath.MarkDirectoryReadThrough(dir, time.Now()) {
+		t.Fatal("directory did not enter read-through mode")
+	}
+
+	dhid, dh := wfs.AcquireDirectoryHandle()
+	defer wfs.ReleaseDirectoryHandle(dhid)
+
+	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)
+		}
+	}
+	if peak >= total {
+		t.Errorf("handle held %d entries at peak, want well under the %d in the directory", peak, total)
+	}
+}

weed/mount/weedfs_invalidate_open_handle_test.go

@@ -156,7 +156,7 @@ func (s *fakeFilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateE
 }
 
 // startFakeFiler serves fake on a local port and points wfs at it.
-func startFakeFiler(t *testing.T, wfs *WFS, fake *fakeFilerServer) {
+func startFakeFiler(t *testing.T, wfs *WFS, fake filer_pb.SeaweedFilerServer) {
 	t.Helper()
 	listener, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {