fix: preserve multipart defaults and rollback compatibility

Signed-off-by: Feng Ruohang <rh@vonng.com>

CHANGELOG.md

@@ -29,7 +29,7 @@ and [complete commit range](https://github.com/pgsty/silo/compare/RELEASE.2026-0
   recreated parents and explicitly reconcile pre-upgrade revocations whose
   history is already lost. Restoring an older backup can lose later revocations;
   keep affected sites isolated until reconciliation/rekeying is complete. See
-  [the operator runbook](https://github.com/pgsty/silo.pgsty.com/blob/29c7f220b3acc556ad570694056d35e11246f1b9/content/operations/replication/iam-upgrade.md).
+  [the operator runbook](https://github.com/pgsty/silo.pgsty.com/blob/7bd2d57c2ce5aaa804d0b1a2fe0e5eed69d15235/content/operations/replication/iam-upgrade.md).
 - Enforce an absolute HTTP/1 request-header deadline through the connection
   wrapper (#196). Repeated small reads no longer extend that deadline, and
   `--read-header-timeout` / `MINIO_READ_HEADER_TIMEOUT` now reaches the HTTP
@@ -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.
@@ -91,7 +100,7 @@ and [complete commit range](https://github.com/pgsty/silo/compare/RELEASE.2026-0
   ordinary metadata. Thanks to Mikhail Khadarenka (@chodorenko) for the fix in #187.
   **Existing data:** these repairs prevent new errors; they do not scan or rewrite
   historical object metadata, recover lost tags or prove that old purge work has
-  converged. Follow the [read-only audit procedure](https://github.com/pgsty/silo.pgsty.com/blob/29c7f220b3acc556ad570694056d35e11246f1b9/content/operations/replication/replica-metadata-audit.md)
+  converged. Follow the [read-only audit procedure](https://github.com/pgsty/silo.pgsty.com/blob/7bd2d57c2ce5aaa804d0b1a2fe0e5eed69d15235/content/operations/replication/replica-metadata-audit.md)
   before planning any repair of stored state.
 
 - Evaluate conditional multipart completion against the logical current object
@@ -104,9 +113,20 @@ and [complete commit range](https://github.com/pgsty/silo/compare/RELEASE.2026-0
   applies when the unreadable pool may not hold the object: absence cannot be
   verified. Retry after the pool recovers. Unconditional completion and the
   single-pool path retain their existing behavior.
-  Ordinary conditional PUT has a separate cross-pool precondition gap tracked
-  in [#199](https://github.com/pgsty/silo/issues/199); the multipart repair does
-  not resolve it.
+
+- Evaluate ordinary multi-pool conditional PUT against the logical current
+  object across all pools, including draining pools, under the existing object
+  lock (#207). A stale destination copy no longer accepts a stale ETag or rejects
+  the current one; a current delete marker is treated as absence.
+  **Availability change:** if any pool's object metadata cannot be verified,
+  the condition fails even when GET can use another pool; read-quorum failures
+  return 503. Restore readability or heal before retrying. Unconditional PUT,
+  single-pool conditions and internal replication retain their existing behavior.
+  A public condition with a destination `versionId` compares the current object
+  while preserving the requested write version. This change does not retire
+  stale copies in other pools, undo historical accepted overwrites or provide
+  a new global clock-ordering guarantee. The multipart-completion repair in #190
+  neither introduced nor repaired this separate PUT defect.
 
 - Reconcile ordinary single-object version DELETE across all pools, including
   null versions, delete markers and unqualified directory-marker DELETE. This

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 {

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)
+	}
+}

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
 	}

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

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 {

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 {

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
 }

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)
+	}
+}

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)`,