fix(multipart): defer persistent listing from public requests

Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
Feng Ruohang
2026-10-18 08:22:31 +08:00
parent 4189869e43
commit 4733e1423c
12 changed files with 359 additions and 141 deletions
+1 -1
View File
@@ -883,7 +883,7 @@ func serverHandleEnvVars() {
}
globalInplaceUpdateDisabled = true
if env.IsSet(api.EnvAPIMultipartListing) {
logger.Warning("%s is ignored: multipart listing always uses persistent metadata; uploads without identity fields are omitted", api.EnvAPIMultipartListing)
logger.Warning("%s is ignored: multipart listing uses the legacy cache/exact-key behavior; persistent listing is deferred", api.EnvAPIMultipartListing)
}
// Check if the supported credential env vars,
@@ -14,7 +14,7 @@ import (
// Run explicitly with -benchtime=1x. Fixture creation is excluded from timing;
// each request still discovers real on-disk metadata. The 16-drive/10k case
// deliberately exceeds the existing aggregate directory-entry budget.
func BenchmarkMultipartListingCapacity(b *testing.B) {
func BenchmarkMultipartDiagnosticScanCapacity(b *testing.B) {
for _, tc := range []struct {
drives, uploads int
mixed, limited bool
@@ -67,7 +67,7 @@ func BenchmarkMultipartListingCapacity(b *testing.B) {
}
b.ResetTimer()
for range b.N {
got, err := z.ListMultipartUploads(b.Context(), bucket, "dir/", "", "", "", 1000)
got, err := z.scanMultipartUploadsForTest(b.Context(), bucket, "dir/", "", "", "", 1000)
if tc.limited {
var limited SlowDown
if !errors.As(err, &limited) || len(got.Uploads) != 0 {
@@ -83,3 +83,88 @@ func BenchmarkMultipartListingCapacity(b *testing.B) {
})
}
}
// The public path is measured with a populated cache as well as after cache
// loss. An empty post-restart response alone is not capacity acceptance.
func BenchmarkMultipartPublicListingCapacity(b *testing.B) {
for _, tc := range []struct {
drives, uploads int
sameKey bool
}{
{4, 10_000, false},
{16, 3_126, false},
{16, 10_000, true},
} {
b.Run(fmt.Sprintf("drives=%d/uploads=%d/same-key=%t", tc.drives, tc.uploads, tc.sameKey), func(b *testing.B) {
obj, dirs, err := prepareErasure(b.Context(), tc.drives)
if err != nil {
b.Fatal(err)
}
z := obj.(*erasureServerPools)
b.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) })
const bucket, other = "multipart-public-capacity", "multipart-public-other"
for _, name := range []string{bucket, other} {
if err := z.MakeBucket(b.Context(), name, MakeBucketOptions{}); err != nil {
b.Fatal(err)
}
}
wantCached, wantExact := 0, 1
for n := range tc.uploads {
target, key := bucket, fmt.Sprintf("dir/%05d", n)
if tc.sameKey {
key = "dir/00000"
} else if n%2 != 0 {
target = other
}
if target == bucket {
wantCached++
}
if _, err := z.NewMultipartUpload(b.Context(), target, key, ObjectOptions{}); err != nil {
b.Fatal(err)
}
}
if tc.sameKey {
wantExact = 1000
}
if tc.drives == 16 && !tc.sameKey {
_, err := z.scanMultipartUploadsForTest(b.Context(), bucket, "", "", "", "", 1000)
var limited SlowDown
if !errors.As(err, &limited) {
b.Fatalf("fixture did not reproduce the previous global budget failure: %v", err)
}
}
var directoryCalls, metadataCalls atomic.Int64
for _, set := range z.serverPools[0].sets {
original := set.getDisks
set.getDisks = func() []StorageAPI {
disks := append([]StorageAPI(nil), original()...)
for i, disk := range disks {
disks[i] = multipartListingCountingDisk{disk, &directoryCalls, &metadataCalls}
}
return disks
}
b.Cleanup(func() { set.getDisks = original })
}
measure := func(name, prefix string, want int) {
b.Run(name, func(b *testing.B) {
directoryCalls.Store(0)
metadataCalls.Store(0)
b.ResetTimer()
for range b.N {
got, err := z.ListMultipartUploads(b.Context(), bucket, prefix, "", "", "", 1000)
if err != nil || len(got.Uploads) != want {
b.Fatalf("public listing: count=%d want=%d err=%v", len(got.Uploads), want, err)
}
}
b.StopTimer()
b.ReportMetric(float64(directoryCalls.Load())/float64(b.N), "ListDir/op")
b.ReportMetric(float64(metadataCalls.Load())/float64(b.N), "ReadVersion/op")
})
}
measure("populated-cache", "", wantCached)
z.mpCache.Clear()
measure("lost-cache", "", 0)
measure("known-key", "dir/00000", wantExact)
})
}
}
+1 -1
View File
@@ -359,7 +359,7 @@ func (z *erasureServerPools) multipartPreflight(ctx context.Context) (multipartP
return multipartPreflightReport{}, err
}
defer s.close()
report := multipartPreflightReport{Complete: true, Mode: "persistent"}
report := multipartPreflightReport{Complete: true, Mode: "legacy"}
for p, pool := range z.serverPools {
for i, set := range pool.sets {
_, _ = set.scanMultipartUploads(s, "", p, i)
+71 -28
View File
@@ -22,10 +22,53 @@ import (
"github.com/minio/minio/internal/config/storageclass"
)
// Check the real handler and storage path, including a successful abort between
// pages. Choose IDs increasing in both initiation time and lexical order so the
// failure does not depend on differing interpretations of S3 marker ordering.
func TestMultipartListingAbortBetweenHTTPPages(t *testing.T) {
// The persistent scanner is retained for diagnostics. This test-only wrapper
// preserves its pagination coverage without exposing it through public listing.
func (z *erasureServerPools) scanMultipartUploadsForTest(ctx context.Context, bucket, prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (ListMultipartsInfo, error) {
if err := checkListMultipartArgs(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter); err != nil {
return ListMultipartsInfo{}, err
}
if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil {
return ListMultipartsInfo{}, toObjectErr(err, bucket)
}
scan, err := startMultipartScan(ctx, false)
if err != nil {
return ListMultipartsInfo{}, err
}
defer scan.close()
var uploads []MultipartInfo
for idx, pool := range z.serverPools {
if z.IsSuspended(idx) {
continue
}
poolUploads, err := pool.scanMultipartUploadsForTest(scan, bucket, idx)
if err != nil {
return ListMultipartsInfo{}, err
}
uploads = append(uploads, poolUploads...)
}
// Confirmed historical records lack a recoverable bucket/key and are
// omitted by the scan. They do not block listing identifiable uploads.
return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil
}
func (s *erasureSets) scanMultipartUploadsForTest(scan *multipartScan, bucket string, poolIdx int) ([]MultipartInfo, error) {
var uploads []MultipartInfo
for i, set := range s.sets {
setUploads, err := set.scanMultipartUploads(scan, bucket, poolIdx, i)
if err != nil {
return nil, err
}
uploads = append(uploads, setUploads...)
}
return uploads, nil
}
// Check that a successful HTTP abort removes the cached upload. The restored
// cache path still ignores page limits; this is a documented legacy limitation.
func TestMultipartListingAbortHTTPUpdatesCache(t *testing.T) {
z, _ := consistencyPools(t)
bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads", "AbortMultipart"}, MakeBucketOptions{})
if err != nil {
@@ -81,8 +124,8 @@ func TestMultipartListingAbortBetweenHTTPPages(t *testing.T) {
return result
}
first := list("", "", 1)
if len(first.Uploads) != 1 || first.Uploads[0].UploadID != firstID || !first.IsTruncated {
t.Fatalf("unexpected first page: %+v", first)
if len(first.Uploads) != 3 || first.Uploads[0].UploadID != firstID || first.IsTruncated {
t.Fatalf("unexpected legacy cache view: %+v", first)
}
rec := request(http.MethodDelete, getAbortMultipartUploadURL("", bucket, "a", firstID))
if rec.Code != http.StatusNoContent {
@@ -93,12 +136,12 @@ func TestMultipartListingAbortBetweenHTTPPages(t *testing.T) {
for _, upload := range rest.Uploads {
keys = append(keys, upload.Key)
}
t.Logf("GET first page=200; DELETE marker=204; GET next page=200 keys=%v truncated=%v", keys, rest.IsTruncated)
t.Logf("GET cache=200; DELETE upload=204; GET cache=200 keys=%v truncated=%v", keys, rest.IsTruncated)
if _, err := z.GetMultipartInfo(t.Context(), bucket, "a", secondID, ObjectOptions{}); err != nil {
t.Fatalf("remaining upload is not valid: %v", err)
}
if len(rest.Uploads) != 2 || rest.Uploads[0].UploadID != secondID {
t.Fatalf("valid remaining upload for key a is missing from continuation: %+v", rest.Uploads)
t.Fatalf("valid remaining upload for key a is missing from cache: %+v", rest.Uploads)
}
}
@@ -149,7 +192,7 @@ func TestMultipartListingLegacyPreflight(t *testing.T) {
}
z.mpCache.Clear()
for _, target := range []string{bucket, other} {
got, err := z.ListMultipartUploads(t.Context(), target, "", "", "", "", 10)
got, err := z.scanMultipartUploadsForTest(t.Context(), target, "", "", "", "", 10)
if err != nil {
t.Fatal(err)
}
@@ -159,14 +202,14 @@ func TestMultipartListingLegacyPreflight(t *testing.T) {
requireMultipartUploadKeys(t, got)
}
}
// Exact-looking prefixes now have the same durable semantics.
exact, err := z.ListMultipartUploads(t.Context(), other, "old", "", "", "", 10)
// Diagnostic discovery cannot recover an unidentified historical key.
exact, err := z.scanMultipartUploadsForTest(t.Context(), other, "old", "", "", "", 10)
if err != nil {
t.Fatal(err)
}
requireMultipartUploadKeys(t, exact)
report, err := z.multipartPreflight(t.Context())
if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "persistent" {
if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "legacy" {
t.Fatalf("legacy preflight: %+v %v", report, err)
}
if err = z.AbortMultipartUpload(t.Context(), other, "old", old.UploadID, ObjectOptions{}); err != nil {
@@ -176,14 +219,14 @@ func TestMultipartListingLegacyPreflight(t *testing.T) {
if err != nil || !report.Ready || report.LegacyUploads != 0 {
t.Fatalf("drained preflight: %+v %v", report, err)
}
got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
if err != nil {
t.Fatal(err)
}
requireMultipartUploadKeys(t, got, "a")
}
func TestMultipartListingIdentityFallback(t *testing.T) {
func TestMultipartDiagnosticIdentityFallback(t *testing.T) {
for _, kind := range []string{"old", "partial", "corrupt", "read-failure", "wrong-bucket"} {
t.Run(kind, func(t *testing.T) {
z, set, bucket := multipartListingFixture(t)
@@ -220,7 +263,7 @@ func TestMultipartListingIdentityFallback(t *testing.T) {
}
return disks
}
got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
if err != nil {
t.Fatal(err)
}
@@ -469,7 +512,7 @@ func (d multipartListingCountingDisk) ReadVersion(ctx context.Context, original,
}
// Counts storage API calls; this is not a deployment throughput benchmark.
func TestMultipartListingScanCosts(t *testing.T) {
func TestMultipartDiagnosticScanCosts(t *testing.T) {
obj, dirs, err := prepareErasure(t.Context(), 4)
if err != nil {
t.Fatal(err)
@@ -507,7 +550,7 @@ func TestMultipartListingScanCosts(t *testing.T) {
t.Helper()
directoryCalls.Store(0)
metadataCalls.Store(0)
result, err := z.ListMultipartUploads(t.Context(), bucket, "dir/", "", "", "", 1)
result, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "dir/", "", "", "", 1)
if err != nil {
t.Fatal(err)
}
@@ -589,7 +632,7 @@ func TestMultipartListingMissingMarkers(t *testing.T) {
}
}
func TestMultipartListingCoverage(t *testing.T) {
func TestMultipartDiagnosticCoverage(t *testing.T) {
z, set, bucket := multipartListingFixture(t)
mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
if err != nil {
@@ -605,12 +648,12 @@ func TestMultipartListingCoverage(t *testing.T) {
}
}
set.getDisks = func() []StorageAPI { return []StorageAPI{disks[0], disks[1], nil, nil} }
_, err = z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
_, err = z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 {
t.Fatalf("incomplete discovery returned success: %v", err)
}
set.getDisks = func() []StorageAPI { return []StorageAPI{nil, disks[1], disks[2], disks[3]} }
got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
if err != nil {
t.Fatal(err)
}
@@ -621,7 +664,7 @@ func TestMultipartListingCoverage(t *testing.T) {
}
}
func TestMultipartListingBudgetAndAdmission(t *testing.T) {
func TestMultipartDiagnosticBudgetAndAdmission(t *testing.T) {
_, set, bucket := multipartListingFixture(t)
if _, err := set.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}); err != nil {
t.Fatal(err)
@@ -651,7 +694,7 @@ func TestMultipartListingBudgetAndAdmission(t *testing.T) {
}
}
func TestMultipartListingAdmissionHTTP(t *testing.T) {
func TestMultipartPublicListingAdmissionHTTP(t *testing.T) {
z, _, _ := multipartListingFixture(t)
bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads"}, MakeBucketOptions{})
if err != nil {
@@ -673,12 +716,12 @@ func TestMultipartListingAdmissionHTTP(t *testing.T) {
}
rec := httptest.NewRecorder()
router.ServeHTTP(rec, req)
var response APIErrorResponse
var response ListMultipartUploadsResponse
if err := xml.Unmarshal(rec.Body.Bytes(), &response); err != nil {
t.Fatal(err)
}
if rec.Code != http.StatusServiceUnavailable || response.Code != "SlowDown" {
t.Fatalf("admission returned %d %s", rec.Code, rec.Body.String())
if rec.Code != http.StatusOK || len(response.Uploads) != 0 {
t.Fatalf("diagnostic admission blocked public listing: %d %s", rec.Code, rec.Body.String())
}
}
@@ -705,12 +748,12 @@ func TestMultipartListingPreflightMinorityLegacy(t *testing.T) {
}
}
report, err := z.multipartPreflight(t.Context())
if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "persistent" {
if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "legacy" {
t.Fatalf("minority legacy copy must prevent readiness: %+v %v", report, err)
}
}
func TestMultipartListingCancellationRetainsAdmission(t *testing.T) {
func TestMultipartDiagnosticCancellationRetainsAdmission(t *testing.T) {
z, set, bucket := multipartListingFixture(t)
for i := range 40 {
if _, err := z.NewMultipartUpload(t.Context(), bucket, fmt.Sprintf("key-%02d", i), ObjectOptions{}); err != nil {
@@ -749,7 +792,7 @@ func TestMultipartListingCancellationRetainsAdmission(t *testing.T) {
defer cancel()
done := make(chan error, 1)
go func() {
_, err := z.ListMultipartUploads(ctx, bucket, "", "", "", "", 10)
_, err := z.scanMultipartUploadsForTest(ctx, bucket, "", "", "", "", 10)
done <- err
}()
for range 16 {
+106
View File
@@ -0,0 +1,106 @@
// Copyright (c) 2026 Ruohang Feng
// SPDX-License-Identifier: AGPL-3.0-or-later
package cmd
import (
"context"
"encoding/xml"
"net/http"
"net/http/httptest"
"sync/atomic"
"testing"
)
type multipartRootScanGuard struct {
StorageAPI
roots *atomic.Int64
}
func (d multipartRootScanGuard) ListDir(ctx context.Context, original, volume, dir string, count int) ([]string, error) {
if volume == minioMetaMultipartBucket && dir == "" {
d.roots.Add(1)
return nil, errFaultyDisk
}
return d.StorageAPI.ListDir(ctx, original, volume, dir, count)
}
// A busy or unreadable diagnostic namespace must not become a prerequisite for
// ordinary listing, including the lower-level exact-object entry points.
func TestMultipartPublicListingIndependentOfDiagnostics(t *testing.T) {
for _, busy := range []bool{false, true} {
t.Run(map[bool]string{false: "unreadable-root", true: "busy-diagnostics"}[busy], func(t *testing.T) {
z, _ := consistencyPools(t)
bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads"}, MakeBucketOptions{})
if err != nil {
t.Fatal(err)
}
const key = "dir/known"
mp, err := z.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{})
if err != nil {
t.Fatal(err)
}
var roots atomic.Int64
for _, pool := range z.serverPools {
for _, set := range pool.sets {
original := set.getDisks
set.getDisks = func() []StorageAPI {
disks := append([]StorageAPI(nil), original()...)
for i, disk := range disks {
if disk != nil {
disks[i] = multipartRootScanGuard{disk, &roots}
}
}
return disks
}
t.Cleanup(func() { set.getDisks = original })
}
}
if busy {
for range cap(multipartScanSlots) {
scan, err := startMultipartScan(t.Context(), true)
if err != nil {
t.Fatal(err)
}
defer scan.close()
}
}
for _, prefix := range []string{"", key, "dir/"} {
req, err := newTestSignedRequestV4(http.MethodGet,
getListMultipartUploadsURLWithParams("", bucket, prefix, "", "", "", "1"),
0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
if err != nil {
t.Fatal(err)
}
rec := httptest.NewRecorder()
router.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("prefix %q: HTTP %d %s", prefix, rec.Code, rec.Body.String())
}
var got ListMultipartUploadsResponse
if err := xml.Unmarshal(rec.Body.Bytes(), &got); err != nil {
t.Fatal(err)
}
if prefix == "dir/" {
if len(got.Uploads) != 0 {
t.Fatalf("partial prefix must retain the documented exact-key limitation: %+v", got)
}
} else if len(got.Uploads) != 1 || got.Uploads[0].UploadID != mp.UploadID {
t.Fatalf("known upload missing: %+v", got)
}
}
for _, pool := range z.serverPools {
for _, list := range []func(context.Context, string, string, string, string, string, int) (ListMultipartsInfo, error){
pool.ListMultipartUploads, pool.getHashedSet(key).ListMultipartUploads,
} {
if _, err := list(t.Context(), bucket, key, "", "", "", 10); err != nil {
t.Fatalf("lower-level exact listing: %v", err)
}
}
}
if roots.Load() != 0 {
t.Fatalf("public listing attempted %d global multipart scans", roots.Load())
}
})
}
}
+30 -30
View File
@@ -4,11 +4,10 @@
package cmd
import (
"bytes"
"context"
"encoding/xml"
"errors"
"net/http"
"net/http/httptest"
"slices"
"testing"
)
@@ -33,14 +32,10 @@ func removeMultipartIdentity(metadata map[string]string) {
delete(metadata, multipartMetaObject)
}
// The actual S3 handler must use the same semantics for every prefix and page,
// even with old records and a stale, misleading node-local cache.
func TestMultipartListingMixedHTTP(t *testing.T) {
z, set, _ := multipartListingFixture(t)
bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads"}, MakeBucketOptions{})
if err != nil {
t.Fatal(err)
}
// The retained diagnostic scanner handles mixed records independently of a
// stale cache. Public HTTP behavior is tested separately.
func TestMultipartDiagnosticMixed(t *testing.T) {
z, set, bucket := multipartListingFixture(t)
for _, key := range []string{"dir/old", "dir/a", "dir/sub/b", "outside"} {
mp, err := z.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{})
if err != nil {
@@ -54,22 +49,11 @@ func TestMultipartListingMixedHTTP(t *testing.T) {
z.mpCache.Store("stale", MultipartInfo{Bucket: bucket, Object: "ghost", UploadID: "stale"})
list := func(prefix, keyMarker, uploadMarker, delimiter string) ListMultipartUploadsResponse {
t.Helper()
req, err := newTestSignedRequestV4(http.MethodGet,
getListMultipartUploadsURLWithParams("", bucket, prefix, keyMarker, uploadMarker, delimiter, "1"),
0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, prefix, keyMarker, uploadMarker, delimiter, 1)
if err != nil {
t.Fatal(err)
}
rec := httptest.NewRecorder()
router.ServeHTTP(rec, req)
var got ListMultipartUploadsResponse
if rec.Code != http.StatusOK {
t.Fatalf("list %q: HTTP %d %s", prefix, rec.Code, rec.Body.String())
}
if err := xml.Unmarshal(rec.Body.Bytes(), &got); err != nil {
t.Fatal(err)
}
return got
return generateListMultipartUploadsResponse(bucket, got, "")
}
var keys []string
keyMarker, uploadMarker := "", ""
@@ -103,7 +87,7 @@ func TestMultipartListingMixedHTTP(t *testing.T) {
}
}
func TestMultipartListingInvalidIdentity(t *testing.T) {
func TestMultipartDiagnosticInvalidIdentity(t *testing.T) {
for _, kind := range []string{"missing-bucket", "missing-object", "empty-bucket", "empty-object", "empty-both", "wrong-hash", "read-failure"} {
t.Run(kind, func(t *testing.T) {
z, set, bucket := multipartListingFixture(t)
@@ -142,7 +126,7 @@ func TestMultipartListingInvalidIdentity(t *testing.T) {
return disks
}
}
got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
if err == nil || len(got.Uploads) != 0 || toAPIError(t.Context(), err).HTTPStatusCode != http.StatusServiceUnavailable {
t.Fatalf("invalid/unreadable record returned a successful page: %+v %v", got, err)
}
@@ -185,7 +169,14 @@ func TestMultipartListingRestart(t *testing.T) {
if _, cached := obj.(*erasureServerPools).mpCache.Load(mp.UploadID); cached {
t.Fatal("restart unexpectedly preserved the upload cache")
}
got, err := obj.ListMultipartUploads(t.Context(), bucket, "dir/", "", "", "", 10)
for _, prefix := range []string{"", "dir/"} {
got, err := obj.ListMultipartUploads(t.Context(), bucket, prefix, "", "", "", 10)
if err != nil {
t.Fatal(err)
}
requireMultipartUploadKeys(t, got)
}
got, err := obj.ListMultipartUploads(t.Context(), bucket, key, "", "", "", 10)
if err != nil {
t.Fatal(err)
}
@@ -193,9 +184,18 @@ func TestMultipartListingRestart(t *testing.T) {
if got.Uploads[0].UploadID != mp.UploadID {
t.Fatal("restart changed upload identity")
}
part, err := obj.PutObjectPart(t.Context(), bucket, key, mp.UploadID, 1,
mustGetPutObjReader(t, bytes.NewBufferString("part"), 4, "", ""), ObjectOptions{})
if err != nil {
t.Fatalf("continue known upload after restart: %v", err)
}
if _, err := obj.CompleteMultipartUpload(t.Context(), bucket, key, mp.UploadID,
[]CompletePart{{PartNumber: 1, ETag: part.ETag}}, ObjectOptions{}); err != nil {
t.Fatalf("complete known upload after restart: %v", err)
}
}
func TestMultipartListingConflictingIdentityCopies(t *testing.T) {
func TestMultipartDiagnosticConflictingIdentityCopies(t *testing.T) {
for _, oldCopies := range []int{1, 3} {
z, set, bucket := multipartListingFixture(t)
mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
@@ -213,14 +213,14 @@ func TestMultipartListingConflictingIdentityCopies(t *testing.T) {
t.Fatal(err)
}
}
got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
if !errors.Is(err, errMultipartListingIdentity) || len(got.Uploads) != 0 {
t.Fatalf("%d contradictory copies returned a successful page: %+v %v", oldCopies, got, err)
}
}
}
func TestMultipartListingUnsupportedStoredID(t *testing.T) {
func TestMultipartDiagnosticUnsupportedStoredID(t *testing.T) {
for _, historical := range []bool{false, true} {
z, set, bucket := multipartListingFixture(t)
mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
@@ -244,7 +244,7 @@ func TestMultipartListingUnsupportedStoredID(t *testing.T) {
t.Fatal(err)
}
}
got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
if historical {
if err != nil || len(got.Uploads) != 0 {
t.Fatalf("unidentified historical ID: %+v %v", got, err)
+2 -11
View File
@@ -408,7 +408,7 @@ func paginateMultipartUploads(uploads []MultipartInfo, prefix, keyMarker, upload
}
// listMultipartUploadsExact preserves the hashed exact-object lookup used by
// multipart write placement. Public listing always uses the durable scan.
// multipart write placement and nonempty public listing prefixes.
func (er erasureObjects) listMultipartUploadsExact(ctx context.Context, bucket, object, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (result ListMultipartsInfo, err error) {
auditObjectErasureSet(ctx, "ListMultipartUploads", object, &er)
@@ -516,16 +516,7 @@ func (er erasureObjects) ListMultipartUploads(ctx context.Context, bucket, prefi
if err := checkListMultipartArgs(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter); err != nil {
return ListMultipartsInfo{}, err
}
scan, err := startMultipartScan(ctx, false)
if err != nil {
return ListMultipartsInfo{}, err
}
defer scan.close()
uploads, err := er.scanMultipartUploads(scan, bucket, 0, 0)
if err != nil {
return ListMultipartsInfo{}, err
}
return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil
return er.listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
}
// newMultipartUpload - wrapper for initializing a new multipart
+30 -12
View File
@@ -1916,27 +1916,45 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p
if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil {
return ListMultipartsInfo{}, toObjectErr(err, bucket)
}
scan, err := startMultipartScan(ctx, false)
if err != nil {
return ListMultipartsInfo{}, err
}
defer scan.close()
var uploads []MultipartInfo
poolResult := ListMultipartsInfo{}
poolResult.MaxUploads = maxUploads
poolResult.KeyMarker = keyMarker
poolResult.Prefix = prefix
poolResult.Delimiter = delimiter
// Preserve the legacy cache view, including its incomplete pagination.
// Persistent discovery remains an explicit admin diagnostic, not a
// prerequisite for public listing or known-upload operations.
if prefix == "" {
z.mpCache.Range(func(_ string, mp MultipartInfo) bool {
if mp.Bucket == bucket {
poolResult.Uploads = append(poolResult.Uploads, mp)
}
return true
})
sort.Slice(poolResult.Uploads, func(i int, j int) bool {
return poolResult.Uploads[i].Initiated.Before(poolResult.Uploads[j].Initiated)
})
return poolResult, nil
}
if z.SinglePool() {
return z.serverPools[0].getHashedSet(prefix).listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
}
for idx, pool := range z.serverPools {
if z.IsSuspended(idx) {
continue
}
poolUploads, err := pool.scanMultipartUploads(scan, bucket, idx)
result, err := pool.getHashedSet(prefix).listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
if err != nil {
return ListMultipartsInfo{}, err
return result, err
}
uploads = append(uploads, poolUploads...)
poolResult.Uploads = append(poolResult.Uploads, result.Uploads...)
}
// Confirmed historical records lack a recoverable bucket/key and are
// omitted by the scan. They do not block listing identifiable uploads.
return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil
return poolResult, nil
}
// Initiate a new multipart upload on a hashedSet based on object name.
+1 -25
View File
@@ -886,31 +886,7 @@ func (s *erasureSets) CopyObject(ctx context.Context, srcBucket, srcObject, dstB
}
func (s *erasureSets) ListMultipartUploads(ctx context.Context, bucket, prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (result ListMultipartsInfo, err error) {
if err := checkListMultipartArgs(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter); err != nil {
return ListMultipartsInfo{}, err
}
scan, err := startMultipartScan(ctx, false)
if err != nil {
return ListMultipartsInfo{}, err
}
defer scan.close()
uploads, err := s.scanMultipartUploads(scan, bucket, 0)
if err != nil {
return ListMultipartsInfo{}, err
}
return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil
}
func (s *erasureSets) scanMultipartUploads(scan *multipartScan, bucket string, poolIdx int) ([]MultipartInfo, error) {
var uploads []MultipartInfo
for i, set := range s.sets {
setUploads, err := set.scanMultipartUploads(scan, bucket, poolIdx, i)
if err != nil {
return nil, err
}
uploads = append(uploads, setUploads...)
}
return uploads, nil
return s.getHashedSet(prefix).ListMultipartUploads(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
}
func (s *erasureSets) listMultipartUploadsExact(ctx context.Context, bucket, object string) (ListMultipartsInfo, error) {
+16 -16
View File
@@ -40,7 +40,7 @@ func requireMultipartUploadKeys(t *testing.T, got ListMultipartsInfo, want ...st
}
}
func TestListMultipartUploadsS3Compatibility(t *testing.T) {
func TestMultipartDiagnosticScanPagination(t *testing.T) {
obj, dirs, err := prepareErasureSets32(t.Context())
if err != nil {
t.Fatal(err)
@@ -81,14 +81,14 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
uploadIDs[object] = mp.UploadID
}
// Durable multipart metadata, rather than this node-local cache, must be
// authoritative after a restart or when another node handles the request.
// The diagnostic scanner must discover metadata independently of the cache.
// These checks do not claim that public listing offers durable pagination.
z.mpCache.Range(func(uploadID string, _ MultipartInfo) bool {
z.mpCache.Delete(uploadID)
return true
})
all, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 100)
all, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 100)
if err != nil {
t.Fatal(err)
}
@@ -97,7 +97,7 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
t.Fatalf("complete listing has truncation state: %+v", all)
}
first, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 1)
first, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 1)
if err != nil {
t.Fatal(err)
}
@@ -108,31 +108,31 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
objects[0], uploadIDs[objects[0]])
}
rest, err := z.ListMultipartUploads(t.Context(), bucket, "", first.NextKeyMarker, first.NextUploadIDMarker, "", 100)
rest, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", first.NextKeyMarker, first.NextUploadIDMarker, "", 100)
if err != nil {
t.Fatal(err)
}
requireMultipartUploadKeys(t, rest, objects[1:]...)
afterKey, err := z.ListMultipartUploads(t.Context(), bucket, "", objects[1], "", "", 100)
afterKey, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", objects[1], "", "", 100)
if err != nil {
t.Fatal(err)
}
requireMultipartUploadKeys(t, afterKey, objects[2:]...)
prefixed, err := z.ListMultipartUploads(t.Context(), bucket, "t/", "", "", "", 100)
prefixed, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "t/", "", "", "", 100)
if err != nil {
t.Fatal(err)
}
requireMultipartUploadKeys(t, prefixed, objects[:3]...)
nested, err := z.ListMultipartUploads(t.Context(), bucket, "t/a_b/", "", "", "", 100)
nested, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "t/a_b/", "", "", "", 100)
if err != nil {
t.Fatal(err)
}
requireMultipartUploadKeys(t, nested, objects[:2]...)
grouped, err := z.ListMultipartUploads(t.Context(), bucket, "t/", "", "", SlashSeparator, 100)
grouped, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "t/", "", "", SlashSeparator, 100)
if err != nil {
t.Fatal(err)
}
@@ -141,7 +141,7 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
t.Fatalf("common prefixes = %v, want %v", grouped.CommonPrefixes, want)
}
groupPage, err := z.ListMultipartUploads(t.Context(), bucket, "t/", "", "", SlashSeparator, 1)
groupPage, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "t/", "", "", SlashSeparator, 1)
if err != nil {
t.Fatal(err)
}
@@ -153,7 +153,7 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
groupPage.NextKeyMarker, groupPage.NextUploadIDMarker, groupPage.IsTruncated)
}
groupRest, err := z.ListMultipartUploads(t.Context(), bucket, "t/", groupPage.NextKeyMarker, "", SlashSeparator, 1)
groupRest, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "t/", groupPage.NextKeyMarker, "", SlashSeparator, 1)
if err != nil {
t.Fatal(err)
}
@@ -197,12 +197,12 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
er.getUploadIDDir(bucket, legacyObject, uploadIDs[legacyObject]), metadata, fi.WriteQuorum(er.defaultWQuorum())); err != nil {
t.Fatal(err)
}
legacy, err := z.ListMultipartUploads(t.Context(), bucket, legacyObject, "", "", "", 100)
legacy, err := z.scanMultipartUploadsForTest(t.Context(), bucket, legacyObject, "", "", "", 100)
if err != nil {
t.Fatal(err)
}
requireMultipartUploadKeys(t, legacy)
remaining, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 100)
remaining, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 100)
if err != nil {
t.Fatal(err)
}
@@ -280,7 +280,7 @@ func TestListMultipartUploadsGlobalPageAcrossPools(t *testing.T) {
return true
})
page, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 2)
page, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 2)
if err != nil {
t.Fatal(err)
}
@@ -289,7 +289,7 @@ func TestListMultipartUploadsGlobalPageAcrossPools(t *testing.T) {
t.Fatalf("first global page = %+v", page)
}
rest, err := z.ListMultipartUploads(t.Context(), bucket, "", page.NextKeyMarker, page.NextUploadIDMarker, "", 2)
rest, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", page.NextKeyMarker, page.NextUploadIDMarker, "", 2)
if err != nil {
t.Fatal(err)
}