From 82f0a9828e4d1d7b5d42d3f2ccdea9bb64d46f30 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Wed, 16 Sep 2026 16:30:43 +0800 Subject: [PATCH] fix: preserve multipart defaults and rollback compatibility Signed-off-by: Feng Ruohang --- CHANGELOG.md | 27 +- cmd/erasure-multipart-listing_test.go | 23 ++ cmd/erasure-multipart-migration_test.go | 331 ++++++++++++++++++++++ cmd/erasure-multipart.go | 49 +++- cmd/erasure-server-pool.go | 16 +- cmd/handler-api.go | 6 +- cmd/list-multipart-uploads-compat_test.go | 12 +- internal/config/api/api.go | 12 +- internal/config/api/api_test.go | 101 ++++++- internal/config/api/help.go | 6 - 10 files changed, 520 insertions(+), 63 deletions(-) create mode 100644 cmd/erasure-multipart-migration_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index e6f1810f2..47995b93a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -56,19 +56,28 @@ and [complete commit range](https://github.com/pgsty/silo/compare/RELEASE.2026-0 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; - `api multipart_listing=legacy` is an explicit temporary migration mode. - Upgrade every writer, drain old uploads and check the read-only admin - `multipart-preflight` report before relying on strict listing. Per-process + 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. -- Confirm multipart cancellation on a strict majority of each relevant set, - and allow retries after partial deletion. Uncertain pools or insufficient - confirmations return 503 rather than acknowledging a cancellation whose - static remnants can later become readable. **Known boundary:** creation - writes that finish after a storage timeout can still restore an upload after - successful cancellation; this change does not add a durable creation fence. +- 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. - 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. diff --git a/cmd/erasure-multipart-listing_test.go b/cmd/erasure-multipart-listing_test.go index cf71a6371..c669a6123 100644 --- a/cmd/erasure-multipart-listing_test.go +++ b/cmd/erasure-multipart-listing_test.go @@ -31,6 +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{}) @@ -147,6 +148,22 @@ 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) + 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) + } + 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) @@ -214,6 +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) @@ -286,6 +304,7 @@ func TestMultipartListingMarkerHTTP(t *testing.T) { if err != nil { t.Fatal(err) } + setMultipartListingTestMode(t, false) for _, tc := range []struct { key, marker string status int @@ -369,6 +388,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}}) @@ -454,6 +474,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} { @@ -523,6 +544,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 } @@ -673,6 +695,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 { diff --git a/cmd/erasure-multipart-migration_test.go b/cmd/erasure-multipart-migration_test.go new file mode 100644 index 000000000..db74fe432 --- /dev/null +++ b/cmd/erasure-multipart-migration_test.go @@ -0,0 +1,331 @@ +// Copyright (c) 2026 Ruohang Feng +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "context" + "encoding/base64" + "errors" + "fmt" + "strings" + "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) { + 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) { + obj, dirs, err := prepareErasure(t.Context(), drives) + if err != nil { + t.Fatal(err) + } + 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) }) + const bucket, key = "multipart-legacy-availability", "keep-available" + if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil { + t.Fatal(err) + } + mp, err := z.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + set := z.serverPools[0].getHashedSet(key) + original := set.getDisks + disks := original() + set.getDisks = func() []StorageAPI { + visible := append([]StorageAPI(nil), disks...) + clear(visible[:offline]) + return visible + } + t.Cleanup(func() { set.getDisks = original }) + if err := z.AbortMultipartUpload(t.Context(), bucket, key, mp.UploadID, ObjectOptions{}); err != nil { + t.Fatalf("released read-quorum availability regressed: %v", err) + } + for _, disk := range disks[offline:] { + _, err := disk.ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, set.getUploadIDDir(bucket, key, mp.UploadID), "", ReadOptions{}) + if !errors.Is(err, errFileNotFound) { + t.Fatalf("online replica not cleaned: %v", err) + } + } + }) + } + } +} + +func TestMultipartAbortLegacyPoolOrder(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) + } + emptySet := z.serverPools[1-owner].getHashedSet("a") + original := emptySet.getDisks + t.Cleanup(func() { emptySet.getDisks = original }) + 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) + } + if owner == 1 { + if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 { + t.Fatalf("unknown earlier pool must retain released error behavior: %v", err) + } + if _, err := z.serverPools[owner].GetMultipartInfo(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil { + t.Fatalf("later pool visited despite earlier error: %v", err) + } + emptySet.getDisks = original + if err := z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil { + t.Fatal(err) + } + } + }) + } +} + +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) + } + }) + } +} + +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) + } + } + }) + } +} + +func TestMultipartAbortPeerNotification(t *testing.T) { + z, set, bucket := multipartListingFixture(t) + tg, err := grid.SetupTestGrid(2) + if err != nil { + t.Fatal(err) + } + t.Cleanup(tg.Cleanup) + var notifications atomic.Int32 + if err := cleanupUploadIDCacheMetaRPC.Register(tg.Managers[1], func(*grid.MSS) (grid.NoPayload, *grid.RemoteErr) { + notifications.Add(1) + return grid.NoPayload{}, nil + }); err != nil { + t.Fatal(err) + } + host, err := xnet.ParseHost(strings.TrimPrefix(tg.Hosts[1], "http://")) + if err != nil { + t.Fatal(err) + } + previous := globalNotificationSys + globalNotificationSys = &NotificationSys{peerClients: []*peerRESTClient{{ + host: host, + gridConn: func() *grid.Connection { return tg.Managers[0].Connection(tg.Hosts[1]) }, + }}} + t.Cleanup(func() { globalNotificationSys = previous }) + // Use the real distributed lock implementation: Unlock cancels its derived + // context. Local locks do not, and would hide a broken notification context. + oldMutex, oldLockers := set.nsMutex, set.getLockers + lockers := []dsync.NetLocker{newLocker(), newLocker()} + 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") + } + } + }) + } +} + +func TestMultipartAbortStrictMajority(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 + disks := original() + 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 }) + if err := z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil { + t.Fatalf("ordinary majority 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. + 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) + } +} diff --git a/cmd/erasure-multipart.go b/cmd/erasure-multipart.go index bccd16001..89355c762 100644 --- a/cmd/erasure-multipart.go +++ b/cmd/erasure-multipart.go @@ -1753,11 +1753,11 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str return fi.ToObjectInfo(bucket, object, opts.Versioned || opts.VersionSuspended), nil } -// abortMultipartUpload confirms absence on a strict majority of this set. -// Unlike an existence read followed by best-effort deletion, it also permits -// retrying a partial deletion which no longer has a readable metadata quorum. -// This 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) { +// 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) { if !opts.NoAuditLog { auditObjectErasureSet(ctx, "AbortMultipartUpload", object, &er) } @@ -1769,21 +1769,34 @@ 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 - for _, err := range errs { + 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 { - return found, nil + if absent >= quorum && !found { + return false, nil } if !found { return false, toObjectErr(errErasureReadQuorum, bucket, object, uploadID) @@ -1801,13 +1814,25 @@ func (er erasureObjects) abortMultipartUpload(ctx context.Context, bucket, objec return err }, i) } - return true, toObjectErr(reduceWriteQuorumErrs(ctx, g.Wait(), nil, quorum), bucket, object, uploadID) + 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 } -// AbortMultipartUpload confirms logical cancellation. Offline part data may -// still need stale-upload cleanup after its drives return. +// AbortMultipartUpload cancels an upload using the configured mode. 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) + found, err := er.abortMultipartUpload(ctx, bucket, object, uploadID, opts, globalAPIConfig.getMultipartListingLegacy()) if err != nil { return err } diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index bdc3bfedd..4f9b43cfc 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -2160,13 +2160,14 @@ func (z *erasureServerPools) AbortMultipartUpload(ctx context.Context, bucket, o return toObjectErr(err, bucket) } - defer func() { - _, absent := err.(InvalidUploadID) - if err == nil || absent { + // Unlock cancels the derived lock context before this notification runs. + // Keep the request context so successful cancellation reaches peer caches. + defer func(ctx context.Context) { + if err == nil { z.mpCache.Delete(uploadID) globalNotificationSys.DeleteUploadID(ctx, uploadID) } - }() + }(ctx) lk := z.NewNSLock(bucket, pathJoin(object, uploadID)) lkctx, err := lk.GetLock(ctx, globalOperationTimeout) @@ -2176,13 +2177,18 @@ 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) + poolFound, err := pool.getHashedSet(object).abortMultipartUpload(ctx, bucket, object, uploadID, opts, legacy) + if legacy && (poolFound || err != nil) { + // Match the released first-matching-pool behavior. + return err + } found = found || poolFound if err != nil && firstErr == nil { firstErr = err diff --git a/cmd/handler-api.go b/cmd/handler-api.go index acd8f97af..0b33bbcab 100644 --- a/cmd/handler-api.go +++ b/cmd/handler-api.go @@ -50,7 +50,7 @@ type apiConfig struct { transitionWorkers int staleUploadsExpiry time.Duration - multipartListingLegacy bool + multipartListingStrict bool staleUploadsCleanupInterval time.Duration deleteCleanupInterval time.Duration enableODirect bool @@ -182,7 +182,7 @@ func (t *apiConfig) init(cfg api.Config, setDriveCounts []int, legacy bool) { t.transitionWorkers = cfg.TransitionWorkers t.staleUploadsExpiry = cfg.StaleUploadsExpiry - t.multipartListingLegacy = cfg.MultipartListing == "legacy" + t.multipartListingStrict = cfg.MultipartListing == "strict" t.deleteCleanupInterval = cfg.DeleteCleanupInterval t.enableODirect = cfg.EnableODirect t.gzipObjects = cfg.GzipObjects @@ -211,7 +211,7 @@ func (t *apiConfig) odirectEnabled() bool { func (t *apiConfig) getMultipartListingLegacy() bool { t.mu.RLock() defer t.mu.RUnlock() - return t.multipartListingLegacy + return !t.multipartListingStrict } func (t *apiConfig) shouldGzipObjects() bool { diff --git a/cmd/list-multipart-uploads-compat_test.go b/cmd/list-multipart-uploads-compat_test.go index 49aab94b3..4ff4c069f 100644 --- a/cmd/list-multipart-uploads-compat_test.go +++ b/cmd/list-multipart-uploads-compat_test.go @@ -47,6 +47,7 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) { t.Fatal(err) } z := obj.(*erasureServerPools) + setMultipartListingTestMode(t, false) t.Cleanup(func() { z.Shutdown(t.Context()) removeRoots(dirs) @@ -201,15 +202,7 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) { if !errors.Is(err, errMultipartListingLegacy) { t.Fatalf("legacy strict listing: %v", err) } - globalAPIConfig.mu.Lock() - oldLegacy := globalAPIConfig.multipartListingLegacy - globalAPIConfig.multipartListingLegacy = true - globalAPIConfig.mu.Unlock() - t.Cleanup(func() { - globalAPIConfig.mu.Lock() - globalAPIConfig.multipartListingLegacy = oldLegacy - globalAPIConfig.mu.Unlock() - }) + setMultipartListingTestMode(t, true) legacy, err := z.ListMultipartUploads(t.Context(), bucket, legacyObject, "", "", "", 100) if err != nil { t.Fatal(err) @@ -267,6 +260,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 { diff --git a/internal/config/api/api.go b/internal/config/api/api.go index 26d479873..0337f1008 100644 --- a/internal/config/api/api.go +++ b/internal/config/api/api.go @@ -135,7 +135,6 @@ var ( Key: apiStaleUploadsExpiry, Value: "24h", }, - config.KV{Key: apiMultipartListing, Value: "strict"}, config.KV{ Key: apiDeleteCleanupInterval, Value: "5m", @@ -210,6 +209,7 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) { apiReplicationWorkers, apiReplicationFailedWorkers, "expiry_workers", + apiMultipartListing, // Ignore the retired shared-config key during migration. } disableODirect := env.Get(EnvAPIDisableODirect, kvs.Get(apiDisableODirect)) == config.EnableOn @@ -324,10 +324,6 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) { return cfg, err } cfg.StaleUploadsExpiry = staleUploadsExpiry - cfg.MultipartListing = env.Get(EnvAPIMultipartListing, kvs.GetWithDefault(apiMultipartListing, DefaultKVS)) - if cfg.MultipartListing != "strict" && cfg.MultipartListing != "legacy" { - return cfg, fmt.Errorf("%s must be strict or legacy", apiMultipartListing) - } cfg.SyncEvents = env.Get(EnvAPISyncEvents, kvs.Get(apiSyncEvents)) == config.EnableOn @@ -348,5 +344,11 @@ 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 } diff --git a/internal/config/api/api_test.go b/internal/config/api/api_test.go index 0c659f65b..ad8f8697a 100644 --- a/internal/config/api/api_test.go +++ b/internal/config/api/api_test.go @@ -4,38 +4,111 @@ package api import ( + "encoding/json" + "strings" "testing" "github.com/minio/minio/internal/config" ) func TestMultipartListingMigrationMode(t *testing.T) { + t.Setenv(EnvAPIRequestsMax, "") + t.Setenv(EnvAPISyncEvents, "") + t.Setenv(EnvAPIObjectMaxVersions, "") for _, tc := range []struct { name, stored, override, want string invalid bool }{ - {name: "default", want: "strict"}, - {name: "explicit-migration", stored: "legacy", want: "legacy"}, - {name: "environment-override", stored: "strict", override: "legacy", want: "legacy"}, - {name: "invalid-stored", stored: "automatic", invalid: true}, - {name: "invalid-environment", stored: "strict", override: "automatic", invalid: true}, + {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}, } { t.Run(tc.name, func(t *testing.T) { t.Setenv(EnvAPIMultipartListing, tc.override) - var kvs config.KVS + kvs := DefaultKVS.Clone() + kvs.Set(apiRequestsMax, "17") + kvs.Set(apiSyncEvents, config.EnableOn) + kvs.Set(apiObjectMaxVersions, "71") if tc.stored != "" { - kvs = config.KVS{{Key: apiMultipartListing, Value: tc.stored}} + kvs.Set(apiMultipartListing, tc.stored) } cfg, err := LookupConfig(kvs) - if tc.invalid { - if err == nil { - t.Fatal("invalid migration mode silently accepted") - } - return + 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 || cfg.MultipartListing != tc.want { - t.Fatalf("mode=%q err=%v, want %q", cfg.MultipartListing, err, tc.want) + if cfg.RequestsMax != 17 || !cfg.SyncEvents || cfg.ObjectMaxVersions != 71 { + t.Fatalf("migration setting discarded unrelated config: %+v", cfg) } }) } } + +func TestMultipartListingConfigPersistence(t *testing.T) { + t.Setenv(EnvAPIRequestsMax, "") + t.Setenv(EnvAPISyncEvents, "") + t.Setenv(EnvAPIObjectMaxVersions, "") + t.Setenv(EnvAPIMultipartListing, "strict") + previous := config.DefaultKVS + config.DefaultKVS = map[string]config.KVS{config.APISubSys: DefaultKVS} + t.Cleanup(func() { config.DefaultKVS = previous }) + for _, help := range Help { + if help.Key == apiMultipartListing { + t.Fatal("process-only setting advertised as shared config") + } + } + for _, input := range []string{ + "", + "api requests_max=17 sync_events=on object_max_versions=71", + `api requests_max=17 comment="later disable multipart_listing=strict"`, + } { + t.Run(input, func(t *testing.T) { + c := config.New().Clone() + if _, err := c.ReadConfig(strings.NewReader(input)); err != nil { + t.Fatal(err) + } + // Exercise the JSON save/load and Merge paths used for shared config. + data, err := json.Marshal(c) + if err != nil { + t.Fatal(err) + } + var loaded config.Config + if err = json.Unmarshal(data, &loaded); err != nil { + t.Fatal(err) + } + kvs := loaded.Merge()[config.APISubSys][config.Default] + if _, present := kvs.Lookup(apiMultipartListing); present { + t.Fatal("process-only 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 input != "" && cfg.RequestsMax != 17 { + t.Fatalf("unrelated setting lost: %+v", cfg) + } + if strings.Contains(input, "comment=") && kvs.Get(config.Comment) != "later disable multipart_listing=strict" { + t.Fatalf("literal comment changed: %q", kvs.Get(config.Comment)) + } + }) + } + // A key persisted by a previous development build is tolerated, retained for + // explicit cleanup, and removable without resetting other API settings. + c := config.New().Clone() + kvs := c[config.APISubSys][config.Default] + kvs.Set(apiMultipartListing, "strict") + kvs.Set(apiRequestsMax, "17") + c[config.APISubSys][config.Default] = kvs + c = c.Merge() + if err := c.DelKVS("api multipart_listing"); err != nil { + t.Fatal(err) + } + kvs = c.Merge()[config.APISubSys][config.Default] + if _, present := kvs.Lookup(apiMultipartListing); present || kvs.Get(apiRequestsMax) != "17" { + t.Fatalf("targeted reset changed unrelated config or restored retired key: %v", kvs) + } +} diff --git a/internal/config/api/help.go b/internal/config/api/help.go index 650a4064d..2c4b8b2bb 100644 --- a/internal/config/api/help.go +++ b/internal/config/api/help.go @@ -26,12 +26,6 @@ var ( // Help holds configuration keys and their default values for api subsystem. Help = config.HelpKVS{ - config.HelpKV{ - Key: apiMultipartListing, - Description: "multipart listing mode: strict, or temporary legacy mode during coordinated upgrade" + defaultHelpPostfix(apiMultipartListing), - Type: "string", - Optional: true, - }, config.HelpKV{ Key: apiRequestsMax, Description: `set the maximum number of concurrent requests (default: auto)`,