From 4733e1423cfe703951f52ff9cbd286df2cf9d79d Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sun, 18 Oct 2026 08:22:31 +0800 Subject: [PATCH] fix(multipart): defer persistent listing from public requests Signed-off-by: Feng Ruohang --- .github/workflows/go.yml | 6 +- CHANGELOG.md | 23 ++-- cmd/common-main.go | 2 +- ...rasure-multipart-listing-benchmark_test.go | 89 ++++++++++++++- cmd/erasure-multipart-listing.go | 2 +- cmd/erasure-multipart-listing_test.go | 99 +++++++++++----- cmd/erasure-multipart-public_test.go | 106 ++++++++++++++++++ cmd/erasure-multipart-unified_test.go | 60 +++++----- cmd/erasure-multipart.go | 13 +-- cmd/erasure-server-pool.go | 42 +++++-- cmd/erasure-sets.go | 26 +---- cmd/list-multipart-uploads-compat_test.go | 32 +++--- 12 files changed, 359 insertions(+), 141 deletions(-) create mode 100644 cmd/erasure-multipart-public_test.go diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index 47e889986..93ce684ed 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -93,8 +93,10 @@ jobs: - name: Run conditional PUT tests under race detector run: go test -race ./cmd -run '^Test(PoolsConditionalPut|SinglePoolConditionalPutHTTP)' -count=1 -timeout=5m - - name: Run multipart listing and cancellation tests under race detector - run: go test -race ./cmd -run '^Test(MultipartListing|MultipartAbort|PaginateMultipartUploads|ListMultipartUploads)' -count=1 -timeout=5m + - name: Run multipart listing, cancellation and cleanup tests under race detector + env: + MINIO_API_REQUESTS_MAX: "10000" + run: go test -race ./cmd -run '^Test(Multipart|PaginateMultipartUploads|ListMultipartUploads|CleanupStaleUploads|ExtractModTime)' -count=1 -timeout=5m - name: Run listing metadata recovery, pagination and cancellation tests under race detector run: go test -race ./cmd -run '^Test(ListFallback|ListingCache|ListingMerge|ListingRecovery|ListingPending|ListingEntry|ListingSecondLock|StreamMetadataParts|LocalLockerListing|DistributedListing|MergeXLV2LatestNullCrossParity|CountXLV2MixedECGroups|ListCrossParity)' -count=1 -timeout=5m diff --git a/CHANGELOG.md b/CHANGELOG.md index 298ceed95..98cd5a52b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -76,19 +76,16 @@ candidate's exact delta. ### Object storage and replication -- Make durable `ListMultipartUploads` scanning the only listing path (#79): - apply prefix, delimiter, markers, ordering and the 1,000-entry page limit - globally across pools and sets, independently of node-local caches. Omit - readable historical uploads only when both bucket/key identity fields are - absent; malformed identities, uncertain coverage and resource limits remain - errors. Old uploads still support part upload, completion and cancellation - with a known bucket/key/upload ID. No backfill or new storage layout is added. - `MINIO_API_MULTIPART_LISTING` is retired and ignored with a startup warning; - historical shared `api multipart_listing` keys remain tolerated. The read-only - admin `multipart-preflight` reports `mode=persistent`; `ready` diagnoses the - absence of old records and no longer gates listing. Each page still scans - durable state within the existing 100,000-directory-entry, 30-second and - per-process concurrency budgets. See the +- Withdraw default persistent `ListMultipartUploads` scanning (#79). Restore + node-local cache listing for an empty prefix and exact-key lookup otherwise; + restart omissions, true-prefix and cache-pagination limitations remain. + Preserve bucket/key identity metadata, path safety and stale-upload cleanup. + `MINIO_API_MULTIPART_LISTING` stays retired and ignored with a startup warning; + obsolete shared `api multipart_listing` keys remain tolerated. **Existing + opt-in strict users lose durable listing and strict cancellation.** The + read-only admin `multipart-preflight` reports `mode=legacy`; its scan budgets + and readiness do not gate public listing or startup. Durable, complete, + scalable enumeration remains deferred. See the [design and limits](https://silo.pgsty.com/blog/design/list-multipart-uploads/#implementation). Thanks to mr javad seydi (@mrjavadseydi) for the original metadata-and-scan implementation (#198). diff --git a/cmd/common-main.go b/cmd/common-main.go index 47bd895ab..3b5a08c01 100644 --- a/cmd/common-main.go +++ b/cmd/common-main.go @@ -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, diff --git a/cmd/erasure-multipart-listing-benchmark_test.go b/cmd/erasure-multipart-listing-benchmark_test.go index 7dd430a5b..539a82b3c 100644 --- a/cmd/erasure-multipart-listing-benchmark_test.go +++ b/cmd/erasure-multipart-listing-benchmark_test.go @@ -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) + }) + } +} diff --git a/cmd/erasure-multipart-listing.go b/cmd/erasure-multipart-listing.go index 252a8ef6b..d747444f3 100644 --- a/cmd/erasure-multipart-listing.go +++ b/cmd/erasure-multipart-listing.go @@ -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) diff --git a/cmd/erasure-multipart-listing_test.go b/cmd/erasure-multipart-listing_test.go index c481cbd67..7f5e21ce2 100644 --- a/cmd/erasure-multipart-listing_test.go +++ b/cmd/erasure-multipart-listing_test.go @@ -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 { diff --git a/cmd/erasure-multipart-public_test.go b/cmd/erasure-multipart-public_test.go new file mode 100644 index 000000000..13d8cafcf --- /dev/null +++ b/cmd/erasure-multipart-public_test.go @@ -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()) + } + }) + } +} diff --git a/cmd/erasure-multipart-unified_test.go b/cmd/erasure-multipart-unified_test.go index a14a0ea73..afe535712 100644 --- a/cmd/erasure-multipart-unified_test.go +++ b/cmd/erasure-multipart-unified_test.go @@ -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) diff --git a/cmd/erasure-multipart.go b/cmd/erasure-multipart.go index 5ffeef81d..6d717df26 100644 --- a/cmd/erasure-multipart.go +++ b/cmd/erasure-multipart.go @@ -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 diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index c2cda0896..2075817e9 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -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. diff --git a/cmd/erasure-sets.go b/cmd/erasure-sets.go index 4ee1a1f3c..daf7d8e99 100644 --- a/cmd/erasure-sets.go +++ b/cmd/erasure-sets.go @@ -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) { diff --git a/cmd/list-multipart-uploads-compat_test.go b/cmd/list-multipart-uploads-compat_test.go index 0c8722c74..df39a2afe 100644 --- a/cmd/list-multipart-uploads-compat_test.go +++ b/cmd/list-multipart-uploads-compat_test.go @@ -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) }