diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index 5cb0e8a833..6f22429fb6 100644 --- a/weed/s3api/filer_multipart.go +++ b/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 } diff --git a/weed/s3api/filer_multipart_list_test.go b/weed/s3api/filer_multipart_list_test.go index 0ed258fd71..c00f38acaf 100644 --- a/weed/s3api/filer_multipart_list_test.go +++ b/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) + } +} diff --git a/weed/s3api/filer_util.go b/weed/s3api/filer_util.go index 5ebffe8f83..3ec242a4d3 100644 --- a/weed/s3api/filer_util.go +++ b/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