s3api: speed up recursive listing of branched directory trees (#12438)

* s3api: look up a trailing-slash prefix's directory instead of listing its parent

A prefix such as "account-0001/" names one directory, but the walk found it by
listing the parent with the name as a prefix. SQL stores answer that with a LIKE
that bounds the scan from below only, so in a wide bucket root the query reads
every sibling after the name: seconds for a prefix holding a few keys.

Look the directory up by name and serve it as the level's only entry. A miss,
an error or a non-directory falls back to the prefix listing.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3api: track failover receives atomically

The failover wrapper flags a received reply through a plain bool, which a
callback reading from the filer on several goroutines would race on. Use an
atomic.Bool so the callback may share its filer client.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3api: read directories ahead of a recursive listing walk

A recursive ListObjects walk enters directories one at a time, so a bucket
made of many directories holding one or two keys each costs one filer
round trip per directory, in series: seconds per 1000-key page on a store
whose reads wait on disk.

Read the first window of the directories the walk will enter on a few
workers, nearest first in walk order, and hand each window over when the
walk gets there. Every window the walk reads itself is drained first so its
subdirectories are queued before the walk descends into the first one. The
number of windows held ahead grows while directories end inside their first
window and falls back when one does not, so a page ending inside a large
directory leaves few reads wasted.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3api: drop the prefetcher's record of every directory it was handed

A page can cross any number of empty directories without spending its key
budget, so a set of every announced directory grows with the tree. A
directory is already handled when it is at or behind the walk, held or in
flight, or queued; check those, which the queue and window caps bound.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3api: end a trailing-slash prefix listing at the file its lookup finds

A file named by the prefix lists nothing under it, the same as the
sibling listing it fell back to, which threw the lookup's answer away
and paid the parent scan anyway.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3api: read a walk's own listing ahead instead of draining it

Draining the window before the walk took its first entry made a page
wait for the end of the listing, which a SQL store answering a name
prefix reaches only after scanning the rest of the directory; a page
that fills before then used to return at once. A background reader now
hands the window's directories to the prefetcher as they arrive while
the walk takes each entry, holding at most 1 MiB of entries ahead.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3api: bound the bytes a prefetched listing window holds

Entries with long chunk lists or inline content made the up to 64
held windows of 32 entries an unbounded amount of memory per listing.
A window now stops at 128 KiB of entries and the walk lists the rest
of that directory from the window's last entry.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* s3api: test recursive listings whose streams break partway

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/s3api/s3api_handlers.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"sync/atomic"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -107,7 +108,7 @@ func (s3a *S3ApiServer) withFilerClientFailover(ctx context.Context, preferred p
 
 	var lastErr error
 	for _, filer := range ordered {
-		received := false
+		var received atomic.Bool
 		fnRan := false
 		// Background, not ctx: WithGrpcClient's context only decides whether an
 		// error invalidates the shared connection, and that call is fn's to make,
@@ -144,7 +145,7 @@ func (s3a *S3ApiServer) withFilerClientFailover(ctx context.Context, preferred p
 		// fn consumed part of a response; a replay would stack a second copy
 		// onto whatever it accumulated, so surface the error unwrapped. The filer
 		// answered, so this also says nothing about its health.
-		if received {
+		if received.Load() {
 			return err
 		}
 		s3a.filerClient.RecordFilerFailure(filer)
@@ -194,18 +195,18 @@ func filerConnReady(ctx context.Context, conn *grpc.ClientConn) bool {
 	}
 }
 
-// receiveTrackingConn flags *received once a unary reply or a streamed message
+// receiveTrackingConn flags received once a unary reply or a streamed message
 // has been handed to the callback, the point past which a failover replay is
-// no longer transparent.
+// no longer transparent. The callback may share the client across goroutines.
 type receiveTrackingConn struct {
 	grpc.ClientConnInterface
-	received *bool
+	received *atomic.Bool
 }
 
 func (c receiveTrackingConn) Invoke(ctx context.Context, method string, args, reply any, opts ...grpc.CallOption) error {
 	err := c.ClientConnInterface.Invoke(ctx, method, args, reply, opts...)
 	if err == nil {
-		*c.received = true
+		c.received.Store(true)
 	}
 	return err
 }
@@ -220,13 +221,13 @@ func (c receiveTrackingConn) NewStream(ctx context.Context, desc *grpc.StreamDes
 
 type receiveTrackingStream struct {
 	grpc.ClientStream
-	received *bool
+	received *atomic.Bool
 }
 
 func (s receiveTrackingStream) RecvMsg(m any) error {
 	err := s.ClientStream.RecvMsg(m)
 	if err == nil {
-		*s.received = true
+		s.received.Store(true)
 	}
 	return err
 }

weed/s3api/s3api_object_handlers_list.go

@@ -428,6 +428,10 @@ func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, req listObjectsReq
 			maxKeys:               maxKeys,
 			prefixEndsOnDelimiter: prefixEndsOnDelimiter,
 		}
+		if delimiter != "/" {
+			cursor.prefetch = newListPrefetcher(ctx, client, maxKeys)
+			defer cursor.prefetch.close()
+		}
 
 		var lastEntryWasCommonPrefix bool
 		var lastCommonPrefix string
@@ -911,6 +915,9 @@ type ListingCursor struct {
 	// callback to its plain key; its "<name>/" CommonPrefix is deferred to the
 	// directory's own block position.
 	prefixObjectKeyOnly bool
+	// prefetch reads directories ahead of a recursive walk; nil lists each directory
+	// when the walk enters it.
+	prefetch *listPrefetcher
 }
 
 // excludedMarkerKey returns the key an exclusive marker names that the walk's cutoff no
@@ -1215,13 +1222,35 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 			InclusiveStartFrom: inclusiveStartFrom,
 		}
 
-		stream, listErr := client.ListEntries(ctx, request)
-		if listErr != nil {
-			if errors.Is(listErr, filer_pb.ErrNotFound) {
+		var stream listEntriesStream
+		windowLimit := request.Limit
+		// A trailing-slash prefix names one directory, the only entry this level
+		// lists; a file of that name lists nothing. Looking the name up spares
+		// listing every sibling it prefixes, which a SQL store answers by scanning
+		// the rest of the directory.
+		if cursor.prefixEndsOnDelimiter && prefix != "" && startName == "" {
+			if resp, lookupErr := client.LookupDirectoryEntry(ctx, &filer_pb.LookupDirectoryEntryRequest{Directory: dir, Name: prefix}); lookupErr == nil && resp.GetEntry() != nil {
+				stream = &listWindow{entries: []*filer_pb.Entry{resp.Entry}}
+			}
+		}
+		if stream == nil {
+			if entries, limit, ok := cursor.prefetch.take(request); ok {
+				stream, windowLimit = &listWindow{entries: entries}, limit
+			}
+		}
+		if stream == nil {
+			onlyPrefix := cursor.prefixEndsOnDelimiter
+			var listErr error
+			stream, listErr = cursor.prefetch.list(ctx, client, request, func(entry *filer_pb.Entry) bool {
+				return listWalkDescends(entry) && entry.Name+"/" > marker && (!onlyPrefix || entry.Name == prefix)
+			})
+			if listErr != nil {
+				if errors.Is(listErr, filer_pb.ErrNotFound) {
+					return
+				}
+				err = fmt.Errorf("list entries %+v: %w", request, listErr)
 				return
 			}
-			err = fmt.Errorf("list entries %+v: %w", request, listErr)
-			return
 		}
 
 		var entriesReceived uint32
@@ -1386,7 +1415,7 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 		if cursor.isTruncated || stopped {
 			return
 		}
-		if entriesReceived < request.Limit {
+		if entriesReceived < windowLimit {
 			err = flushDirs("")
 			return
 		}
@@ -1395,6 +1424,24 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 	}
 }
 
+type listEntriesStream interface {
+	Recv() (*filer_pb.ListEntriesResponse, error)
+}
+
+// listWindow serves entries already in hand the way a ListEntries stream does.
+type listWindow struct {
+	entries []*filer_pb.Entry
+}
+
+func (w *listWindow) Recv() (*filer_pb.ListEntriesResponse, error) {
+	if len(w.entries) == 0 {
+		return nil, io.EOF
+	}
+	entry := w.entries[0]
+	w.entries = w.entries[1:]
+	return &filer_pb.ListEntriesResponse{Entry: entry}, nil
+}
+
 // hiddenProbePageSize is the window one probe request asks the filer for.
 // hiddenProbeBudget caps how many entries one prefix's probe may scan; it
 // resets for every candidate prefix so one large all-deleted subtree cannot

weed/s3api/s3api_object_handlers_list_prefetch.go

@@ -0,0 +1,326 @@
+package s3api
+
+import (
+	"context"
+	"io"
+	"sort"
+	"strings"
+	"sync"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"google.golang.org/protobuf/proto"
+)
+
+const (
+	listPrefetchWorkers = 8
+	listPrefetchAhead   = 64
+	listPrefetchWindow  = 32
+	listPrefetchQueued  = 4096
+	listPrefetchBytes   = 128 << 10
+	listReadAheadBytes  = 1 << 20
+)
+
+// listPrefetcher reads directories ahead of a recursive listing walk. The walk enters
+// directories one at a time, so a tree of many small directories costs one filer round
+// trip per directory, in series. The prefetcher keeps several of those trips in flight:
+// it reads the first window of each directory the walk will enter, nearest first in
+// walk order ("<path>/" byte order), and hands the window over when the walk gets there.
+// A directory the walk has passed is dropped. The windows held or in flight are capped
+// at ahead, which starts at the worker count and doubles up to maxAhead while the walk
+// keeps finding directories that end inside their first window; a larger directory
+// keeps the walk busy on its own, and the cap falls back so a page ending inside it
+// leaves few windows unread. A window also ends at listPrefetchBytes of entries, and
+// the walk reads the rest of that directory itself.
+type listPrefetcher struct {
+	ctx      context.Context
+	cancel   context.CancelFunc
+	client   filer_pb.SeaweedFilerClient
+	window   uint32
+	maxAhead int
+	workers  int
+
+	running  sync.WaitGroup
+	mu       sync.Mutex
+	wake     *sync.Cond
+	streamed *sync.Cond
+	closed   bool
+	started  bool
+	ahead    int
+	pos      string
+	queued   []string
+	windows  map[string]*prefetchedWindow
+}
+
+type prefetchedWindow struct {
+	done    chan struct{}
+	entries []*filer_pb.Entry
+	more    bool
+	err     error
+}
+
+func newListPrefetcher(ctx context.Context, client filer_pb.SeaweedFilerClient, maxKeys uint16) *listPrefetcher {
+	p := &listPrefetcher{
+		client:   client,
+		window:   min(uint32(listPrefetchWindow), uint32(maxKeys)+2),
+		maxAhead: min(listPrefetchAhead, int(maxKeys)),
+		windows:  map[string]*prefetchedWindow{},
+	}
+	p.ctx, p.cancel = context.WithCancel(ctx)
+	p.workers = min(listPrefetchWorkers, p.maxAhead)
+	p.ahead = p.workers
+	p.wake = sync.NewCond(&p.mu)
+	p.streamed = sync.NewCond(&p.mu)
+	return p
+}
+
+// listWalkDescends reports whether a recursive walk lists the directory's children.
+func listWalkDescends(entry *filer_pb.Entry) bool {
+	return entry.IsDirectory && entry.Name != s3_constants.MultipartUploadsFolder &&
+		!strings.HasSuffix(entry.Name, s3_constants.VersionsFolder)
+}
+
+func (p *listPrefetcher) add(dir string) {
+	if p == nil {
+		return
+	}
+	key := dir + "/"
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.closed || key <= p.pos || p.windows[key] != nil {
+		return
+	}
+	i := sort.SearchStrings(p.queued, key)
+	if i >= listPrefetchQueued || (i < len(p.queued) && p.queued[i] == key) {
+		return
+	}
+	p.queued = append(p.queued, "")
+	copy(p.queued[i+1:], p.queued[i:])
+	p.queued[i] = key
+	if len(p.queued) > listPrefetchQueued {
+		p.queued = p.queued[:listPrefetchQueued]
+	}
+	if !p.started {
+		p.started = true
+		p.running.Add(p.workers)
+		for range p.workers {
+			go p.run()
+		}
+	}
+	p.wake.Signal()
+}
+
+// take hands over the prefetched window for the walk's request, along with the limit
+// the walk must compare the window against: a window shorter than that limit is the
+// end of the directory. ok is false when the request has to go to the filer.
+func (p *listPrefetcher) take(request *filer_pb.ListEntriesRequest) (entries []*filer_pb.Entry, limit uint32, ok bool) {
+	if p == nil {
+		return nil, 0, false
+	}
+	key := request.Directory + "/"
+	p.mu.Lock()
+	if key > p.pos {
+		p.pos = key
+		n := sort.SearchStrings(p.queued, key)
+		if n < len(p.queued) && p.queued[n] == key {
+			n++
+		}
+		p.queued = p.queued[n:]
+		for k := range p.windows {
+			if k < key {
+				delete(p.windows, k)
+			}
+		}
+	}
+	w := p.windows[key]
+	delete(p.windows, key)
+	p.wake.Broadcast()
+	p.mu.Unlock()
+
+	if w == nil || request.Prefix != "" || request.StartFromFileName != "" {
+		return nil, 0, false
+	}
+	select {
+	case <-w.done:
+	case <-p.ctx.Done():
+		return nil, 0, false
+	}
+	if w.err != nil {
+		return nil, 0, false
+	}
+	p.mu.Lock()
+	if w.more {
+		p.ahead = p.workers
+	} else {
+		p.ahead = min(p.ahead*2, p.maxAhead)
+		p.wake.Broadcast()
+	}
+	p.mu.Unlock()
+	entries, limit = w.entries, request.Limit
+	if uint32(len(entries)) >= limit {
+		entries = entries[:limit]
+	} else if w.more {
+		limit = uint32(len(entries))
+	}
+	return entries, limit, true
+}
+
+// close stops the reads and waits for them, so none outlives the filer client.
+func (p *listPrefetcher) close() {
+	if p == nil {
+		return
+	}
+	p.mu.Lock()
+	p.closed = true
+	p.wake.Broadcast()
+	p.streamed.Broadcast()
+	p.mu.Unlock()
+	p.cancel()
+	p.running.Wait()
+}
+
+func (p *listPrefetcher) run() {
+	defer p.running.Done()
+	for {
+		p.mu.Lock()
+		for !p.closed && (len(p.queued) == 0 || len(p.windows) >= p.ahead) {
+			p.wake.Wait()
+		}
+		if p.closed {
+			p.mu.Unlock()
+			return
+		}
+		key := p.queued[0]
+		p.queued = p.queued[1:]
+		w := &prefetchedWindow{done: make(chan struct{})}
+		p.windows[key] = w
+		p.mu.Unlock()
+
+		p.read(strings.TrimSuffix(key, "/"), w)
+	}
+}
+
+func (p *listPrefetcher) read(dir string, w *prefetchedWindow) {
+	defer close(w.done)
+	ctx, cancel := context.WithCancel(p.ctx)
+	defer cancel()
+	stream, err := p.client.ListEntries(ctx, &filer_pb.ListEntriesRequest{Directory: dir, Limit: p.window})
+	if err != nil {
+		w.err = err
+		return
+	}
+	for size := 0; ; {
+		resp, recvErr := stream.Recv()
+		if recvErr == io.EOF {
+			w.more = uint32(len(w.entries)) == p.window
+			break
+		}
+		if recvErr != nil {
+			w.err = recvErr
+			return
+		}
+		if resp.Entry == nil {
+			continue
+		}
+		w.entries = append(w.entries, resp.Entry)
+		if size += proto.Size(resp.Entry); size >= listPrefetchBytes {
+			w.more = true
+			break
+		}
+	}
+	for _, entry := range w.entries {
+		if listWalkDescends(entry) {
+			p.add(dir + "/" + entry.Name)
+		}
+	}
+}
+
+// listAhead is the walk's own listing of a directory, read ahead of the walk: each
+// directory the walk will descend into goes to the prefetcher as it arrives, so its
+// siblings are read while the walk is inside the first. The walk still takes each
+// entry as it arrives, and at most listReadAheadBytes of entries wait for it.
+type listAhead struct {
+	p       *listPrefetcher
+	entries []*filer_pb.Entry
+	sizes   []int
+	size    int
+	err     error
+}
+
+// list opens the walk's listing of a directory; descends picks the entries to prefetch.
+func (p *listPrefetcher) list(ctx context.Context, client filer_pb.SeaweedFilerClient, request *filer_pb.ListEntriesRequest, descends func(*filer_pb.Entry) bool) (listEntriesStream, error) {
+	if p == nil {
+		return client.ListEntries(ctx, request)
+	}
+	p.mu.Lock()
+	if p.closed {
+		p.mu.Unlock()
+		return client.ListEntries(ctx, request)
+	}
+	p.running.Add(1)
+	p.mu.Unlock()
+	stream, err := client.ListEntries(p.ctx, request)
+	if err != nil {
+		p.running.Done()
+		return nil, err
+	}
+	a := &listAhead{p: p}
+	go a.read(stream, request.Directory, descends)
+	return a, nil
+}
+
+func (a *listAhead) read(stream listEntriesStream, dir string, descends func(*filer_pb.Entry) bool) {
+	p := a.p
+	defer p.running.Done()
+	for {
+		p.mu.Lock()
+		for a.size >= listReadAheadBytes && !p.closed {
+			p.streamed.Wait()
+		}
+		closed := p.closed
+		p.mu.Unlock()
+		var resp *filer_pb.ListEntriesResponse
+		err := context.Canceled
+		if !closed {
+			resp, err = stream.Recv()
+		}
+		p.mu.Lock()
+		if err != nil {
+			a.err = err
+			p.streamed.Broadcast()
+			p.mu.Unlock()
+			return
+		}
+		entry := resp.Entry
+		if entry != nil {
+			size := proto.Size(entry)
+			a.entries = append(a.entries, entry)
+			a.sizes = append(a.sizes, size)
+			a.size += size
+			p.streamed.Broadcast()
+		}
+		p.mu.Unlock()
+		if entry != nil && descends(entry) {
+			p.add(dir + "/" + entry.Name)
+		}
+	}
+}
+
+func (a *listAhead) Recv() (*filer_pb.ListEntriesResponse, error) {
+	p := a.p
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	for len(a.entries) == 0 && a.err == nil {
+		p.streamed.Wait()
+	}
+	if len(a.entries) == 0 {
+		return nil, a.err
+	}
+	entry := a.entries[0]
+	a.entries[0] = nil
+	a.entries = a.entries[1:]
+	a.size -= a.sizes[0]
+	a.sizes = a.sizes[1:]
+	p.streamed.Broadcast()
+	return &filer_pb.ListEntriesResponse{Entry: entry}, nil
+}

weed/s3api/s3api_object_handlers_list_traversal_test.go

@@ -0,0 +1,400 @@
+package s3api
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"sort"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	grpc "google.golang.org/grpc"
+)
+
+// recordingFilerClient records the directories a listing asks the filer to list.
+type recordingFilerClient struct {
+	*testFilerClient
+	mu     sync.Mutex
+	listed []string
+}
+
+func (c *recordingFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) {
+	c.mu.Lock()
+	c.listed = append(c.listed, in.Directory)
+	c.mu.Unlock()
+	return c.testFilerClient.ListEntries(ctx, in, opts...)
+}
+
+func plainFile(name string) *filer_pb.Entry {
+	return &filer_pb.Entry{Name: name, Attributes: &filer_pb.FuseAttributes{}}
+}
+
+// A trailing-slash prefix names one directory. Listing its parent by name prefix
+// makes a SQL store scan every sibling after the name, which in a wide bucket root
+// costs seconds for a prefix holding a handful of keys.
+func TestTrailingSlashPrefixLooksUpItsDirectory(t *testing.T) {
+	newClient := func() *recordingFilerClient {
+		return &recordingFilerClient{testFilerClient: &testFilerClient{entriesByDir: map[string][]*filer_pb.Entry{
+			"/buckets/test": {
+				newDir("account-0001"), newDir("account-0001-old"), plainFile("account-0001.txt"),
+				newDir("account-00010"), plainFile("solo"),
+			},
+			"/buckets/test/account-0001":          {newDir("Contract"), plainFile("note")},
+			"/buckets/test/account-0001/Contract": {plainFile("document.pdf")},
+			"/buckets/test/account-0001-old":      {plainFile("stale")},
+			"/buckets/test/account-00010":         {plainFile("other")},
+		}}}
+	}
+
+	client := newClient()
+	seen := listedNames(t, client, listDirectoryRequest{dir: "/buckets/test", prefix: "account-0001", bucket: "test"}, &ListingCursor{maxKeys: 1000, prefixEndsOnDelimiter: true})
+	assert.Equal(t, []string{"document.pdf", "note"}, seen)
+	assert.NotContains(t, client.listed, "/buckets/test", "the named directory is looked up, not found by listing its parent")
+
+	client = newClient()
+	seen = listedNames(t, client, listDirectoryRequest{dir: "/buckets/test", prefix: "account-0001", bucket: "test"}, &ListingCursor{maxKeys: 1, prefixEndsOnDelimiter: true})
+	assert.Equal(t, []string{"document.pdf"}, seen)
+
+	for _, prefix := range []string{"missing", "solo", "account-000"} {
+		seen = listedNames(t, newClient(), listDirectoryRequest{dir: "/buckets/test", prefix: prefix, bucket: "test"}, &ListingCursor{maxKeys: 1000, prefixEndsOnDelimiter: true})
+		assert.Empty(t, seen, "prefix %q/ names no directory", prefix)
+	}
+
+	client = newClient()
+	seen = listedNames(t, client, listDirectoryRequest{dir: "/buckets/test", prefix: "solo", bucket: "test"}, &ListingCursor{maxKeys: 1000, prefixEndsOnDelimiter: true})
+	assert.Empty(t, seen)
+	assert.NotContains(t, client.listed, "/buckets/test", "a file found by the lookup ends the listing")
+}
+
+// branchedTreeKeys builds keys shaped like a bucket of many accounts, each holding a
+// few documents several directories down, with siblings that sort around the
+// "<name>/" block of a directory, and some directories longer than a prefetch window.
+func branchedTreeKeys(accounts int) (keys, emptyDirs []string) {
+	for i := 0; i < accounts; i++ {
+		keys = append(keys, fmt.Sprintf("acct-%02d/Contract/rec-%02d/2026/10/doc.pdf", i, i))
+		if i%3 == 0 {
+			keys = append(keys, fmt.Sprintf("acct-%02d/Contract/rec-%02d/2026/11/doc.pdf", i, i))
+		}
+		if i%4 == 0 {
+			keys = append(keys, fmt.Sprintf("acct-%02d/note.txt", i))
+		}
+		if i%5 == 0 {
+			keys = append(keys, fmt.Sprintf("acct-%02d-old/x", i))
+		}
+		if i%6 == 0 {
+			keys = append(keys, fmt.Sprintf("acct-%02d.txt", i))
+		}
+		if i%7 == 0 {
+			emptyDirs = append(emptyDirs, fmt.Sprintf("acct-%02d/empty", i))
+		}
+		if i%10 == 0 {
+			for j := 0; j < listPrefetchWindow+8; j++ {
+				keys = append(keys, fmt.Sprintf("acct-%02d/bulk/f-%02d", i, j))
+			}
+		}
+	}
+	sort.Strings(keys)
+	return keys, emptyDirs
+}
+
+func filerTreeOf(bucketDir string, keys, emptyDirs []string) map[string][]*filer_pb.Entry {
+	tree := map[string][]*filer_pb.Entry{bucketDir: {}}
+	seen := map[string]bool{}
+	add := func(path string, isDir bool) {
+		for {
+			dir, name := toDirAndName(path)
+			parent := bucketDir
+			if dir != "" {
+				parent += "/" + dir
+			}
+			if !seen[parent+"/"+name] {
+				seen[parent+"/"+name] = true
+				if isDir {
+					tree[parent] = append(tree[parent], newDir(name))
+					if _, ok := tree[parent+"/"+name]; !ok {
+						tree[parent+"/"+name] = []*filer_pb.Entry{}
+					}
+				} else {
+					tree[parent] = append(tree[parent], plainFile(name))
+				}
+			}
+			if dir == "" {
+				return
+			}
+			path, isDir = dir, true
+		}
+	}
+	for _, key := range keys {
+		add(key, false)
+	}
+	for _, dir := range emptyDirs {
+		add(dir, true)
+	}
+	for _, entries := range tree {
+		sort.Slice(entries, func(i, j int) bool { return entries[i].Name < entries[j].Name })
+	}
+	return tree
+}
+
+// slowFilerClient answers each ListEntries after a delay, like a store that has to
+// read a page from disk, and records how many listings were in flight at once.
+type slowFilerClient struct {
+	*testFilerClient
+	delay       time.Duration
+	mu          sync.Mutex
+	calls       int
+	inFlight    int
+	maxInFlight int
+}
+
+func (c *slowFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) {
+	c.mu.Lock()
+	c.calls++
+	c.inFlight++
+	c.maxInFlight = max(c.maxInFlight, c.inFlight)
+	c.mu.Unlock()
+	time.Sleep(c.delay)
+	c.mu.Lock()
+	c.inFlight--
+	c.mu.Unlock()
+	return c.testFilerClient.ListEntries(ctx, in, opts...)
+}
+
+// A recursive walk over many small directories pays one listing round trip per
+// directory. Reading ahead keeps several of them in flight and must leave the page,
+// and where the next page resumes, exactly as the serial walk has them.
+func TestRecursiveListingReadsAhead(t *testing.T) {
+	keys, emptyDirs := branchedTreeKeys(40)
+	t.Run("small entries", func(t *testing.T) {
+		testReadAheadMatchesSerialWalk(t, filerTreeOf("/buckets/test", keys, emptyDirs), keys)
+	})
+	// Entries this large stop prefetched windows and the read-ahead of the walk's own
+	// listing at their byte limits, short of their entry limits.
+	t.Run("large entries", func(t *testing.T) {
+		tree := filerTreeOf("/buckets/test", keys, emptyDirs)
+		for dir, entries := range tree {
+			size := listPrefetchBytes / 3
+			if dir == "/buckets/test" {
+				size = listReadAheadBytes / 4
+			}
+			for _, entry := range entries {
+				if !entry.IsDirectory {
+					entry.Content = make([]byte, size)
+				}
+			}
+		}
+		testReadAheadMatchesSerialWalk(t, tree, keys)
+	})
+}
+
+func testReadAheadMatchesSerialWalk(t *testing.T, tree map[string][]*filer_pb.Entry, keys []string) {
+	s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}}
+
+	walk := func(maxKeys uint16, readAhead bool) (seen []string, nextMarker string, cursor *ListingCursor, client *slowFilerClient) {
+		client = &slowFilerClient{testFilerClient: &testFilerClient{entriesByDir: tree}, delay: time.Millisecond}
+		cursor = &ListingCursor{maxKeys: maxKeys}
+		if readAhead {
+			cursor.prefetch = newListPrefetcher(context.Background(), client, maxKeys)
+			defer cursor.prefetch.close()
+		}
+		nextMarker, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test", bucket: "test"}, cursor, func(dir string, entry *filer_pb.Entry) {
+			seen = append(seen, strings.TrimPrefix(dir+"/"+entry.Name, "/buckets/test/"))
+			cursor.maxKeys--
+		})
+		require.NoError(t, err)
+		return
+	}
+
+	for _, maxKeys := range []uint16{1, 2, 7, 30, 1000} {
+		t.Run(fmt.Sprintf("max=%d", maxKeys), func(t *testing.T) {
+			wantSeen, wantMarker, wantCursor, _ := walk(maxKeys, false)
+			gotSeen, gotMarker, gotCursor, client := walk(maxKeys, true)
+			assert.Equal(t, wantSeen, gotSeen)
+			assert.Equal(t, wantMarker, gotMarker)
+			assert.Equal(t, wantCursor.isTruncated, gotCursor.isTruncated)
+			if maxKeys == 1000 {
+				assert.Equal(t, keys, gotSeen)
+				assert.Greater(t, client.maxInFlight, 1, "directories are listed concurrently")
+			}
+		})
+	}
+}
+
+// Paging a recursive listing through the S3 response builder, with read-ahead on,
+// returns every key once and in order.
+func TestRecursiveListingPagesWithReadAhead(t *testing.T) {
+	keys, emptyDirs := branchedTreeKeys(40)
+	filer := &fakeBucketDeleteFiler{entriesByDir: filerTreeOf("/buckets/test", keys, emptyDirs)}
+	s3a := newFailoverTestServer(t, startFakeFiler(t, filer))
+	s3a.option.BucketsPath = "/buckets"
+	s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
+	s3a.bucketConfigCache.Set("test", &BucketConfig{Name: "test"})
+
+	for _, prefix := range []string{"", "acct-0", "acct-03/", "acct-03/Contract/"} {
+		var want []string
+		for _, key := range keys {
+			if strings.HasPrefix(key, prefix) {
+				want = append(want, key)
+			}
+		}
+		for _, pageSize := range []uint16{1, 3, 10, 1000} {
+			t.Run(fmt.Sprintf("prefix=%s/max=%d", prefix, pageSize), func(t *testing.T) {
+				var got []string
+				marker := ""
+				for page := 0; ; page++ {
+					require.Less(t, page, len(keys)+1, "listing did not terminate")
+					response, err := s3a.listFilerEntries(context.Background(), listObjectsRequest{
+						bucket: "test", prefix: prefix, marker: marker, maxKeys: pageSize,
+					})
+					require.NoError(t, err)
+					for _, entry := range response.Contents {
+						got = append(got, entry.Key)
+					}
+					if !response.IsTruncated {
+						break
+					}
+					require.Greater(t, response.NextMarker, marker, "pagination must advance")
+					marker = response.NextMarker
+				}
+				assert.Equal(t, want, got)
+			})
+		}
+	}
+}
+
+// stallingFilerClient holds a listing open after its last entry until the caller
+// gives up, like a SQL store still scanning a directory past the last match.
+type stallingFilerClient struct {
+	*testFilerClient
+	stall func(*filer_pb.ListEntriesRequest) bool
+}
+
+type stallingStream struct {
+	grpc.ServerStreamingClient[filer_pb.ListEntriesResponse]
+	ctx context.Context
+}
+
+func (s stallingStream) Recv() (*filer_pb.ListEntriesResponse, error) {
+	resp, err := s.ServerStreamingClient.Recv()
+	if err == io.EOF {
+		<-s.ctx.Done()
+		return nil, s.ctx.Err()
+	}
+	return resp, err
+}
+
+func (c *stallingFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) {
+	stream, err := c.testFilerClient.ListEntries(ctx, in, opts...)
+	if err != nil || !c.stall(in) {
+		return stream, err
+	}
+	return stallingStream{ServerStreamingClient: stream, ctx: ctx}, nil
+}
+
+// A page that fills before a listing ends must not wait for the listing's end, and
+// the reads still open when the page is done are cancelled and joined.
+func TestRecursiveListingDoesNotWaitForStalledListings(t *testing.T) {
+	keys, emptyDirs := branchedTreeKeys(40)
+	tree := filerTreeOf("/buckets/test", keys, emptyDirs)
+
+	for _, maxKeys := range []uint16{2, 5} {
+		want := listedNames(t, &testFilerClient{entriesByDir: tree}, listDirectoryRequest{dir: "/buckets/test", bucket: "test"}, &ListingCursor{maxKeys: maxKeys})
+
+		ctx, cancel := context.WithCancel(context.Background())
+		client := &stallingFilerClient{testFilerClient: &testFilerClient{entriesByDir: tree}, stall: func(in *filer_pb.ListEntriesRequest) bool {
+			return in.Directory == "/buckets/test" || !strings.HasPrefix(in.Directory, "/buckets/test/acct-00")
+		}}
+		done := make(chan []string)
+		go func() {
+			cursor := &ListingCursor{maxKeys: maxKeys}
+			cursor.prefetch = newListPrefetcher(ctx, client, maxKeys)
+			got := listedNames(t, client, listDirectoryRequest{dir: "/buckets/test", bucket: "test"}, cursor)
+			cursor.prefetch.close()
+			done <- got
+		}()
+		select {
+		case got := <-done:
+			assert.Equal(t, want, got)
+		case <-time.After(10 * time.Second):
+			cancel()
+			t.Fatalf("max=%d: listing waited for a stalled stream", maxKeys)
+		}
+		cancel()
+	}
+}
+
+// failingFilerClient breaks the listings it picks after their first entry.
+type failingFilerClient struct {
+	*testFilerClient
+	fail   func(*filer_pb.ListEntriesRequest) bool
+	failed atomic.Int32
+}
+
+type failingStream struct {
+	grpc.ServerStreamingClient[filer_pb.ListEntriesResponse]
+	client   *failingFilerClient
+	received bool
+}
+
+func (s *failingStream) Recv() (*filer_pb.ListEntriesResponse, error) {
+	if s.received {
+		s.client.failed.Add(1)
+		return nil, errors.New("connection reset")
+	}
+	resp, err := s.ServerStreamingClient.Recv()
+	s.received = err == nil
+	return resp, err
+}
+
+func (c *failingFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) {
+	stream, err := c.testFilerClient.ListEntries(ctx, in, opts...)
+	if err != nil || !c.fail(in) {
+		return stream, err
+	}
+	return &failingStream{ServerStreamingClient: stream, client: c}, nil
+}
+
+// A listing that breaks before the page fills fails the walk, and a prefetched
+// window that breaks is read again by the walk itself.
+func TestRecursiveListingStreamErrors(t *testing.T) {
+	keys, emptyDirs := branchedTreeKeys(40)
+	tree := filerTreeOf("/buckets/test", keys, emptyDirs)
+	s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}}
+	walk := func(client *failingFilerClient, readAhead bool) (seen []string, err error) {
+		cursor := &ListingCursor{maxKeys: 1000}
+		if readAhead {
+			cursor.prefetch = newListPrefetcher(context.Background(), client, cursor.maxKeys)
+			defer cursor.prefetch.close()
+		}
+		_, err = s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test", bucket: "test"}, cursor, func(dir string, entry *filer_pb.Entry) {
+			seen = append(seen, strings.TrimPrefix(dir+"/"+entry.Name, "/buckets/test/"))
+			cursor.maxKeys--
+		})
+		return
+	}
+
+	for _, readAhead := range []bool{false, true} {
+		for _, dir := range []string{"/buckets/test", "/buckets/test/acct-03/Contract"} {
+			client := &failingFilerClient{testFilerClient: &testFilerClient{entriesByDir: tree}, fail: func(in *filer_pb.ListEntriesRequest) bool {
+				return in.Directory == dir
+			}}
+			_, err := walk(client, readAhead)
+			assert.ErrorContains(t, err, "connection reset", "readAhead=%v dir=%s", readAhead, dir)
+		}
+	}
+
+	client := &failingFilerClient{testFilerClient: &testFilerClient{entriesByDir: tree}, fail: func(in *filer_pb.ListEntriesRequest) bool {
+		return in.Limit == listPrefetchWindow
+	}}
+	seen, err := walk(client, true)
+	require.NoError(t, err)
+	assert.Equal(t, keys, seen)
+	assert.Positive(t, client.failed.Load(), "prefetched windows broke")
+}