From 60b06b688170383a8acfe756cfca5c0d6c021f61 Mon Sep 17 00:00:00 2001 From: Jay Hemnani <193022578+jayhemnani9910@users.noreply.github.com> Date: Tue, 20 Oct 2026 06:40:06 +0530 Subject: [PATCH] s3: sort ListMultipartUploads by key and honor key-marker (#11833) * s3: sort ListMultipartUploads by key and honor key-marker * s3: skip upload folders without an object key in ListMultipartUploads * s3: order same-key multipart uploads by initiation time The upload id suffix is now a time-ordered UUIDv7, so sorting by upload id lists a key's uploads oldest first, like S3. Pagination is unchanged: ordering and the upload-id-marker comparison share the same key. selectMultipartUploads now builds the whole result, including NextKeyMarker and NextUploadIdMarker, and the pagination test pages with the response markers. --------- Co-authored-by: Chris Lu --- weed/s3api/filer_multipart.go | 96 ++++++++++++-------- weed/s3api/filer_multipart_list_test.go | 114 ++++++++++++++++++++++++ 2 files changed, 173 insertions(+), 37 deletions(-) create mode 100644 weed/s3api/filer_multipart_list_test.go diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index 06fe3091b3..5cb0e8a833 100644 --- a/weed/s3api/filer_multipart.go +++ b/weed/s3api/filer_multipart.go @@ -69,7 +69,8 @@ func (s3a *S3ApiServer) createMultipartUpload(r *http.Request, input *s3.CreateM uploadIdString := s3a.generateUploadID(*input.Key) - uploadIdString = uploadIdString + "_" + strings.ReplaceAll(uuid.New().String(), "-", "") + // a time-ordered uuid keeps uploads of one key in initiation order + uploadIdString = uploadIdString + "_" + strings.ReplaceAll(uuid.Must(uuid.NewV7()).String(), "-", "") // Validate checksum algorithm before creating the upload directory checksumAlgo, checksumHeaderName, checksumErrCode := detectRequestedChecksumAlgorithm(r) @@ -1291,56 +1292,77 @@ func (s3a *S3ApiServer) listMultipartUploads(input *s3.ListMultipartUploadsInput glog.V(2).Infof("listMultipartUploads input %v", input) - output = &ListMultipartUploadsResult{ - Bucket: input.Bucket, - Delimiter: input.Delimiter, - EncodingType: input.EncodingType, - KeyMarker: input.KeyMarker, - MaxUploads: input.MaxUploads, - Prefix: input.Prefix, - IsTruncated: aws.Bool(false), - } - - entries, _, err := s3a.list(s3a.genUploadsFolder(*input.Bucket), "", *input.UploadIdMarker, false, math.MaxInt32) + entries, _, err := s3a.list(s3a.genUploadsFolder(*input.Bucket), "", "", 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. if isFilerNotFound(err) { - return output, s3err.ErrNone + return selectMultipartUploads(input, nil), s3err.ErrNone } // surface a real store error; a masked empty 200 makes a resuming client treat the upload as gone glog.Errorf("listMultipartUploads %s error: %v", *input.Bucket, err) - return output, s3err.ErrInternalError + return nil, s3err.ErrInternalError } - uploadsCount := int64(0) + return selectMultipartUploads(input, entries), 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 { + output := &ListMultipartUploadsResult{ + Bucket: input.Bucket, + Delimiter: input.Delimiter, + EncodingType: input.EncodingType, + KeyMarker: input.KeyMarker, + MaxUploads: input.MaxUploads, + 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 entry.Extended != nil { - key := string(entry.Extended[s3_constants.ExtMultipartObjectKey]) - if *input.KeyMarker != "" && *input.KeyMarker != key { - continue - } - if *input.Prefix != "" && !strings.HasPrefix(key, *input.Prefix) { - continue - } - upload := &s3.MultipartUpload{ - Key: objectKey(aws.String(key)), - UploadId: aws.String(entry.Name), - } - if entry.Attributes != nil { - upload.Initiated = aws.Time(time.Unix(entry.Attributes.Crtime, int64(entry.Attributes.CrtimeNs))) - } - output.Upload = append(output.Upload, upload) - uploadsCount += 1 + if !isMultipartUploadEntry(entry) { + continue } - if uploadsCount >= *input.MaxUploads { - output.IsTruncated = aws.Bool(true) - output.NextUploadIdMarker = aws.String(entry.Name) - break + 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)) + }) - return + 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), + } + if c.entry.Attributes != nil { + upload.Initiated = aws.Time(time.Unix(c.entry.Attributes.Crtime, int64(c.entry.Attributes.CrtimeNs))) + } + output.Upload = append(output.Upload, upload) + } + return output } type ListPartsResult struct { diff --git a/weed/s3api/filer_multipart_list_test.go b/weed/s3api/filer_multipart_list_test.go new file mode 100644 index 0000000000..0ed258fd71 --- /dev/null +++ b/weed/s3api/filer_multipart_list_test.go @@ -0,0 +1,114 @@ +package s3api + +import ( + "slices" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/s3" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" +) + +// uploadEntry is an upload folder as createMultipartUpload leaves it. +func uploadEntry(id, key string, crtime int64) *filer_pb.Entry { + return &filer_pb.Entry{ + Name: id, + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtMultipartObjectKey: []byte(key)}, + Attributes: &filer_pb.FuseAttributes{Crtime: crtime}, + } +} + +func listUploadsInput(prefix, keyMarker, uploadIdMarker string, maxUploads int64) *s3.ListMultipartUploadsInput { + return &s3.ListMultipartUploadsInput{ + Prefix: aws.String(prefix), + KeyMarker: aws.String(keyMarker), + UploadIdMarker: aws.String(uploadIdMarker), + MaxUploads: aws.Int64(maxUploads), + } +} + +func listedUploads(entries []*filer_pb.Entry, prefix, keyMarker, uploadIdMarker string, maxUploads int64) ([]string, bool) { + out := selectMultipartUploads(listUploadsInput(prefix, keyMarker, uploadIdMarker, maxUploads), entries) + var got []string + for _, u := range out.Upload { + got = append(got, *u.Key+"/"+*u.UploadId) + } + return got, *out.IsTruncated +} + +func TestSelectMultipartUploads(t *testing.T) { + // listed from the filer in upload id order, which is not key order + entries := []*filer_pb.Entry{ + uploadEntry("1", "c", 10), + uploadEntry("2", "a", 30), + uploadEntry("3", "b", 10), + uploadEntry("4", "a", 20), + uploadEntry("5", "up/x", 10), + // a folder with Extended set but no object key is not an upload + {Name: "6", IsDirectory: true, Extended: map[string][]byte{}}, + } + + tests := []struct { + name string + prefix string + keyMarker string + uploadIdMarker string + maxUploads int64 + want []string + truncated bool + }{ + {name: "sorted by key, then by upload id", maxUploads: 1000, want: []string{"a/2", "a/4", "b/3", "c/1", "up/x/5"}}, + {name: "key-marker starts after that key", keyMarker: "a", maxUploads: 1000, want: []string{"b/3", "c/1", "up/x/5"}}, + {name: "key-marker between keys", keyMarker: "bb", maxUploads: 1000, want: []string{"c/1", "up/x/5"}}, + {name: "upload-id-marker continues inside its key", keyMarker: "a", uploadIdMarker: "2", maxUploads: 1000, want: []string{"a/4", "b/3", "c/1", "up/x/5"}}, + {name: "upload-id-marker of an aborted upload", keyMarker: "a", uploadIdMarker: "3", maxUploads: 1000, want: []string{"a/4", "b/3", "c/1", "up/x/5"}}, + {name: "upload-id-marker without key-marker is ignored", uploadIdMarker: "4", maxUploads: 1000, want: []string{"a/2", "a/4", "b/3", "c/1", "up/x/5"}}, + {name: "prefix", prefix: "up/", maxUploads: 1000, want: []string{"up/x/5"}}, + {name: "page", maxUploads: 2, want: []string{"a/2", "a/4"}, truncated: true}, + {name: "exact last page is not truncated", keyMarker: "b", maxUploads: 2, want: []string{"c/1", "up/x/5"}}, + {name: "max-uploads 0", maxUploads: 0, want: nil}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, truncated := listedUploads(entries, tt.prefix, tt.keyMarker, tt.uploadIdMarker, tt.maxUploads) + if !slices.Equal(got, tt.want) || truncated != tt.truncated { + t.Fatalf("got %v truncated=%v, want %v truncated=%v", got, truncated, tt.want, tt.truncated) + } + }) + } +} + +func TestSelectMultipartUploadsPaginatesEveryUploadOnce(t *testing.T) { + var entries []*filer_pb.Entry + for i, key := range []string{"k3", "k1", "k2", "k1", "k3", "k1", "k2"} { + entries = append(entries, uploadEntry(string(rune('a'+i)), key, int64(i))) + } + seen := map[string]bool{} + input := listUploadsInput("", "", "", 2) + pages := 0 + for { + out := selectMultipartUploads(input, entries) + for _, u := range out.Upload { + if seen[*u.UploadId] { + t.Fatalf("upload %s listed twice", *u.UploadId) + } + seen[*u.UploadId] = true + } + pages++ + if !*out.IsTruncated { + break + } + if out.NextKeyMarker == nil || out.NextUploadIdMarker == nil { + t.Fatalf("truncated page without next markers") + } + input.KeyMarker, input.UploadIdMarker = out.NextKeyMarker, out.NextUploadIdMarker + } + if pages != 4 { + t.Fatalf("got %d pages, want 4", pages) + } + if len(seen) != len(entries) { + t.Fatalf("listed %d of %d uploads", len(seen), len(entries)) + } +}