s3: group ListMultipartUploads by delimiter into CommonPrefixes (#12069)

* s3: group ListMultipartUploads by delimiter into CommonPrefixes

The delimiter was echoed but ignored, so every upload was listed. Keys
that continue past the delimiter after the prefix are now listed once as
a common prefix, counted toward max-uploads like an upload.

* s3: bound tracked multipart common prefixes to the page

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

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

* s3: omit upload id marker when a page ends on a common prefix

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

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

---------

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

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)

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)
+	}
+}