Merge pull request #243: unify durable multipart listing

Use the existing persistent scanner for public multipart listing, tolerate confirmed unidentified historical records individually, and retire the migration mode while preserving default Abort availability.

Keep capacity and delayed-create boundaries explicit. Related to #79; this does not close its remaining lifecycle and capacity work. Companion documentation: pgsty/silo.pgsty.com#43.

Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
Feng Ruohang
2026-10-17 10:15:39 +08:00
committed by GitHub
15 changed files with 589 additions and 499 deletions
+23 -29
View File
@@ -54,35 +54,29 @@ and [complete commit range](https://github.com/pgsty/silo/compare/RELEASE.2026-0
### Object storage and replication
- Make `ListMultipartUploads` discover quorum-valid uploads from durable state
across pools, erasure sets and drives, then apply S3 prefix, delimiter,
marker, ordering and 1,000-entry pagination semantics globally (#198). New uploads
store their canonical bucket and key as reserved fields in the existing
quorum-written `xl.meta`; completion removes those upload-only fields. Native
markers remain usable after their upload is completed or canceled. Strict
listing returns a diagnostic 503 for legacy uploads or uncertain coverage;
the default remains the released exact-key/cache-based `legacy` behavior.
Opt into strict mode only through `MINIO_API_MULTIPART_LISTING=strict`, after
upgrading every writer, draining old uploads, checking the read-only admin
`multipart-preflight` report and validating scan capacity. The process-only
setting is not persisted into shared API configuration. Historical
`multipart_listing` keys are ignored and can be removed with a targeted
`mcli admin config reset ALIAS api multipart_listing` before rollback. Per-process
admission, directory-entry, worker and time budgets bound scan scheduling;
each page still scans durable state. See [issue #79](https://github.com/pgsty/silo/issues/79)
and its [design record](https://silo.pgsty.com/blog/design/list-multipart-uploads/).
Thanks to mr javad seydi (@mrjavadseydi) for the original implementation.
- Retain released read-quorum and best-effort multipart cancellation in default
legacy mode. Strict mode requires majority deletion acknowledgements and
permits retries below read quorum. When most drives were already empty,
failed deletion of an observed remnant now returns 503 instead of being
masked by empty-drive successes. Wrong-key or wrong-bucket cancellation
preserves the valid upload's cache entry; successful cancellation notifies
peers with the request context after the distributed lock is released.
The HTTP response for an absent
upload remains 204; this is not proof of physical cleanup. **Known boundary:**
delayed creation writes can still restore an upload after cancellation;
this change does not add a durable creation fence.
- 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
[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).
- Keep the default read-quorum, best-effort multipart cancellation and
first-matching-pool behavior when retiring the listing switch. The optional
strict cancellation branch is removed. Wrong-target protection, upload-ID
path validation and peer notification with the request context remain.
HTTP 204 for an absent upload remains idempotent and does not prove physical
cleanup. **Known boundary:** delayed creation writes can still restore a
readable upload after cancellation; this change adds no creation fence.
- Preserve object tags during multi-pool metadata reconciliation by reading the
resolved tag field together with its revision (#189). Previously, reconciliation
could replace existing tags with an empty value.
+1 -3
View File
@@ -450,7 +450,7 @@ const (
ErrAdminNoSecretKey
ErrIAMNotInitialized
ErrMultipartListingLegacy
ErrMultipartListingLegacy // Reserved: legacy uploads no longer block listing.
ErrMultipartListingIdentity
ErrSlowDown
@@ -2191,8 +2191,6 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
err = unwrapAll(err)
switch err {
case errMultipartListingLegacy:
apiErr = ErrMultipartListingLegacy
case errMultipartListingIdentity:
apiErr = ErrMultipartListingIdentity
case errCompleteMultipartChecksumMismatch, errCompleteMultipartChecksumTypeMismatch:
+4
View File
@@ -56,6 +56,7 @@ import (
"github.com/minio/minio/internal/auth"
"github.com/minio/minio/internal/color"
"github.com/minio/minio/internal/config"
"github.com/minio/minio/internal/config/api"
"github.com/minio/minio/internal/handlers"
"github.com/minio/minio/internal/kms"
"github.com/minio/minio/internal/logger"
@@ -881,6 +882,9 @@ func serverHandleEnvVars() {
logger.Warning("%s=%s is ignored: Silo in-place updates are permanently disabled", config.EnvUpdate, updateSetting)
}
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)
}
// Check if the supported credential env vars,
// "MINIO_ROOT_USER" and "MINIO_ROOT_PASSWORD" are provided
@@ -0,0 +1,85 @@
// Copyright (c) 2026 Ruohang Feng
// SPDX-License-Identifier: AGPL-3.0-or-later
package cmd
import (
"context"
"errors"
"fmt"
"sync/atomic"
"testing"
)
// 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) {
for _, tc := range []struct {
drives, uploads int
mixed, limited bool
}{
{4, 1_000, false, false},
{4, 10_000, false, false},
{4, 10_000, true, false},
{16, 1_000, false, false},
{16, 10_000, false, true},
} {
b.Run(fmt.Sprintf("drives=%d/uploads=%d/mixed=%t", tc.drives, tc.uploads, tc.mixed), 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-capacity", "multipart-unrelated"
for _, name := range []string{bucket, other} {
if err := z.MakeBucket(b.Context(), name, MakeBucketOptions{}); err != nil {
b.Fatal(err)
}
}
for n := range tc.uploads {
target := bucket
if tc.mixed && n%2 != 0 {
target = other
}
key := fmt.Sprintf("dir/%05d", n)
mp, err := z.NewMultipartUpload(b.Context(), target, key, ObjectOptions{})
if err != nil {
b.Fatal(err)
}
if tc.mixed && n%10 == 0 {
rewriteMultipartIdentity(b, z.serverPools[0].getHashedSet(key), target, key, mp.UploadID, removeMultipartIdentity)
}
}
z.mpCache.Clear()
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 })
}
b.ResetTimer()
for range b.N {
got, err := z.ListMultipartUploads(b.Context(), bucket, "dir/", "", "", "", 1000)
if tc.limited {
var limited SlowDown
if !errors.As(err, &limited) || len(got.Uploads) != 0 {
b.Fatalf("overflow must return SlowDown without a partial page: %+v %v", got, err)
}
} else if err != nil || len(got.Uploads) != 1000 {
b.Fatalf("capacity listing: count=%d err=%v", len(got.Uploads), err)
}
}
b.StopTimer()
b.ReportMetric(float64(directoryCalls.Load())/float64(b.N), "ListDir/op")
b.ReportMetric(float64(metadataCalls.Load())/float64(b.N), "ReadVersion/op")
})
}
}
+36 -17
View File
@@ -23,7 +23,6 @@ const multipartScanEntryLimit = 100_000
var (
multipartScanSlots = make(chan struct{}, 2)
errMultipartListingLegacy = errors.New("legacy multipart uploads require a coordinated upgrade and drain")
errMultipartListingIdentity = errors.New("multipart upload identity is invalid")
)
@@ -133,6 +132,14 @@ func (er erasureObjects) multipartIdentity(fi FileInfo, shaDir string) (string,
IsValidObjectPrefix(object) && er.getMultipartSHADir(bucket, object) == shaDir
}
// Only the absence of both fields identifies the historical format. Empty,
// partial or invalid identities must not hide a damaged new upload as legacy.
func multipartLegacyIdentity(fi FileInfo) bool {
_, bucketPresent := fi.Metadata[multipartMetaBucket]
_, objectPresent := fi.Metadata[multipartMetaObject]
return !bucketPresent && !objectPresent
}
func (er erasureObjects) readMultipartUploadCandidate(s *multipartScan, bucket, candidate string, source StorageAPI) (MultipartInfo, bool, bool, error) {
if s.ctx.Err() != nil {
return MultipartInfo{}, false, false, SlowDown{}
@@ -162,8 +169,8 @@ func (er erasureObjects) readMultipartUploadCandidate(s *multipartScan, bucket,
disks := er.getDisks()
metadata, errs := readAllFileInfo(s.ctx, disks, bucket, minioMetaMultipartBucket, candidate, "", false, false)
if s.preflight {
// Readiness is stronger than listing liveness: even one readable legacy
// copy must be drained, and an unreadable copy cannot certify readiness.
// The diagnostic counts even a single readable legacy copy. It does not
// gate listing; unreadable copies still prevent a complete diagnosis.
var oldest MultipartInfo
legacy := false
_, nativeID := multipartUploadTime(uploadUUID)
@@ -176,10 +183,11 @@ func (er erasureObjects) readMultipartUploadCandidate(s *multipartScan, bucket,
}
fi := metadata[i]
storedBucket, storedObject, valid := er.multipartIdentity(fi, shaDir)
if storedBucket != "" && storedObject != "" && !valid {
isLegacy := multipartLegacyIdentity(fi)
if !isLegacy && (!valid || !nativeID) {
return MultipartInfo{}, false, false, errMultipartListingIdentity
}
if !valid || !nativeID {
if isLegacy {
info := multipartUploadInfo(storedBucket, storedObject, uploadUUID, fi.ModTime)
if !legacy || info.Initiated.Before(oldest.Initiated) {
oldest = info
@@ -203,16 +211,29 @@ func (er erasureObjects) readMultipartUploadCandidate(s *multipartScan, bucket,
if err != nil {
return MultipartInfo{}, false, false, err
}
// The generic metadata quorum does not compare these identity fields.
// Contradictory readable copies cannot certify a historical upload: picking
// a copy with missing fields must not silently hide a new-format record.
legacy := multipartLegacyIdentity(fi)
storedBucket, storedObject, valid := er.multipartIdentity(fi, shaDir)
for i, other := range metadata {
if errs[i] != nil || !other.IsValid() {
continue
}
if multipartLegacyIdentity(other) != legacy ||
(!legacy && (other.Metadata[multipartMetaBucket] != storedBucket || other.Metadata[multipartMetaObject] != storedObject)) {
return MultipartInfo{}, false, false, fmt.Errorf("%w: inconsistent copies at %s", errMultipartListingIdentity, candidate)
}
}
info := multipartUploadInfo(storedBucket, storedObject, uploadUUID, fi.ModTime)
if storedBucket == "" || storedObject == "" {
if legacy {
return info, false, true, nil
}
if !valid {
return MultipartInfo{}, false, false, fmt.Errorf("%w: %s", errMultipartListingIdentity, candidate)
}
if _, ok := multipartUploadTime(uploadUUID); !ok {
return info, false, true, nil
return MultipartInfo{}, false, false, fmt.Errorf("%w: %s", errMultipartListingIdentity, candidate)
}
if bucket != "" && bucket != storedBucket {
return MultipartInfo{}, false, false, nil
@@ -220,7 +241,7 @@ func (er erasureObjects) readMultipartUploadCandidate(s *multipartScan, bucket,
return info, true, false, nil
}
func (er erasureObjects) scanMultipartUploads(s *multipartScan, bucket string, poolIdx, setIdx int) ([]MultipartInfo, bool, error) {
func (er erasureObjects) scanMultipartUploads(s *multipartScan, bucket string, poolIdx, setIdx int) ([]MultipartInfo, error) {
disks := er.getDisks()
report := multipartScanSet{Pool: poolIdx, Set: setIdx, Drives: er.setDriveCount}
var resultErr error
@@ -272,7 +293,7 @@ func (er erasureObjects) scanMultipartUploads(s *multipartScan, bucket string, p
resultErr = toObjectErr(errErasureReadQuorum, bucket)
}
if resultErr != nil {
return nil, false, resultErr
return nil, resultErr
}
type candidate struct {
path string
@@ -325,25 +346,23 @@ dispatch:
if s.ctx.Err() != nil {
resultErr = SlowDown{}
}
return uploads, report.LegacyUploads != 0, resultErr
return uploads, resultErr
}
// multipartPreflight scans all pools, sets and drives, independent of caches.
// A majority suffices for normal listing; upgrade readiness requires every
// drive to have been inspected. The operator must also upgrade all writers.
// A majority suffices for normal listing; a complete diagnosis requires every
// drive to have been inspected. Ready describes the absence of legacy copies,
// not whether listing is enabled.
func (z *erasureServerPools) multipartPreflight(ctx context.Context) (multipartPreflightReport, error) {
s, err := startMultipartScan(ctx, true)
if err != nil {
return multipartPreflightReport{}, err
}
defer s.close()
report := multipartPreflightReport{Complete: true, Mode: "strict"}
if globalAPIConfig.getMultipartListingLegacy() {
report.Mode = "legacy"
}
report := multipartPreflightReport{Complete: true, Mode: "persistent"}
for p, pool := range z.serverPools {
for i, set := range pool.sets {
_, _, _ = set.scanMultipartUploads(s, "", p, i)
_, _ = set.scanMultipartUploads(s, "", p, i)
}
}
report.Sets = s.sets
+42 -80
View File
@@ -31,7 +31,7 @@ func TestMultipartListingAbortBetweenHTTPPages(t *testing.T) {
if err != nil {
t.Fatal(err)
}
setMultipartListingTestMode(t, false)
var firstID, secondID string
for attempt := 0; attempt < 32; attempt++ {
one, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
@@ -148,28 +148,25 @@ func TestMultipartListingLegacyPreflight(t *testing.T) {
t.Fatal(err)
}
z.mpCache.Clear()
t.Run("legacy-upgrade", func(t *testing.T) {
setMultipartListingTestMode(t, true)
exact, err := z.ListMultipartUploads(t.Context(), other, "old", "", "", "", 10)
for _, target := range []string{bucket, other} {
got, err := z.ListMultipartUploads(t.Context(), target, "", "", "", "", 10)
if err != nil {
t.Fatal(err)
}
requireMultipartUploadKeys(t, exact, "old")
const empty = "multipart-legacy-empty"
if err := z.MakeBucket(t.Context(), empty, MakeBucketOptions{}); err != nil {
t.Fatal(err)
if target == bucket {
requireMultipartUploadKeys(t, got, "a")
} else {
requireMultipartUploadKeys(t, got)
}
got, err := z.ListMultipartUploads(t.Context(), empty, "", "", "", "", 10)
if err != nil || len(got.Uploads) != 0 {
t.Fatalf("old upload in another bucket broke legacy listing: %+v %v", got, err)
}
})
_, err = z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
if !errors.Is(err, errMultipartListingLegacy) {
t.Fatalf("old upload in another bucket: %v", err)
}
// Exact-looking prefixes now have the same durable semantics.
exact, err := z.ListMultipartUploads(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 {
if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "persistent" {
t.Fatalf("legacy preflight: %+v %v", report, err)
}
if err = z.AbortMultipartUpload(t.Context(), other, "old", old.UploadID, ObjectOptions{}); err != nil {
@@ -187,7 +184,7 @@ func TestMultipartListingLegacyPreflight(t *testing.T) {
}
func TestMultipartListingIdentityFallback(t *testing.T) {
for _, kind := range []string{"old", "corrupt", "read-failure", "wrong-bucket"} {
for _, kind := range []string{"old", "partial", "corrupt", "read-failure", "wrong-bucket"} {
t.Run(kind, func(t *testing.T) {
z, set, bucket := multipartListingFixture(t)
if _, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}); err != nil {
@@ -209,8 +206,11 @@ func TestMultipartListingIdentityFallback(t *testing.T) {
return FileInfo{}, errFileCorrupt
}
fi.Metadata = cloneMSS(fi.Metadata)
if kind == "old" {
if kind == "old" || kind == "partial" {
delete(fi.Metadata, multipartMetaBucket)
if kind == "old" {
delete(fi.Metadata, multipartMetaObject)
}
} else {
fi.Metadata[multipartMetaBucket] = "another-bucket"
}
@@ -231,7 +231,7 @@ func TestMultipartListingIdentityFallback(t *testing.T) {
func TestMultipartAbortPoolsAndRetry(t *testing.T) {
z, bucket := consistencyPools(t)
setMultipartListingTestMode(t, false)
mp, err := z.serverPools[1].NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
if err != nil {
t.Fatal(err)
@@ -252,10 +252,12 @@ func TestMultipartAbortPoolsAndRetry(t *testing.T) {
if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 {
t.Fatalf("unknown pool must not acknowledge cancellation: %v", err)
}
if _, err := z.serverPools[1].GetMultipartInfo(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatalf("earlier pool error changed the later pool: %v", err)
}
emptySet.getDisks = original
err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{})
if _, ok := err.(InvalidUploadID); !ok {
t.Fatalf("retry after all pools confirm absence: %v", err)
if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatalf("retry after pool recovery: %v", err)
}
mp, err = z.serverPools[1].NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
if err != nil {
@@ -289,12 +291,13 @@ func TestMultipartAbortRetryBelowReadQuorum(t *testing.T) {
}}
return disks
}
if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err == nil {
t.Fatal("lost acknowledgement must fail")
if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatalf("best-effort cleanup must retain read-quorum availability: %v", err)
}
set.getDisks = original
if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatalf("one remaining metadata copy prevented retry: %v", err)
var invalid InvalidUploadID
if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) {
t.Fatalf("minority remnant must retain absent-upload result: %v", err)
}
}
@@ -304,7 +307,7 @@ func TestMultipartListingMarkerHTTP(t *testing.T) {
if err != nil {
t.Fatal(err)
}
setMultipartListingTestMode(t, false)
for _, tc := range []struct {
key, marker string
status int
@@ -388,7 +391,7 @@ func TestMultipartAbortLateCreateBoundary(t *testing.T) {
t.Fatal(err)
}
z := obj.(*erasureServerPools)
setMultipartListingTestMode(t, false)
t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) })
saved := globalStorageClass
globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: 8}})
@@ -417,9 +420,7 @@ func TestMultipartAbortLateCreateBoundary(t *testing.T) {
}
set.getDisks = func() []StorageAPI {
disks := append([]StorageAPI(nil), original()...)
for i := 2; i < 9; i++ {
disks[i] = nil
}
disks[8] = nil
return disks
}
if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
@@ -431,7 +432,7 @@ func TestMultipartAbortLateCreateBoundary(t *testing.T) {
}
}
set.getDisks = original
_, metadata, err := set.checkUploadIDExists(t.Context(), bucket, "a", mp.UploadID, true)
_, metadata, err := set.checkUploadIDExists(t.Context(), bucket, "a", mp.UploadID, false)
if err != nil {
t.Fatalf("late creation boundary changed; review and remove the documented limitation: %v", err)
}
@@ -441,10 +442,10 @@ func TestMultipartAbortLateCreateBoundary(t *testing.T) {
count++
}
}
if count != 14 {
t.Fatalf("expected fourteen resurrected copies, got %d", count)
if count != 8 {
t.Fatalf("expected eight resurrected copies, got %d", count)
}
t.Log("KNOWN UNRESOLVED BOUNDARY: seven delayed creation writes plus seven offline copies restore a writable upload after acknowledged cancellation")
t.Log("KNOWN UNRESOLVED BOUNDARY: seven delayed creation writes plus one offline copy restore a readable upload after acknowledged cancellation")
}
type multipartListingCountingDisk struct {
@@ -474,7 +475,7 @@ func TestMultipartListingScanCosts(t *testing.T) {
t.Fatal(err)
}
z := obj.(*erasureServerPools)
setMultipartListingTestMode(t, false)
t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) })
const bucket, otherBucket = "r9-scan-target", "r9-scan-unrelated"
for _, name := range []string{bucket, otherBucket} {
@@ -544,7 +545,7 @@ func multipartListingFixture(t *testing.T) (*erasureServerPools, *erasureObjects
if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil {
t.Fatal(err)
}
setMultipartListingTestMode(t, false)
return z, z.serverPools[0].getHashedSet("a"), bucket
}
@@ -588,45 +589,6 @@ func TestMultipartListingMissingMarkers(t *testing.T) {
}
}
func TestMultipartListingAbortRecovery(t *testing.T) {
for _, offline := range []int{1, 2} {
t.Run(fmt.Sprint(offline), func(t *testing.T) {
z, set, bucket := multipartListingFixture(t)
mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
original := set.getDisks
t.Cleanup(func() { set.getDisks = original })
set.getDisks = func() []StorageAPI {
disks := append([]StorageAPI(nil), original()...)
for i := 0; i < offline; i++ {
disks[i] = nil
}
return disks
}
err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{})
if offline == 1 && err != nil {
t.Fatal(err)
}
if offline == 2 && (err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503) {
t.Fatalf("two offline drives must not acknowledge cancellation: %v", err)
}
set.getDisks = original
if offline == 2 {
// Retry a partially completed deletion after the original disks return.
if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatal(err)
}
}
got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 100)
if err != nil || len(got.Uploads) != 0 {
t.Fatalf("canceled upload reappeared: %+v %v", got, err)
}
})
}
}
func TestMultipartListingCoverage(t *testing.T) {
z, set, bucket := multipartListingFixture(t)
mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
@@ -670,7 +632,7 @@ func TestMultipartListingBudgetAndAdmission(t *testing.T) {
}
defer scan.close()
scan.remaining.Store(1)
_, _, err = set.scanMultipartUploads(scan, bucket, 0, 0)
_, err = set.scanMultipartUploads(scan, bucket, 0, 0)
var limited SlowDown
if !errors.As(err, &limited) {
t.Fatalf("budget: %v", err)
@@ -695,7 +657,7 @@ func TestMultipartListingAdmissionHTTP(t *testing.T) {
if err != nil {
t.Fatal(err)
}
setMultipartListingTestMode(t, false)
for range cap(multipartScanSlots) {
scan, err := startMultipartScan(t.Context(), false)
if err != nil {
@@ -743,7 +705,7 @@ func TestMultipartListingPreflightMinorityLegacy(t *testing.T) {
}
}
report, err := z.multipartPreflight(t.Context())
if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 {
if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "persistent" {
t.Fatalf("minority legacy copy must prevent readiness: %+v %v", report, err)
}
}
+80 -175
View File
@@ -12,41 +12,13 @@ import (
"sync/atomic"
"testing"
"github.com/minio/minio/internal/config/api"
"github.com/minio/minio/internal/config/storageclass"
"github.com/minio/minio/internal/dsync"
"github.com/minio/minio/internal/grid"
xnet "github.com/pgsty/silo-pkg/v3/net"
)
func setMultipartListingTestMode(t *testing.T, legacy bool) {
t.Helper()
globalAPIConfig.mu.Lock()
previous := globalAPIConfig.multipartListingStrict
globalAPIConfig.multipartListingStrict = !legacy
globalAPIConfig.mu.Unlock()
t.Cleanup(func() {
globalAPIConfig.mu.Lock()
globalAPIConfig.multipartListingStrict = previous
globalAPIConfig.mu.Unlock()
})
}
func TestMultipartListingDefaultMode(t *testing.T) {
var uninitialized apiConfig
if !uninitialized.getMultipartListingLegacy() {
t.Fatal("runtime zero value enabled strict mode before config initialization")
}
for _, mode := range []string{"", "legacy", "strict"} {
var local apiConfig
local.init(api.Config{RequestsMax: 1, MultipartListing: mode}, []int{4}, false)
if local.getMultipartListingLegacy() != (mode != "strict") {
t.Fatalf("unexpected mode for %q", mode)
}
}
}
func TestMultipartAbortLegacyAvailability(t *testing.T) {
func TestMultipartAbortDefaultAvailability(t *testing.T) {
for _, drives := range []int{4, 16} {
for _, offline := range []int{0, 1, drives / 2} {
t.Run(fmt.Sprintf("drives=%d/offline=%d", drives, offline), func(t *testing.T) {
@@ -56,7 +28,7 @@ func TestMultipartAbortLegacyAvailability(t *testing.T) {
}
z := obj.(*erasureServerPools)
t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) })
setMultipartListingTestMode(t, true)
previous := globalStorageClass
globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: drives / 2}})
t.Cleanup(func() { globalStorageClass.Update(previous) })
@@ -91,11 +63,11 @@ func TestMultipartAbortLegacyAvailability(t *testing.T) {
}
}
func TestMultipartAbortLegacyPoolOrder(t *testing.T) {
func TestMultipartAbortDefaultPoolOrder(t *testing.T) {
for _, owner := range []int{0, 1} {
t.Run(fmt.Sprintf("owner=%d", owner), func(t *testing.T) {
z, bucket := consistencyPools(t)
setMultipartListingTestMode(t, true)
mp, err := z.serverPools[owner].NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
if err != nil {
t.Fatal(err)
@@ -106,7 +78,7 @@ func TestMultipartAbortLegacyPoolOrder(t *testing.T) {
emptySet.getDisks = func() []StorageAPI { return make([]StorageAPI, emptySet.setDriveCount) }
err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{})
if owner == 0 && err != nil {
t.Fatalf("unrelated later pool blocked legacy cancellation: %v", err)
t.Fatalf("unrelated later pool blocked cancellation: %v", err)
}
if owner == 1 {
if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 {
@@ -124,123 +96,56 @@ func TestMultipartAbortLegacyPoolOrder(t *testing.T) {
}
}
func TestMultipartAbortMinorityLegacyCleanup(t *testing.T) {
for _, failDelete := range []bool{false, true} {
t.Run(fmt.Sprintf("delete-fails=%v", failDelete), func(t *testing.T) {
z, set, bucket := multipartListingFixture(t)
mp, err := z.NewMultipartUpload(t.Context(), bucket, "old", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
path := set.getUploadIDDir(bucket, "old", mp.UploadID)
original := set.getDisks
disks := original()
fi, err := disks[0].ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, path, "", ReadOptions{})
if err != nil {
t.Fatal(err)
}
delete(fi.Metadata, multipartMetaBucket)
delete(fi.Metadata, multipartMetaObject)
if err := disks[0].WriteMetadata(t.Context(), bucket, minioMetaMultipartBucket, path, fi); err != nil {
t.Fatal(err)
}
for _, d := range disks[1:] {
if err := d.Delete(t.Context(), minioMetaMultipartBucket, path, DeleteOptions{Recursive: true}); err != nil {
t.Fatal(err)
}
}
report, err := z.multipartPreflight(t.Context())
if err != nil || report.Ready || report.LegacyUploads != 1 {
t.Fatalf("invalid legacy fixture: %+v %v", report, err)
}
if failDelete {
set.getDisks = func() []StorageAPI {
wrapped := append([]StorageAPI(nil), disks...)
wrapped[0] = multipartListingFaultDisk{StorageAPI: disks[0], delete: func(context.Context, string, string, DeleteOptions) error { return errFaultyDisk }}
return wrapped
}
t.Cleanup(func() { set.getDisks = original })
err := z.AbortMultipartUpload(t.Context(), bucket, "old", mp.UploadID, ObjectOptions{})
if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 {
t.Fatalf("empty drives masked failed deletion of the observed replica: %v", err)
}
if _, present := z.mpCache.Load(mp.UploadID); !present {
t.Fatal("failed cancellation evicted upload from cache")
}
set.getDisks = original
}
if err := z.AbortMultipartUpload(t.Context(), bucket, "old", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatal(err)
}
report, err = z.multipartPreflight(t.Context())
if err != nil || !report.Ready || report.LegacyUploads != 0 {
t.Fatalf("acknowledged cleanup left online legacy remnants: %+v %v", report, err)
}
if _, present := z.mpCache.Load(mp.UploadID); present {
t.Fatal("successful cancellation retained cache entry")
}
})
}
}
func TestMultipartAbortWrongTargetKeepsCache(t *testing.T) {
for _, legacy := range []bool{true, false} {
t.Run(fmt.Sprintf("legacy=%v", legacy), func(t *testing.T) {
z, _, bucket := multipartListingFixture(t)
setMultipartListingTestMode(t, legacy)
const other = "multipart-other-bucket"
if err := z.MakeBucket(t.Context(), other, MakeBucketOptions{}); err != nil {
t.Fatal(err)
}
mp, err := z.NewMultipartUpload(t.Context(), bucket, "valid-key", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
for _, target := range [][2]string{{bucket, "wrong-key"}, {other, "valid-key"}} {
err := z.AbortMultipartUpload(t.Context(), target[0], target[1], mp.UploadID, ObjectOptions{})
var invalid InvalidUploadID
if !errors.As(err, &invalid) {
t.Fatalf("wrong target result: %v", err)
}
if _, present := z.mpCache.Load(mp.UploadID); !present {
t.Fatal("wrong-target abort evicted another upload")
}
if _, err := z.GetMultipartInfo(t.Context(), bucket, "valid-key", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatalf("wrong-target abort harmed valid upload: %v", err)
}
}
if err := z.AbortMultipartUpload(t.Context(), bucket, "valid-key", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatal(err)
}
var invalid InvalidUploadID
if err := z.AbortMultipartUpload(t.Context(), bucket, "valid-key", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) {
t.Fatalf("repeat abort: %v", err)
}
})
z, _, bucket := multipartListingFixture(t)
const other = "multipart-other-bucket"
if err := z.MakeBucket(t.Context(), other, MakeBucketOptions{}); err != nil {
t.Fatal(err)
}
mp, err := z.NewMultipartUpload(t.Context(), bucket, "valid-key", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
for _, target := range [][2]string{{bucket, "wrong-key"}, {other, "valid-key"}} {
err := z.AbortMultipartUpload(t.Context(), target[0], target[1], mp.UploadID, ObjectOptions{})
var invalid InvalidUploadID
if !errors.As(err, &invalid) {
t.Fatalf("wrong target result: %v", err)
}
if _, present := z.mpCache.Load(mp.UploadID); !present {
t.Fatal("wrong-target abort evicted another upload")
}
if _, err := z.GetMultipartInfo(t.Context(), bucket, "valid-key", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatalf("wrong-target abort harmed valid upload: %v", err)
}
}
if err := z.AbortMultipartUpload(t.Context(), bucket, "valid-key", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatal(err)
}
var invalid InvalidUploadID
if err := z.AbortMultipartUpload(t.Context(), bucket, "valid-key", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) {
t.Fatalf("repeat abort: %v", err)
}
}
func TestMultipartAbortRejectsUnsafeID(t *testing.T) {
for _, legacy := range []bool{true, false} {
t.Run(fmt.Sprintf("legacy=%v", legacy), func(t *testing.T) {
z, _, bucket := multipartListingFixture(t)
setMultipartListingTestMode(t, legacy)
mp, err := z.NewMultipartUpload(t.Context(), bucket, "keep", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
for _, suffix := range []string{".", "..", "../other", "a/b", "a\\b", ""} {
id := base64.RawURLEncoding.EncodeToString([]byte("deployment." + suffix))
err := z.AbortMultipartUpload(t.Context(), bucket, "keep", id, ObjectOptions{})
var invalid InvalidUploadID
if !errors.As(err, &invalid) {
t.Fatalf("unsafe ID accepted: suffix=%q err=%v", suffix, err)
}
if _, err := z.GetMultipartInfo(t.Context(), bucket, "keep", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatalf("valid upload harmed by rejected ID: %v", err)
}
}
})
z, _, bucket := multipartListingFixture(t)
mp, err := z.NewMultipartUpload(t.Context(), bucket, "keep", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
for _, suffix := range []string{".", "..", "../other", "a/b", "a\\b", ""} {
id := base64.RawURLEncoding.EncodeToString([]byte("deployment." + suffix))
err := z.AbortMultipartUpload(t.Context(), bucket, "keep", id, ObjectOptions{})
var invalid InvalidUploadID
if !errors.As(err, &invalid) {
t.Fatalf("unsafe ID accepted: suffix=%q err=%v", suffix, err)
}
if _, err := z.GetMultipartInfo(t.Context(), bucket, "keep", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatalf("valid upload harmed by rejected ID: %v", err)
}
}
}
@@ -275,34 +180,30 @@ func TestMultipartAbortPeerNotification(t *testing.T) {
set.nsMutex = newNSLock(true)
set.getLockers = func() ([]dsync.NetLocker, string) { return lockers, "multipart-test" }
t.Cleanup(func() { set.nsMutex, set.getLockers = oldMutex, oldLockers })
for _, legacy := range []bool{true, false} {
t.Run(fmt.Sprintf("legacy=%v", legacy), func(t *testing.T) {
setMultipartListingTestMode(t, legacy)
for range 4 {
mp, err := z.NewMultipartUpload(t.Context(), bucket, "valid", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
before := notifications.Load()
var invalid InvalidUploadID
if err := z.AbortMultipartUpload(t.Context(), bucket, "wrong", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) {
t.Fatalf("wrong target: %v", err)
}
if notifications.Load() != before {
t.Fatal("wrong-target abort sent a destructive peer notification")
}
if err := z.AbortMultipartUpload(t.Context(), bucket, "valid", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatal(err)
}
if notifications.Load() != before+1 {
t.Fatal("successful abort lost peer notification after unlocking")
}
}
})
for range 4 {
mp, err := z.NewMultipartUpload(t.Context(), bucket, "valid", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
before := notifications.Load()
var invalid InvalidUploadID
if err := z.AbortMultipartUpload(t.Context(), bucket, "wrong", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) {
t.Fatalf("wrong target: %v", err)
}
if notifications.Load() != before {
t.Fatal("wrong-target abort sent a destructive peer notification")
}
if err := z.AbortMultipartUpload(t.Context(), bucket, "valid", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatal(err)
}
if notifications.Load() != before+1 {
t.Fatal("successful abort lost peer notification after unlocking")
}
}
}
func TestMultipartAbortStrictMajority(t *testing.T) {
func TestMultipartAbortBestEffortCleanup(t *testing.T) {
z, set, bucket := multipartListingFixture(t)
mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
if err != nil {
@@ -317,15 +218,19 @@ func TestMultipartAbortStrictMajority(t *testing.T) {
}
t.Cleanup(func() { set.getDisks = original })
if err := z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatalf("ordinary majority cancellation was tightened: %v", err)
t.Fatalf("read-quorum cancellation was tightened: %v", err)
}
// One failed deletion is tolerated in the ordinary majority path. Retrying
// after recovery must also clean the now-minority remnant.
// Best-effort deletion may leave a remnant. Retrying after recovery cleans
// it, but the read quorum no longer recognizes it as an active upload.
if _, err := disks[0].ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, set.getUploadIDDir(bucket, "a", mp.UploadID), "", ReadOptions{}); err != nil {
t.Fatalf("fault fixture did not leave a remnant: %v", err)
}
set.getDisks = original
if err := z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatal(err)
var invalid InvalidUploadID
if err := z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) {
t.Fatalf("minority retry must preserve default absent-upload result: %v", err)
}
if _, err := disks[0].ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, set.getUploadIDDir(bucket, "a", mp.UploadID), "", ReadOptions{}); !errors.Is(err, errFileNotFound) {
t.Fatalf("best-effort retry did not clean remnant: %v", err)
}
}
+256
View File
@@ -0,0 +1,256 @@
// Copyright (c) 2026 Ruohang Feng
// SPDX-License-Identifier: AGPL-3.0-or-later
package cmd
import (
"context"
"encoding/xml"
"errors"
"net/http"
"net/http/httptest"
"slices"
"testing"
)
func rewriteMultipartIdentity(t testing.TB, set *erasureObjects, bucket, key, id string, change func(map[string]string)) {
t.Helper()
fi, metadata, err := set.checkUploadIDExists(t.Context(), bucket, key, id, true)
if err != nil {
t.Fatal(err)
}
for i := range metadata {
change(metadata[i].Metadata)
}
if _, err := writeAllMetadata(t.Context(), set.getDisks(), bucket, minioMetaMultipartBucket,
set.getUploadIDDir(bucket, key, id), metadata, fi.WriteQuorum(set.defaultWQuorum())); err != nil {
t.Fatal(err)
}
}
func removeMultipartIdentity(metadata map[string]string) {
delete(metadata, multipartMetaBucket)
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)
}
for _, key := range []string{"dir/old", "dir/a", "dir/sub/b", "outside"} {
mp, err := z.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{})
if err != nil {
t.Fatal(err)
}
if key == "dir/old" {
rewriteMultipartIdentity(t, set, bucket, key, mp.UploadID, removeMultipartIdentity)
}
}
z.mpCache.Clear()
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)
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
}
var keys []string
keyMarker, uploadMarker := "", ""
for page := 0; ; page++ {
if page >= 4 {
t.Fatal("pagination did not terminate")
}
got := list("", keyMarker, uploadMarker, "")
if len(got.Uploads) != 1 {
t.Fatalf("page size: %+v", got)
}
keys = append(keys, got.Uploads[0].Key)
if !got.IsTruncated {
break
}
keyMarker, uploadMarker = got.NextKeyMarker, got.NextUploadIDMarker
}
if !slices.Equal(keys, []string{"dir/a", "dir/sub/b", "outside"}) {
t.Fatalf("mixed full listing: %v", keys)
}
first := list("dir/", "", "", "/")
if len(first.Uploads) != 1 || first.Uploads[0].Key != "dir/a" || !first.IsTruncated {
t.Fatalf("prefix first page: %+v", first)
}
rest := list("dir/", first.NextKeyMarker, first.NextUploadIDMarker, "/")
if len(rest.Uploads) != 0 || rest.IsTruncated || len(rest.CommonPrefixes) != 1 || rest.CommonPrefixes[0].Prefix != "dir/sub/" {
t.Fatalf("prefix continuation: %+v", rest)
}
if got := list("dir/old", "", "", ""); len(got.Uploads) != 0 || got.IsTruncated {
t.Fatalf("historical exact-looking prefix: %+v", got)
}
}
func TestMultipartListingInvalidIdentity(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)
mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
rewriteMultipartIdentity(t, set, bucket, "a", mp.UploadID, func(metadata map[string]string) {
switch kind {
case "missing-bucket":
delete(metadata, multipartMetaBucket)
case "missing-object":
delete(metadata, multipartMetaObject)
case "empty-bucket":
metadata[multipartMetaBucket] = ""
case "empty-object":
metadata[multipartMetaObject] = ""
case "empty-both":
metadata[multipartMetaBucket], metadata[multipartMetaObject] = "", ""
case "wrong-hash":
metadata[multipartMetaObject] = "wrong-key"
case "read-failure":
removeMultipartIdentity(metadata)
}
})
if kind == "read-failure" {
original := set.getDisks
t.Cleanup(func() { set.getDisks = original })
set.getDisks = func() []StorageAPI {
disks := append([]StorageAPI(nil), original()...)
for i, disk := range disks {
disks[i] = multipartListingFaultDisk{StorageAPI: disk, read: func(context.Context, string, string, string, string, ReadOptions) (FileInfo, error) {
return FileInfo{}, errFaultyDisk
}}
}
return disks
}
}
got, err := z.ListMultipartUploads(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)
}
if kind != "read-failure" && !errors.Is(err, errMultipartListingIdentity) {
t.Fatalf("identity error: %v", err)
}
report, err := z.multipartPreflight(t.Context())
if err != nil || report.Complete || report.Ready || report.LegacyUploads != 0 {
t.Fatalf("invalid/unreadable record counted as historical: %+v %v", report, err)
}
})
}
}
func TestMultipartListingRestart(t *testing.T) {
ctx, cancel := context.WithCancel(t.Context())
obj, dirs, err := prepareErasure(ctx, 4)
if err != nil {
cancel()
t.Fatal(err)
}
t.Cleanup(func() { cancel(); obj.Shutdown(context.Background()); removeRoots(dirs) })
const bucket, key = "multipart-restart", "dir/new"
if err := obj.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil {
t.Fatal(err)
}
mp, err := obj.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{})
if err != nil {
t.Fatal(err)
}
cancel()
if err := obj.Shutdown(context.Background()); err != nil {
t.Fatal(err)
}
reopened, _, err := initObjectLayer(t.Context(), mustGetPoolEndpoints(0, dirs...))
if err != nil {
t.Fatal(err)
}
obj = reopened
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)
if err != nil {
t.Fatal(err)
}
requireMultipartUploadKeys(t, got, key)
if got.Uploads[0].UploadID != mp.UploadID {
t.Fatal("restart changed upload identity")
}
}
func TestMultipartListingConflictingIdentityCopies(t *testing.T) {
for _, oldCopies := range []int{1, 3} {
z, set, bucket := multipartListingFixture(t)
mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
path := set.getUploadIDDir(bucket, "a", mp.UploadID)
for _, disk := range set.getDisks()[:oldCopies] {
fi, err := disk.ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, path, "", ReadOptions{})
if err != nil {
t.Fatal(err)
}
removeMultipartIdentity(fi.Metadata)
if err := disk.WriteMetadata(t.Context(), bucket, minioMetaMultipartBucket, path, fi); err != nil {
t.Fatal(err)
}
}
got, err := z.ListMultipartUploads(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) {
for _, historical := range []bool{false, true} {
z, set, bucket := multipartListingFixture(t)
mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
oldPath := set.getUploadIDDir(bucket, "a", mp.UploadID)
unsupportedPath := pathJoin(set.getMultipartSHADir(bucket, "a"), "unsupported-id")
for _, disk := range set.getDisks() {
fi, err := disk.ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, oldPath, "", ReadOptions{})
if err != nil {
t.Fatal(err)
}
if historical {
removeMultipartIdentity(fi.Metadata)
}
if err := disk.WriteMetadata(t.Context(), bucket, minioMetaMultipartBucket, unsupportedPath, fi); err != nil {
t.Fatal(err)
}
if err := disk.Delete(t.Context(), minioMetaMultipartBucket, oldPath, DeleteOptions{Recursive: true}); err != nil {
t.Fatal(err)
}
}
got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
if historical {
if err != nil || len(got.Uploads) != 0 {
t.Fatalf("unidentified historical ID: %+v %v", got, err)
}
} else if !errors.Is(err, errMultipartListingIdentity) {
t.Fatalf("invalid new ID silently treated as historical: %+v %v", got, err)
}
}
}
+13 -69
View File
@@ -414,7 +414,7 @@ func paginateMultipartUploads(uploads []MultipartInfo, prefix, keyMarker, upload
}
// listMultipartUploadsExact preserves the hashed exact-object lookup used by
// multipart write placement and by rolling-upgrade legacy mode.
// multipart write placement. Public listing always uses the durable scan.
func (er erasureObjects) listMultipartUploadsExact(ctx context.Context, bucket, object, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (result ListMultipartsInfo, err error) {
auditObjectErasureSet(ctx, "ListMultipartUploads", object, &er)
@@ -527,13 +527,10 @@ func (er erasureObjects) ListMultipartUploads(ctx context.Context, bucket, prefi
return ListMultipartsInfo{}, err
}
defer scan.close()
uploads, legacy, err := er.scanMultipartUploads(scan, bucket, 0, 0)
uploads, err := er.scanMultipartUploads(scan, bucket, 0, 0)
if err != nil {
return ListMultipartsInfo{}, err
}
if legacy {
return ListMultipartsInfo{}, errMultipartListingLegacy
}
return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil
}
@@ -1788,11 +1785,9 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
return fi.ToObjectInfo(bucket, object, opts.Versioned || opts.VersionSuspended), nil
}
// abortMultipartUpload retains read-quorum validation and best-effort cleanup
// in legacy mode. Strict mode requires majority deletion acknowledgements and
// permits retrying remnants below read quorum. Neither mode fences creation
// writes still executing after a storage timeout.
func (er erasureObjects) abortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions, legacy bool) (bool, error) {
// abortMultipartUpload retains read-quorum validation and best-effort cleanup.
// It does not fence creation writes still executing after a storage timeout.
func (er erasureObjects) abortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) (bool, error) {
if !opts.NoAuditLog {
auditObjectErasureSet(ctx, "AbortMultipartUpload", object, &er)
}
@@ -1804,70 +1799,19 @@ func (er erasureObjects) abortMultipartUpload(ctx context.Context, bucket, objec
if !ok || internalID == "" || internalID == "." || internalID == ".." || strings.ContainsAny(internalID, "/\\") {
return false, InvalidUploadID{Bucket: bucket, Object: object, UploadID: uploadID}
}
if legacy {
// Keep the released read-quorum and best-effort cleanup behavior.
// The upload ID safety check above applies to both modes.
defer er.deleteAll(ctx, minioMetaMultipartBucket, er.getUploadIDDir(bucket, object, uploadID))
_, _, err := er.checkUploadIDExists(ctx, bucket, object, uploadID, false)
err = toObjectErr(err, bucket, object, uploadID)
if _, absent := err.(InvalidUploadID); absent {
return false, nil
}
return err == nil, err
}
disks := er.getDisks()
uploadPath := er.getUploadIDDir(bucket, object, uploadID)
_, errs := readAllFileInfo(ctx, disks, bucket, minioMetaMultipartBucket, uploadPath, "", false, false)
quorum := er.setDriveCount/2 + 1
found, absent := false, 0
observed := make([]bool, len(disks))
for i, err := range errs {
switch {
case err == nil, errors.Is(err, errFileCorrupt):
found = true
observed[i] = true
case errors.Is(err, errFileNotFound), errors.Is(err, errFileVersionNotFound):
absent++
}
}
if absent >= quorum && !found {
defer er.deleteAll(ctx, minioMetaMultipartBucket, er.getUploadIDDir(bucket, object, uploadID))
_, _, err = er.checkUploadIDExists(ctx, bucket, object, uploadID, false)
err = toObjectErr(err, bucket, object, uploadID)
if _, absent := err.(InvalidUploadID); absent {
return false, nil
}
if !found {
return false, toObjectErr(errErasureReadQuorum, bucket, object, uploadID)
}
g := errgroup.WithNErrs(len(disks))
for i, disk := range disks {
g.Go(func() error {
if disk == nil {
return errDiskNotFound
}
err := disk.Delete(ctx, minioMetaMultipartBucket, uploadPath, DeleteOptions{Recursive: true, Immediate: false})
if errors.Is(err, errFileNotFound) || errors.Is(err, errFileVersionNotFound) {
return nil
}
return err
}, i)
}
deleteErrs := g.Wait()
if err := reduceWriteQuorumErrs(ctx, deleteErrs, nil, quorum); err != nil {
return true, toObjectErr(err, bucket, object, uploadID)
}
if absent >= quorum {
// Missing disks must not mask a failed cleanup of known remnants.
for i, found := range observed {
if found && deleteErrs[i] != nil {
return true, toObjectErr(errErasureWriteQuorum, bucket, object, uploadID)
}
}
}
return true, nil
return err == nil, err
}
// AbortMultipartUpload cancels an upload using the configured mode. Offline
// part data may still need stale-upload cleanup after its drives return.
// AbortMultipartUpload cancels an upload using read-quorum validation.
// Offline part data may still need stale-upload cleanup after its drives return.
func (er erasureObjects) AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) (err error) {
found, err := er.abortMultipartUpload(ctx, bucket, object, uploadID, opts, globalAPIConfig.getMultipartListingLegacy())
found, err := er.abortMultipartUpload(ctx, bucket, object, uploadID, opts)
if err != nil {
return err
}
+6 -70
View File
@@ -1916,9 +1916,6 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p
if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil {
return ListMultipartsInfo{}, toObjectErr(err, bucket)
}
if globalAPIConfig.getMultipartListingLegacy() {
return z.listMultipartUploadsLegacy(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
}
scan, err := startMultipartScan(ctx, false)
if err != nil {
return ListMultipartsInfo{}, err
@@ -1926,70 +1923,22 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p
defer scan.close()
var uploads []MultipartInfo
var keyless bool
for idx, pool := range z.serverPools {
if z.IsSuspended(idx) {
continue
}
poolUploads, poolKeyless, err := pool.scanMultipartUploads(scan, bucket, idx)
poolUploads, err := pool.scanMultipartUploads(scan, bucket, idx)
if err != nil {
return ListMultipartsInfo{}, err
}
uploads = append(uploads, poolUploads...)
keyless = keyless || poolKeyless
}
// The old format cannot be enumerated authoritatively. Migration mode is
// explicit: another bucket must never silently change this API's semantics.
if keyless {
return ListMultipartsInfo{}, errMultipartListingLegacy
}
// 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 (z *erasureServerPools) listMultipartUploadsLegacy(ctx context.Context, bucket, prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (ListMultipartsInfo, error) {
poolResult := ListMultipartsInfo{}
poolResult.MaxUploads = maxUploads
poolResult.KeyMarker = keyMarker
poolResult.Prefix = prefix
poolResult.Delimiter = delimiter
// if no prefix provided, return the list from cache
if prefix == "" {
if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil {
return ListMultipartsInfo{}, toObjectErr(err, bucket)
}
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
}
result, err := pool.getHashedSet(prefix).listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
if err != nil {
return result, err
}
poolResult.Uploads = append(poolResult.Uploads, result.Uploads...)
}
return poolResult, nil
}
// Initiate a new multipart upload on a hashedSet based on object name.
func (z *erasureServerPools) NewMultipartUpload(ctx context.Context, bucket, object string, opts ObjectOptions) (mp *NewMultipartUploadResult, err error) {
if err := checkNewMultipartArgs(ctx, bucket, object); err != nil {
@@ -2199,28 +2148,15 @@ func (z *erasureServerPools) AbortMultipartUpload(ctx context.Context, bucket, o
ctx = lkctx.Context()
defer lk.Unlock(lkctx)
legacy := globalAPIConfig.getMultipartListingLegacy()
found := false
var firstErr error
for idx, pool := range z.serverPools {
if z.IsSuspended(idx) {
continue
}
poolFound, err := pool.getHashedSet(object).abortMultipartUpload(ctx, bucket, object, uploadID, opts, legacy)
if legacy && (poolFound || err != nil) {
// Match the released first-matching-pool behavior.
found, err := pool.getHashedSet(object).abortMultipartUpload(ctx, bucket, object, uploadID, opts)
if found || err != nil {
// Preserve the default first-matching-pool behavior.
return err
}
found = found || poolFound
if err != nil && firstErr == nil {
firstErr = err
}
}
if firstErr != nil {
return firstErr
}
if found {
return nil
}
return InvalidUploadID{
Bucket: bucket,
+5 -10
View File
@@ -894,28 +894,23 @@ func (s *erasureSets) ListMultipartUploads(ctx context.Context, bucket, prefix,
return ListMultipartsInfo{}, err
}
defer scan.close()
uploads, legacy, err := s.scanMultipartUploads(scan, bucket, 0)
uploads, err := s.scanMultipartUploads(scan, bucket, 0)
if err != nil {
return ListMultipartsInfo{}, err
}
if legacy {
return ListMultipartsInfo{}, errMultipartListingLegacy
}
return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil
}
func (s *erasureSets) scanMultipartUploads(scan *multipartScan, bucket string, poolIdx int) ([]MultipartInfo, bool, error) {
func (s *erasureSets) scanMultipartUploads(scan *multipartScan, bucket string, poolIdx int) ([]MultipartInfo, error) {
var uploads []MultipartInfo
var keyless bool
for i, set := range s.sets {
setUploads, setKeyless, err := set.scanMultipartUploads(scan, bucket, poolIdx, i)
setUploads, err := set.scanMultipartUploads(scan, bucket, poolIdx, i)
if err != nil {
return nil, false, err
return nil, err
}
uploads = append(uploads, setUploads...)
keyless = keyless || setKeyless
}
return uploads, keyless, nil
return uploads, nil
}
func (s *erasureSets) listMultipartUploadsExact(ctx context.Context, bucket, object string) (ListMultipartsInfo, error) {
-8
View File
@@ -50,7 +50,6 @@ type apiConfig struct {
transitionWorkers int
staleUploadsExpiry time.Duration
multipartListingStrict bool
staleUploadsCleanupInterval time.Duration
deleteCleanupInterval time.Duration
enableODirect bool
@@ -182,7 +181,6 @@ func (t *apiConfig) init(cfg api.Config, setDriveCounts []int, legacy bool) {
t.transitionWorkers = cfg.TransitionWorkers
t.staleUploadsExpiry = cfg.StaleUploadsExpiry
t.multipartListingStrict = cfg.MultipartListing == "strict"
t.deleteCleanupInterval = cfg.DeleteCleanupInterval
t.enableODirect = cfg.EnableODirect
t.gzipObjects = cfg.GzipObjects
@@ -208,12 +206,6 @@ func (t *apiConfig) odirectEnabled() bool {
return t.enableODirect
}
func (t *apiConfig) getMultipartListingLegacy() bool {
t.mu.RLock()
defer t.mu.RUnlock()
return !t.multipartListingStrict
}
func (t *apiConfig) shouldGzipObjects() bool {
t.mu.RLock()
defer t.mu.RUnlock()
+19 -11
View File
@@ -19,7 +19,6 @@ package cmd
import (
"bytes"
"errors"
"fmt"
"slices"
"testing"
@@ -47,7 +46,7 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
t.Fatal(err)
}
z := obj.(*erasureServerPools)
setMultipartListingTestMode(t, false)
t.Cleanup(func() {
z.Shutdown(t.Context())
removeRoots(dirs)
@@ -182,8 +181,8 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
}
// Simulate an upload written by a pre-upgrade server. Its key cannot be
// recovered by scanning the hashed namespace. Strict listing must fail;
// the old path is available only through an explicit migration setting.
// recovered by scanning the hashed namespace. Omit only that record, and
// preserve normal operations when its bucket, key and upload ID are known.
legacyObject := objects[1]
er := sets.getHashedSet(legacyObject)
fi, metadata, err := er.checkUploadIDExists(t.Context(), bucket, legacyObject, uploadIDs[legacyObject], true)
@@ -198,16 +197,25 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
er.getUploadIDDir(bucket, legacyObject, uploadIDs[legacyObject]), metadata, fi.WriteQuorum(er.defaultWQuorum())); err != nil {
t.Fatal(err)
}
_, err = z.ListMultipartUploads(t.Context(), bucket, legacyObject, "", "", "", 100)
if !errors.Is(err, errMultipartListingLegacy) {
t.Fatalf("legacy strict listing: %v", err)
}
setMultipartListingTestMode(t, true)
legacy, err := z.ListMultipartUploads(t.Context(), bucket, legacyObject, "", "", "", 100)
if err != nil {
t.Fatal(err)
}
requireMultipartUploadKeys(t, legacy, legacyObject)
requireMultipartUploadKeys(t, legacy)
remaining, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 100)
if err != nil {
t.Fatal(err)
}
requireMultipartUploadKeys(t, remaining, objects[2:]...)
part, err = z.PutObjectPart(t.Context(), bucket, legacyObject, uploadIDs[legacyObject], 1,
mustGetPutObjReader(t, bytes.NewBufferString("old part"), 8, "", ""), ObjectOptions{})
if err != nil {
t.Fatalf("continue historical upload: %v", err)
}
if _, err = z.CompleteMultipartUpload(t.Context(), bucket, legacyObject, uploadIDs[legacyObject],
[]CompletePart{{PartNumber: 1, ETag: part.ETag}}, ObjectOptions{}); err != nil {
t.Fatalf("complete historical upload: %v", err)
}
}
func TestPaginateMultipartUploads(t *testing.T) {
@@ -260,7 +268,7 @@ func TestPaginateMultipartUploads(t *testing.T) {
func TestListMultipartUploadsGlobalPageAcrossPools(t *testing.T) {
z, bucket := consistencyPools(t)
setMultipartListingTestMode(t, false)
objects := []string{"a/one", "b/two", "c/three", "d/four"}
for i, object := range objects {
if _, err := z.serverPools[i%len(z.serverPools)].NewMultipartUpload(t.Context(), bucket, object, ObjectOptions{}); err != nil {
+2 -9
View File
@@ -68,7 +68,7 @@ const (
EnvAPIStaleUploadsCleanupInterval = "MINIO_API_STALE_UPLOADS_CLEANUP_INTERVAL"
EnvAPIStaleUploadsExpiry = "MINIO_API_STALE_UPLOADS_EXPIRY"
EnvAPIMultipartListing = "MINIO_API_MULTIPART_LISTING"
EnvAPIMultipartListing = "MINIO_API_MULTIPART_LISTING" // Retired; used only for a startup warning.
EnvAPIDeleteCleanupInterval = "MINIO_API_DELETE_CLEANUP_INTERVAL"
EnvDeleteCleanupInterval = "MINIO_DELETE_CLEANUP_INTERVAL"
EnvAPIODirect = "MINIO_API_ODIRECT"
@@ -180,7 +180,6 @@ type Config struct {
TransitionWorkers int `json:"transition_workers"`
StaleUploadsCleanupInterval time.Duration `json:"stale_uploads_cleanup_interval"`
StaleUploadsExpiry time.Duration `json:"stale_uploads_expiry"`
MultipartListing string `json:"multipart_listing"`
DeleteCleanupInterval time.Duration `json:"delete_cleanup_interval"`
EnableODirect bool `json:"enable_odirect"`
GzipObjects bool `json:"gzip_objects"`
@@ -209,7 +208,7 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) {
apiReplicationWorkers,
apiReplicationFailedWorkers,
"expiry_workers",
apiMultipartListing, // Ignore the retired shared-config key during migration.
apiMultipartListing, // Tolerate the retired key without changing other API settings.
}
disableODirect := env.Get(EnvAPIDisableODirect, kvs.Get(apiDisableODirect)) == config.EnableOn
@@ -344,11 +343,5 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) {
cfg.ObjectMaxVersions = math.MaxInt64
}
cfg.MultipartListing = env.Get(EnvAPIMultipartListing, "legacy")
if cfg.MultipartListing != "strict" && cfg.MultipartListing != "legacy" {
cfg.MultipartListing = "legacy"
return cfg, fmt.Errorf("%s must be strict or legacy; using legacy", EnvAPIMultipartListing)
}
return cfg, nil
}
+17 -18
View File
@@ -11,22 +11,21 @@ import (
"github.com/minio/minio/internal/config"
)
func TestMultipartListingMigrationMode(t *testing.T) {
func TestMultipartListingRetiredConfiguration(t *testing.T) {
t.Setenv(EnvAPIRequestsMax, "")
t.Setenv(EnvAPISyncEvents, "")
t.Setenv(EnvAPIObjectMaxVersions, "")
for _, tc := range []struct {
name, stored, override, want string
invalid bool
name, stored, override string
}{
{name: "default", want: "legacy"},
{name: "explicit-legacy", override: "legacy", want: "legacy"},
{name: "explicit-strict", override: "strict", want: "strict"},
{name: "retired-stored-strict", stored: "strict", want: "legacy"},
{name: "retired-stored-legacy", stored: "legacy", want: "legacy"},
{name: "retired-stored-invalid", stored: "automatic", want: "legacy"},
{name: "environment-override", stored: "legacy", override: "strict", want: "strict"},
{name: "invalid-environment", stored: "strict", override: "automatic", want: "legacy", invalid: true},
{name: "default"},
{name: "explicit-legacy", override: "legacy"},
{name: "explicit-strict", override: "strict"},
{name: "retired-stored-strict", stored: "strict"},
{name: "retired-stored-legacy", stored: "legacy"},
{name: "retired-stored-invalid", stored: "automatic"},
{name: "environment-override", stored: "legacy", override: "strict"},
{name: "invalid-environment", stored: "strict", override: "automatic"},
} {
t.Run(tc.name, func(t *testing.T) {
t.Setenv(EnvAPIMultipartListing, tc.override)
@@ -38,11 +37,11 @@ func TestMultipartListingMigrationMode(t *testing.T) {
kvs.Set(apiMultipartListing, tc.stored)
}
cfg, err := LookupConfig(kvs)
if (err != nil) != tc.invalid || cfg.MultipartListing != tc.want {
t.Fatalf("mode=%q err=%v, want %q invalid=%v", cfg.MultipartListing, err, tc.want, tc.invalid)
if err != nil {
t.Fatalf("retired setting must be ignored: %v", err)
}
if cfg.RequestsMax != 17 || !cfg.SyncEvents || cfg.ObjectMaxVersions != 71 {
t.Fatalf("migration setting discarded unrelated config: %+v", cfg)
t.Fatalf("retired setting discarded unrelated config: %+v", cfg)
}
})
}
@@ -58,7 +57,7 @@ func TestMultipartListingConfigPersistence(t *testing.T) {
t.Cleanup(func() { config.DefaultKVS = previous })
for _, help := range Help {
if help.Key == apiMultipartListing {
t.Fatal("process-only setting advertised as shared config")
t.Fatal("retired setting advertised as shared config")
}
}
for _, input := range []string{
@@ -82,11 +81,11 @@ func TestMultipartListingConfigPersistence(t *testing.T) {
}
kvs := loaded.Merge()[config.APISubSys][config.Default]
if _, present := kvs.Lookup(apiMultipartListing); present {
t.Fatal("process-only setting persisted in shared config")
t.Fatal("retired setting persisted in shared config")
}
cfg, err := LookupConfig(kvs)
if err != nil || cfg.MultipartListing != "strict" {
t.Fatalf("environment mode lost: %+v %v", cfg, err)
if err != nil {
t.Fatalf("retired environment setting caused failure: %+v %v", cfg, err)
}
if input != "" && cfg.RequestsMax != 17 {
t.Fatalf("unrelated setting lost: %+v", cfg)