diff --git a/weed/s3api/s3api_handlers.go b/weed/s3api/s3api_handlers.go index baaf58435a..1c55695a15 100644 --- a/weed/s3api/s3api_handlers.go +++ b/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 } diff --git a/weed/s3api/s3api_object_handlers_list.go b/weed/s3api/s3api_object_handlers_list.go index 080d7321bd..87dfb7b4b8 100644 --- a/weed/s3api/s3api_object_handlers_list.go +++ b/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 "/" 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 diff --git a/weed/s3api/s3api_object_handlers_list_prefetch.go b/weed/s3api/s3api_object_handlers_list_prefetch.go new file mode 100644 index 0000000000..b4917a92ba --- /dev/null +++ b/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 ("/" 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 +} diff --git a/weed/s3api/s3api_object_handlers_list_traversal_test.go b/weed/s3api/s3api_object_handlers_list_traversal_test.go new file mode 100644 index 0000000000..187838a7f9 --- /dev/null +++ b/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 +// "/" 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") +}