s3: bound ListMultipartUploads memory to the page size (#11971)

* s3: bound ListMultipartUploads memory to the page size

listMultipartUploads loaded every open upload in the bucket into a slice
and selectMultipartUploads filtered and sorted a second full copy, so one
request cost O(total open uploads) heap regardless of prefix, key-marker,
or max-uploads; concurrent calls multiplied it and could OOM the gateway.

The response is ordered by the object key kept in extended attributes, not
the upload id the filer sorts by, so every entry still has to be seen.
Streaming the listing into a bounded top-K collector keeps at most
maxUploads (capped at maxUploadsList) candidates, bounding heap by the
page instead of the folder. A caller-supplied max-uploads above the
server cap now also paginates rather than accumulating unboundedly.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* s3: cover bounded ListMultipartUploads accumulation

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* s3: report the effective max-uploads cap in ListMultipartUploads

* s3: cover ListMultipartUploads retry through the collector

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/s3api/filer_multipart.go

@@ -2,6 +2,7 @@ package s3api
 
 import (
 	"cmp"
+	"container/heap"
 	"context"
 	"crypto/md5"
 	"crypto/rand"
@@ -1292,7 +1293,15 @@ func (s3a *S3ApiServer) listMultipartUploads(input *s3.ListMultipartUploadsInput
 
 	glog.V(2).Infof("listMultipartUploads input %v", input)
 
-	entries, _, err := s3a.list(s3a.genUploadsFolder(*input.Bucket), "", "", false, math.MaxInt32)
+	uploadsFolder := s3a.genUploadsFolder(*input.Bucket)
+	var page *multipartUploadPage
+	err := listRetry(uploadsFolder, func() error {
+		page = newMultipartUploadPage(input)
+		return filer_pb.List(context.Background(), s3a, uploadsFolder, "", func(entry *filer_pb.Entry, isLastEntry bool) error {
+			page.add(entry)
+			return nil
+		}, "", false, math.MaxInt32)
+	})
 	if err != nil {
 		// A missing .uploads folder normally lists as empty with no error; a
 		// store that reports it as not-found still means an empty list.
@@ -1304,55 +1313,104 @@ func (s3a *S3ApiServer) listMultipartUploads(input *s3.ListMultipartUploadsInput
 		return nil, s3err.ErrInternalError
 	}
 
-	return selectMultipartUploads(input, entries), s3err.ErrNone
+	return page.result(), s3err.ErrNone
 }
 
 // selectMultipartUploads orders uploads by key, then by upload id, and resumes
 // after the (key-marker, upload-id-marker) position, so an aborted marker
 // upload does not hide the rest of its key.
 func selectMultipartUploads(input *s3.ListMultipartUploadsInput, entries []*filer_pb.Entry) *ListMultipartUploadsResult {
+	page := newMultipartUploadPage(input)
+	for _, entry := range entries {
+		page.add(entry)
+	}
+	return page.result()
+}
+
+// multipartUploadPage accumulates one page of a ListMultipartUploads result.
+// Upload order is the object key kept in extended attributes, not the upload
+// id the filer sorts by, so every entry has to be seen; keeping only the
+// maxUploads smallest candidates bounds the heap by the page, not the folder.
+type multipartUploadPage struct {
+	input *s3.ListMultipartUploadsInput
+	limit int
+	top   candidateHeap
+	more  bool
+}
+
+type multipartCandidate struct {
+	key   string
+	entry *filer_pb.Entry
+}
+
+func compareMultipartCandidate(a, b multipartCandidate) int {
+	return cmp.Or(strings.Compare(a.key, b.key), strings.Compare(a.entry.Name, b.entry.Name))
+}
+
+// candidateHeap is a max-heap on (key, upload id): the root sorts last of the
+// candidates held, so popping it keeps the page's smallest entries.
+type candidateHeap []multipartCandidate
+
+func (h candidateHeap) Len() int { return len(h) }
+func (h candidateHeap) Less(i, j int) bool {
+	return compareMultipartCandidate(h[i], h[j]) > 0
+}
+func (h candidateHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
+func (h *candidateHeap) Push(x any)   { *h = append(*h, x.(multipartCandidate)) }
+func (h *candidateHeap) Pop() any {
+	old := *h
+	n := len(old)
+	*h = old[:n-1]
+	return old[n-1]
+}
+
+func newMultipartUploadPage(input *s3.ListMultipartUploadsInput) *multipartUploadPage {
+	limit := maxUploadsList
+	if *input.MaxUploads < int64(limit) {
+		limit = int(*input.MaxUploads)
+	}
+	return &multipartUploadPage{input: input, limit: limit}
+}
+
+func (p *multipartUploadPage) add(entry *filer_pb.Entry) {
+	if p.limit <= 0 || !isMultipartUploadEntry(entry) {
+		return
+	}
+	input := p.input
+	key := *objectKey(aws.String(string(entry.Extended[s3_constants.ExtMultipartObjectKey])))
+	if *input.Prefix != "" && !strings.HasPrefix(key, *input.Prefix) {
+		return
+	}
+	if *input.KeyMarker != "" && (key < *input.KeyMarker || key == *input.KeyMarker && (*input.UploadIdMarker == "" || entry.Name <= *input.UploadIdMarker)) {
+		return
+	}
+	c := multipartCandidate{key: key, entry: entry}
+	if p.top.Len() >= p.limit {
+		p.more = true
+		if compareMultipartCandidate(c, p.top[0]) >= 0 {
+			return
+		}
+		heap.Pop(&p.top)
+	}
+	heap.Push(&p.top, c)
+}
+
+func (p *multipartUploadPage) result() *ListMultipartUploadsResult {
+	input := p.input
 	output := &ListMultipartUploadsResult{
 		Bucket:         input.Bucket,
 		Delimiter:      input.Delimiter,
 		EncodingType:   input.EncodingType,
 		KeyMarker:      input.KeyMarker,
-		MaxUploads:     input.MaxUploads,
+		MaxUploads:     aws.Int64(int64(p.limit)),
 		Prefix:         input.Prefix,
 		UploadIdMarker: input.UploadIdMarker,
 		IsTruncated:    aws.Bool(false),
 	}
-	if *input.MaxUploads <= 0 {
-		return output
-	}
-	type candidate struct {
-		key   string
-		entry *filer_pb.Entry
-	}
-	var candidates []candidate
-	for _, entry := range entries {
-		if !isMultipartUploadEntry(entry) {
-			continue
-		}
-		key := *objectKey(aws.String(string(entry.Extended[s3_constants.ExtMultipartObjectKey])))
-		if *input.Prefix != "" && !strings.HasPrefix(key, *input.Prefix) {
-			continue
-		}
-		if *input.KeyMarker != "" && (key < *input.KeyMarker || key == *input.KeyMarker && (*input.UploadIdMarker == "" || entry.Name <= *input.UploadIdMarker)) {
-			continue
-		}
-		candidates = append(candidates, candidate{key: key, entry: entry})
-	}
-	slices.SortFunc(candidates, func(a, b candidate) int {
-		return cmp.Or(strings.Compare(a.key, b.key), strings.Compare(a.entry.Name, b.entry.Name))
-	})
-
+	candidates := make([]multipartCandidate, len(p.top))
+	copy(candidates, p.top)
+	slices.SortFunc(candidates, compareMultipartCandidate)
 	for _, c := range candidates {
-		if int64(len(output.Upload)) >= *input.MaxUploads {
-			last := output.Upload[len(output.Upload)-1]
-			output.IsTruncated = aws.Bool(true)
-			output.NextKeyMarker, output.NextUploadIdMarker = last.Key, last.UploadId
-			return output
-		}
 		upload := &s3.MultipartUpload{
 			Key:      aws.String(c.key),
 			UploadId: aws.String(c.entry.Name),
@@ -1362,6 +1420,11 @@ func selectMultipartUploads(input *s3.ListMultipartUploadsInput, entries []*file
 		}
 		output.Upload = append(output.Upload, upload)
 	}
+	if p.more {
+		last := output.Upload[len(output.Upload)-1]
+		output.IsTruncated = aws.Bool(true)
+		output.NextKeyMarker, output.NextUploadIdMarker = last.Key, last.UploadId
+	}
 	return output
 }
 

weed/s3api/filer_multipart_list_test.go

@@ -1,6 +1,9 @@
 package s3api
 
 import (
+	"context"
+	"fmt"
+	"math"
 	"slices"
 	"testing"
 
@@ -8,6 +11,8 @@ import (
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 // uploadEntry is an upload folder as createMultipartUpload leaves it.
@@ -112,3 +117,84 @@ func TestSelectMultipartUploadsPaginatesEveryUploadOnce(t *testing.T) {
 		t.Fatalf("listed %d of %d uploads", len(seen), len(entries))
 	}
 }
+
+func TestSelectMultipartUploadsHoldsOnlyThePage(t *testing.T) {
+	// a folder far larger than the page must not accumulate in the collector:
+	// peak memory tracks maxUploads, not the open-upload count
+	page := newMultipartUploadPage(listUploadsInput("", "", "", 3))
+	const total = 20000
+	for i := 0; i < total; i++ {
+		page.add(uploadEntry(fmt.Sprintf("%05d", i), fmt.Sprintf("k%05d", total-1-i), int64(i)))
+	}
+	if page.top.Len() != 3 {
+		t.Fatalf("collector holds %d candidates, want the 3-entry page", page.top.Len())
+	}
+	out := page.result()
+	var got []string
+	for _, u := range out.Upload {
+		got = append(got, *u.Key+"/"+*u.UploadId)
+	}
+	want := []string{"k00000/19999", "k00001/19998", "k00002/19997"}
+	if !slices.Equal(got, want) {
+		t.Fatalf("got %v, want %v", got, want)
+	}
+	if !*out.IsTruncated || *out.NextKeyMarker != "k00002" || *out.NextUploadIdMarker != "19997" {
+		t.Fatalf("truncation markers got (%v, %v, %v)", *out.IsTruncated, *out.NextKeyMarker, *out.NextUploadIdMarker)
+	}
+}
+
+func TestSelectMultipartUploadsCapsRequestedPageSize(t *testing.T) {
+	// max-uploads is caller-controlled; the page stays bounded at the server cap
+	page := newMultipartUploadPage(listUploadsInput("", "", "", 10*maxUploadsList))
+	if page.limit != maxUploadsList {
+		t.Fatalf("page limit %d, want capped at %d", page.limit, maxUploadsList)
+	}
+	if got := *page.result().MaxUploads; got != int64(maxUploadsList) {
+		t.Fatalf("MaxUploads reports %d, want the effective cap %d", got, maxUploadsList)
+	}
+}
+
+// A transient failure mid-stream replays through a fresh page, so the uploads
+// the aborted attempt already handed to the collector must not bleed into the
+// result: each upload lists once and the markers track the retried attempt.
+func TestListMultipartUploadsRetryCollectsEachUploadOnce(t *testing.T) {
+	entries := []*filer_pb.Entry{
+		uploadEntry("1", "c", 10),
+		uploadEntry("2", "a", 30),
+		uploadEntry("3", "b", 10),
+	}
+	client := &listRetryClient{
+		entries:   entries,
+		recvErrs:  []error{status.Error(codes.Unavailable, "transport is closing"), nil},
+		recvAfter: []int{2, 0},
+	}
+	accessor := &listRetryAccessor{client: client}
+	input := listUploadsInput("", "", "", 2)
+
+	var page *multipartUploadPage
+	err := listRetry("/buckets/b/.uploads", func() error {
+		page = newMultipartUploadPage(input)
+		return filer_pb.List(context.Background(), accessor, "/buckets/b/.uploads", "", func(entry *filer_pb.Entry, isLastEntry bool) error {
+			page.add(entry)
+			return nil
+		}, "", false, math.MaxInt32)
+	})
+	if err != nil {
+		t.Fatalf("listRetry: %v", err)
+	}
+	if client.attempts != 2 {
+		t.Fatalf("got %d attempts, want 2", client.attempts)
+	}
+	out := page.result()
+	var got []string
+	for _, u := range out.Upload {
+		got = append(got, *u.Key+"/"+*u.UploadId)
+	}
+	want := []string{"a/2", "b/3"}
+	if !slices.Equal(got, want) {
+		t.Fatalf("got %v, want %v (no entry from the aborted attempt may carry over)", got, want)
+	}
+	if !*out.IsTruncated || *out.NextKeyMarker != "b" || *out.NextUploadIdMarker != "3" {
+		t.Fatalf("truncation markers got (%v, %v, %v)", *out.IsTruncated, *out.NextKeyMarker, *out.NextUploadIdMarker)
+	}
+}

weed/s3api/filer_util.go

@@ -93,17 +93,17 @@ func isRetryableListError(err error) bool {
 	return err != nil && !isFilerNotFound(err) && util.IsTransientError(err)
 }
 
-// listWithRetry replays doList while the filer answers with a transient error.
+// listRetry replays doList while the filer answers with a transient error.
 // Both failure points, the ListEntries call itself and the stream.Recv that
 // follows it, surface as a plain error out of filer_pb.List, so a single retry
-// point above it covers both.
-func listWithRetry(parentDirectoryPath string, doList func() (entries []*filer_pb.Entry, isLast bool, err error)) (entries []*filer_pb.Entry, isLast bool, err error) {
+// point above it covers both. A replayed attempt may re-deliver entries the
+// last attempt already saw, so a doList that accumulates must reset first.
+func listRetry(parentDirectoryPath string, doList func() error) (err error) {
 
 	backoff := listRetryInitialBackoff
 	for attempt := 1; ; attempt++ {
-		entries, isLast, err = doList()
-		if err == nil || attempt >= listRetryAttempts || !isRetryableListError(err) {
-			return entries, isLast, err
+		if err = doList(); err == nil || attempt >= listRetryAttempts || !isRetryableListError(err) {
+			return err
 		}
 		glog.V(1).Infof("list %s attempt %d/%d hit a transient error, retrying in %v: %v", parentDirectoryPath, attempt, listRetryAttempts, backoff, err)
 		time.Sleep(backoff)
@@ -112,6 +112,16 @@ func listWithRetry(parentDirectoryPath string, doList func() (entries []*filer_p
 
 }
 
+func listWithRetry(parentDirectoryPath string, doList func() (entries []*filer_pb.Entry, isLast bool, err error)) (entries []*filer_pb.Entry, isLast bool, err error) {
+
+	err = listRetry(parentDirectoryPath, func() error {
+		entries, isLast, err = doList()
+		return err
+	})
+	return entries, isLast, err
+
+}
+
 // A delete is idempotent at the filer, which answers an entry that is already
 // gone with an empty resp.Error, so a reply the transport dropped can be
 // reissued instead of surfaced: as a 500 on the bucket delete, or as a per-key