diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index 299c9899cd..117edab974 100644 --- a/weed/s3api/filer_multipart.go +++ b/weed/s3api/filer_multipart.go @@ -1286,6 +1286,7 @@ type ListMultipartUploadsResult struct { Prefix *string `type:"string"` UploadIdMarker *string `type:"string"` Upload []*s3.MultipartUpload `locationName:"Upload" type:"list" flattened:"true"` + CommonPrefixes []PrefixEntry `xml:"CommonPrefixes,omitempty"` } func (s3a *S3ApiServer) listMultipartUploads(input *s3.ListMultipartUploadsInput) (output *ListMultipartUploadsResult, code s3err.ErrorCode) { @@ -1331,20 +1332,32 @@ func selectMultipartUploads(input *s3.ListMultipartUploadsInput, entries []*file // 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. +// prefixes dedups common prefixes still in the page — an evicted one is +// unmarked; it could only lose again, so the map stays bounded by the heap. type multipartUploadPage struct { - input *s3.ListMultipartUploadsInput - limit int - top candidateHeap - more bool + input *s3.ListMultipartUploadsInput + limit int + top candidateHeap + more bool + prefixes map[string]bool } +// multipartCandidate is an upload, or with a nil entry a common prefix that +// groups the uploads whose keys continue past the delimiter. type multipartCandidate struct { key string entry *filer_pb.Entry } +func (c multipartCandidate) uploadId() string { + if c.entry == nil { + return "" + } + return c.entry.Name +} + func compareMultipartCandidate(a, b multipartCandidate) int { - return cmp.Or(strings.Compare(a.key, b.key), strings.Compare(a.entry.Name, b.entry.Name)) + return cmp.Or(strings.Compare(a.key, b.key), strings.Compare(a.uploadId(), b.uploadId())) } // candidateHeap is a max-heap on (key, upload id): the root sorts last of the @@ -1369,7 +1382,7 @@ func newMultipartUploadPage(input *s3.ListMultipartUploadsInput) *multipartUploa if *input.MaxUploads < int64(limit) { limit = int(*input.MaxUploads) } - return &multipartUploadPage{input: input, limit: limit} + return &multipartUploadPage{input: input, limit: limit, prefixes: map[string]bool{}} } func (p *multipartUploadPage) add(entry *filer_pb.Entry) { @@ -1385,14 +1398,30 @@ func (p *multipartUploadPage) add(entry *filer_pb.Entry) { return } c := multipartCandidate{key: key, entry: entry} + // with a delimiter, keys that go on past it after the prefix are listed + // once as their common prefix, which counts toward max-uploads + if delimiter := aws.StringValue(input.Delimiter); delimiter != "" { + if i := strings.Index(key[len(*input.Prefix):], delimiter); i >= 0 { + commonPrefix := key[:len(*input.Prefix)+i+len(delimiter)] + if p.prefixes[commonPrefix] || *input.KeyMarker != "" && commonPrefix <= *input.KeyMarker { + return + } + c = multipartCandidate{key: commonPrefix} + } + } if p.top.Len() >= p.limit { p.more = true if compareMultipartCandidate(c, p.top[0]) >= 0 { return } - heap.Pop(&p.top) + if evicted := heap.Pop(&p.top).(multipartCandidate); evicted.entry == nil { + delete(p.prefixes, evicted.key) + } } heap.Push(&p.top, c) + if c.entry == nil { + p.prefixes[c.key] = true + } } func (p *multipartUploadPage) result() *ListMultipartUploadsResult { @@ -1410,6 +1439,10 @@ func (p *multipartUploadPage) result() *ListMultipartUploadsResult { copy(candidates, p.top) slices.SortFunc(candidates, compareMultipartCandidate) for _, c := range candidates { + if c.entry == nil { + output.CommonPrefixes = append(output.CommonPrefixes, PrefixEntry{Prefix: c.key}) + continue + } upload := &s3.MultipartUpload{ Key: aws.String(c.key), UploadId: aws.String(c.entry.Name), @@ -1422,7 +1455,10 @@ func (p *multipartUploadPage) result() *ListMultipartUploadsResult { if p.more { last := candidates[len(candidates)-1] output.IsTruncated = aws.Bool(true) - output.NextKeyMarker, output.NextUploadIdMarker = aws.String(last.key), aws.String(last.entry.Name) + output.NextKeyMarker = aws.String(last.key) + if last.entry != nil { + output.NextUploadIdMarker = aws.String(last.entry.Name) + } } if aws.StringValue(input.EncodingType) == s3.EncodingTypeUrl { // AWS url-encodes these key-named elements when encoding-type=url is @@ -1436,6 +1472,9 @@ func (p *multipartUploadPage) result() *ListMultipartUploadsResult { for _, upload := range output.Upload { upload.Key = encodeMultipartUploadURLField(upload.Key) } + for i := range output.CommonPrefixes { + output.CommonPrefixes[i].Prefix = urlQueryEscape(output.CommonPrefixes[i].Prefix) + } } else { output.KeyMarker = encodeListContinuationTokenField(output.KeyMarker) output.NextKeyMarker = encodeListContinuationTokenField(output.NextKeyMarker) diff --git a/weed/s3api/filer_multipart_list_test.go b/weed/s3api/filer_multipart_list_test.go index 3d2a73789c..6e1b5c1dab 100644 --- a/weed/s3api/filer_multipart_list_test.go +++ b/weed/s3api/filer_multipart_list_test.go @@ -261,3 +261,71 @@ func TestListMultipartUploadsRetryCollectsEachUploadOnce(t *testing.T) { t.Fatalf("truncation markers got (%v, %v, %v)", *out.IsTruncated, *out.NextKeyMarker, *out.NextUploadIdMarker) } } + +// With a delimiter, uploads whose keys go on past it after the prefix are +// listed once under CommonPrefixes, which count toward max-uploads. +func TestSelectMultipartUploadsDelimiter(t *testing.T) { + entries := []*filer_pb.Entry{ + uploadEntry("1", "a/1", 10), + uploadEntry("2", "a/2", 10), + uploadEntry("3", "b/x/1", 10), + uploadEntry("4", "c", 10), + uploadEntry("5", "a/sub/3", 10), + } + list := func(prefix, keyMarker string, maxUploads int64) ([]string, []string, bool, string) { + input := listUploadsInput(prefix, keyMarker, "", maxUploads) + input.Delimiter = aws.String("/") + out := selectMultipartUploads(input, entries) + var uploads, prefixes []string + for _, u := range out.Upload { + uploads = append(uploads, *u.Key) + } + for _, p := range out.CommonPrefixes { + prefixes = append(prefixes, p.Prefix) + } + return uploads, prefixes, *out.IsTruncated, aws.StringValue(out.NextKeyMarker) + } + + uploads, prefixes, truncated, _ := list("", "", 1000) + if !slices.Equal(uploads, []string{"c"}) || !slices.Equal(prefixes, []string{"a/", "b/"}) || truncated { + t.Errorf("no prefix: uploads %v prefixes %v truncated %v", uploads, prefixes, truncated) + } + + uploads, prefixes, _, _ = list("a/", "", 1000) + if !slices.Equal(uploads, []string{"a/1", "a/2"}) || !slices.Equal(prefixes, []string{"a/sub/"}) { + t.Errorf("prefix a/: uploads %v prefixes %v", uploads, prefixes) + } + + // a page that ends on a common prefix resumes after it + uploads, prefixes, truncated, next := list("", "", 1) + if len(uploads) != 0 || !slices.Equal(prefixes, []string{"a/"}) || !truncated || next != "a/" { + t.Fatalf("page 1: uploads %v prefixes %v truncated %v next %q", uploads, prefixes, truncated, next) + } + uploads, prefixes, truncated, _ = list("", next, 1000) + if !slices.Equal(uploads, []string{"c"}) || !slices.Equal(prefixes, []string{"b/"}) || truncated { + t.Errorf("page 2: uploads %v prefixes %v truncated %v", uploads, prefixes, truncated) + } +} + +// A common prefix that does not fit the page is not retained: prefixes tracks +// only groups still in the page, so a small page stays bounded by max-uploads +// rather than by every prefix in the bucket. +func TestSelectMultipartUploadsDelimiterDropsLosingPrefixes(t *testing.T) { + input := listUploadsInput("", "", "", 1) + input.Delimiter = aws.String("/") + page := newMultipartUploadPage(input) + page.add(uploadEntry("1", "z/1", 10)) // holds the single slot, then evicted + page.add(uploadEntry("2", "a/1", 10)) + page.add(uploadEntry("3", "z/2", 10)) // "z/" was evicted: loses again + page.add(uploadEntry("4", "z/3", 10)) + if len(page.prefixes) > page.limit { + t.Fatalf("retained %d prefixes with limit %d", len(page.prefixes), page.limit) + } + out := page.result() + if len(out.CommonPrefixes) != 1 || out.CommonPrefixes[0].Prefix != "a/" || !*out.IsTruncated { + t.Fatalf("prefixes %v truncated %v", out.CommonPrefixes, *out.IsTruncated) + } + if out.NextUploadIdMarker != nil { + t.Fatalf("page ending on a common prefix has upload id marker %v", *out.NextUploadIdMarker) + } +}