Merge pull request #198 from mrjavadseydi/fix/issue-79-list-multipart-uploads

Retain the original contributor commit and integrate the reviewed durable listing, cancellation confirmation and migration fixes. Capacity acceptance and delayed creation-write fencing remain open in issue #79.

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

.github/workflows/go.yml

@@ -93,6 +93,9 @@ jobs:
       - name: Run conditional PUT tests under race detector
         run: go test -race ./cmd -run '^Test(PoolsConditionalPut|SinglePoolConditionalPutHTTP)' -count=1 -timeout=5m
 
+      - name: Run multipart listing and cancellation tests under race detector
+        run: go test -race ./cmd -run '^Test(MultipartListing|MultipartAbort|PaginateMultipartUploads|ListMultipartUploads)' -count=1 -timeout=5m
+
   crosscompile:
     name: Cross Compile
     runs-on: ubuntu-latest

CHANGELOG.md

@@ -49,6 +49,26 @@ 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;
+  `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
+  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.
 - 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.

buildscripts/rebrand-guard/compat-baseline.json

@@ -134,6 +134,7 @@
     "MINIO_API_GZIP_OBJECTS",
     "MINIO_API_LEGACY_BUCKET_RESOURCE_MATCH",
     "MINIO_API_LIST_QUORUM",
+    "MINIO_API_MULTIPART_LISTING",
     "MINIO_API_OBJECT_MAX_VERSIONS",
     "MINIO_API_ODIRECT",
     "MINIO_API_REMOTE_TRANSPORT_DEADLINE",
@@ -766,6 +767,7 @@
     "/metrics/v3",
     "/minio/grid/",
     "/minio/grid/lock/",
+    "/multipart-preflight",
     "/netperf",
     "/notification",
     "/oauth2/callback",

cmd/admin-router.go

@@ -163,6 +163,7 @@ func registerAdminRouter(router *mux.Router, enableConfigOps bool) {
 
 		// StorageInfo operations
 		adminRouter.Methods(http.MethodGet).Path(adminVersion + "/storageinfo").HandlerFunc(adminMiddleware(adminAPI.StorageInfoHandler, traceAllFlag))
+		adminRouter.Methods(http.MethodGet).Path(adminVersion + "/multipart-preflight").HandlerFunc(adminMiddleware(adminAPI.MultipartPreflightHandler, traceAllFlag))
 		// DataUsageInfo operations
 		adminRouter.Methods(http.MethodGet).Path(adminVersion + "/datausageinfo").HandlerFunc(adminMiddleware(adminAPI.DataUsageInfoHandler, traceAllFlag))
 		// Metrics operation

cmd/api-errors.go

@@ -450,6 +450,9 @@ const (
 	ErrAdminNoSecretKey
 
 	ErrIAMNotInitialized
+	ErrMultipartListingLegacy
+	ErrMultipartListingIdentity
+	ErrSlowDown
 
 	apiErrCodeEnd // This is used only for the testing code
 )
@@ -1336,6 +1339,21 @@ var errorCodes = errorCodeMap{
 		Description:    "IAM sub-system not initialized yet, please try again.",
 		HTTPStatusCode: http.StatusServiceUnavailable,
 	},
+	ErrMultipartListingLegacy: {
+		Code:           "MultipartListingNotReady",
+		Description:    "Legacy multipart uploads prevent a complete listing. Upgrade all writers, drain old uploads and run the multipart preflight check.",
+		HTTPStatusCode: http.StatusServiceUnavailable,
+	},
+	ErrMultipartListingIdentity: {
+		Code:           "MultipartListingMetadataInvalid",
+		Description:    "Multipart upload metadata is inconsistent. Run the multipart preflight check to locate the affected storage set.",
+		HTTPStatusCode: http.StatusServiceUnavailable,
+	},
+	ErrSlowDown: {
+		Code:           "SlowDown",
+		Description:    "Please reduce your request rate",
+		HTTPStatusCode: http.StatusServiceUnavailable,
+	},
 	ErrBucketMetadataNotInitialized: {
 		Code:           "XMinioBucketMetadataNotInitialized",
 		Description:    "Bucket metadata not initialized yet, please try again.",
@@ -2173,6 +2191,10 @@ 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:
 		apiErr = ErrBadDigest
 	case errMissingPartChecksum:
@@ -2303,6 +2325,8 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
 	}
 
 	switch err.(type) {
+	case SlowDown:
+		apiErr = ErrSlowDown
 	case StorageFull:
 		apiErr = ErrStorageFull
 	case hash.BadDigest:

cmd/api-response.go

@@ -41,7 +41,7 @@ import (
 const (
 	maxObjectList  = 1000  // Limit number of objects in a listObjectsResponse/listObjectsVersionsResponse.
 	maxDeleteList  = 1000  // Limit number of objects deleted in a delete call.
-	maxUploadsList = 10000 // Limit number of uploads in a listUploadsResponse.
+	maxUploadsList = 1000  // Limit number of uploads in a listUploadsResponse.
 	maxPartsList   = 10000 // Limit number of parts in a listPartsResponse.
 )
 

cmd/bucket-handlers.go

@@ -277,14 +277,6 @@ func (api objectAPIHandlers) ListMultipartUploadsHandler(w http.ResponseWriter,
 		return
 	}
 
-	if keyMarker != "" {
-		// Marker not common with prefix is not implemented.
-		if !HasPrefix(keyMarker, prefix) {
-			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
-			return
-		}
-	}
-
 	listMultipartsInfo, err := objectAPI.ListMultipartUploads(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
 	if err != nil {
 		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)

cmd/bucket-handlers_test.go

@@ -425,7 +425,7 @@ func testListMultipartUploadsHandler(obj ObjectLayer, instanceType, bucketName s
 			shouldPass:         true,
 		},
 		// Test case - 4.
-		// Setting Invalid prefix and marker combination.
+		// A key marker outside the prefix is valid and produces an empty page.
 		{
 			bucket:             bucketName,
 			prefix:             "asia",
@@ -435,8 +435,8 @@ func testListMultipartUploadsHandler(obj ObjectLayer, instanceType, bucketName s
 			maxUploads:         "0",
 			accessKey:          credentials.AccessKey,
 			secretKey:          credentials.SecretKey,
-			expectedRespStatus: http.StatusNotImplemented,
-			shouldPass:         false,
+			expectedRespStatus: http.StatusOK,
+			shouldPass:         true,
 		},
 		// Test case - 5.
 		// Invalid upload id and marker combination.

cmd/erasure-multipart-listing.go

@@ -0,0 +1,385 @@
+// Copyright (c) 2026 mr javad seydi and Ruohang Feng
+//
+// This file is part of Silo Object Storage stack.
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/pgsty/silo-pkg/v3/policy"
+)
+
+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")
+)
+
+// One budget and admission slot cover the entire request, including all pools.
+// A slot is released only after the scan workers have actually stopped.
+type multipartScan struct {
+	ctx           context.Context
+	cancel        context.CancelFunc
+	remaining     atomic.Int64
+	metadataSlots chan struct{}
+	preflight     bool
+	sets          []multipartScanSet
+}
+
+type multipartScanSet struct {
+	Pool            int       `json:"pool"`
+	Set             int       `json:"set"`
+	Drives          int       `json:"drives"`
+	ScannedDrives   int       `json:"scannedDrives"`
+	UncoveredDrives []int     `json:"uncoveredDrives,omitempty"`
+	Candidates      int       `json:"candidates"`
+	LegacyUploads   int       `json:"legacyUploads"`
+	OldestLegacy    time.Time `json:"oldestLegacy,omitempty"`
+	Error           string    `json:"error,omitempty"`
+}
+
+type multipartPreflightReport struct {
+	Ready          bool               `json:"ready"`
+	Complete       bool               `json:"complete"`
+	Mode           string             `json:"mode"`
+	ScannedEntries int64              `json:"scannedEntries"`
+	LegacyUploads  int                `json:"legacyUploads"`
+	Sets           []multipartScanSet `json:"sets"`
+}
+
+func startMultipartScan(ctx context.Context, preflight bool) (*multipartScan, error) {
+	select {
+	case multipartScanSlots <- struct{}{}:
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	default:
+		return nil, SlowDown{}
+	}
+	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	s := &multipartScan{ctx: ctx, cancel: cancel, preflight: preflight, metadataSlots: make(chan struct{}, multipartMetadataScanConcurrency)}
+	s.remaining.Store(multipartScanEntryLimit)
+	return s, nil
+}
+
+func (s *multipartScan) close() {
+	s.cancel()
+	<-multipartScanSlots
+}
+
+func (s *multipartScan) listDir(disk StorageAPI, bucket, dir string) ([]string, error) {
+	if err := s.ctx.Err(); err != nil {
+		return nil, SlowDown{}
+	}
+	remaining := s.remaining.Load()
+	if remaining <= 0 {
+		return nil, SlowDown{}
+	}
+	// Request one extra entry to detect overflow, never a silently partial page.
+	entries, err := disk.ListDir(s.ctx, bucket, minioMetaMultipartBucket, dir, int(remaining)+1)
+	if errors.Is(err, errFileNotFound) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	if s.remaining.Add(-int64(len(entries))) < 0 {
+		return nil, SlowDown{}
+	}
+	return entries, nil
+}
+
+func (s *multipartScan) listUploadDirs(disk StorageAPI, bucket string) ([]string, error) {
+	hashDirs, err := s.listDir(disk, bucket, "")
+	if err != nil {
+		return nil, err
+	}
+	var candidates []string
+	for _, hashDir := range hashDirs {
+		if !strings.HasSuffix(hashDir, SlashSeparator) {
+			continue
+		}
+		hashDir = strings.TrimSuffix(hashDir, SlashSeparator)
+		uploadDirs, err := s.listDir(disk, bucket, hashDir)
+		if err != nil {
+			return nil, err
+		}
+		for _, uploadDir := range uploadDirs {
+			if strings.HasSuffix(uploadDir, SlashSeparator) {
+				candidates = append(candidates, pathJoin(hashDir, strings.TrimSuffix(uploadDir, SlashSeparator)))
+			}
+		}
+	}
+	return candidates, nil
+}
+
+// Identity is immutable at hash/uploadUUID. Check the hash before using even
+// a single source drive to exclude another bucket. Missing/bad identities must
+// fall back to the full metadata read; they cannot prove absence.
+func (er erasureObjects) multipartIdentity(fi FileInfo, shaDir string) (string, string, bool) {
+	bucket, object := fi.Metadata[multipartMetaBucket], fi.Metadata[multipartMetaObject]
+	return bucket, object, bucket != "" && object != "" && IsValidBucketName(bucket) &&
+		IsValidObjectPrefix(object) && er.getMultipartSHADir(bucket, object) == shaDir
+}
+
+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{}
+	}
+	shaDir, uploadUUID, ok := strings.Cut(candidate, SlashSeparator)
+	if !ok || shaDir == "" || uploadUUID == "" || strings.Contains(uploadUUID, SlashSeparator) {
+		return MultipartInfo{}, false, false, errMultipartListingIdentity
+	}
+	if !s.preflight && bucket != "" && source != nil {
+		fi, err := source.ReadVersion(s.ctx, bucket, minioMetaMultipartBucket, candidate, "", ReadOptions{})
+		if err == nil {
+			storedBucket, _, valid := er.multipartIdentity(fi, shaDir)
+			if valid && storedBucket != bucket {
+				return MultipartInfo{}, false, false, nil
+			}
+		}
+	}
+	if s.ctx.Err() != nil {
+		return MultipartInfo{}, false, false, SlowDown{}
+	}
+	select {
+	case s.metadataSlots <- struct{}{}:
+		defer func() { <-s.metadataSlots }()
+	case <-s.ctx.Done():
+		return MultipartInfo{}, false, false, SlowDown{}
+	}
+	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.
+		var oldest MultipartInfo
+		legacy := false
+		_, nativeID := multipartUploadTime(uploadUUID)
+		for i, err := range errs {
+			if errors.Is(err, errFileNotFound) || errors.Is(err, errFileVersionNotFound) {
+				continue
+			}
+			if err != nil {
+				return MultipartInfo{}, false, false, err
+			}
+			fi := metadata[i]
+			storedBucket, storedObject, valid := er.multipartIdentity(fi, shaDir)
+			if storedBucket != "" && storedObject != "" && !valid {
+				return MultipartInfo{}, false, false, errMultipartListingIdentity
+			}
+			if !valid || !nativeID {
+				info := multipartUploadInfo(storedBucket, storedObject, uploadUUID, fi.ModTime)
+				if !legacy || info.Initiated.Before(oldest.Initiated) {
+					oldest = info
+				}
+				legacy = true
+			}
+		}
+		if legacy {
+			return oldest, false, true, nil
+		}
+	}
+	readQuorum, _, err := objectQuorumFromMeta(s.ctx, metadata, errs, er.defaultParityCount)
+	if err != nil {
+		return MultipartInfo{}, false, false, err
+	}
+	_, modTime, etag := listOnlineDisks(disks, metadata, errs, readQuorum)
+	if err := reduceReadQuorumErrs(s.ctx, errs, objectOpIgnoredErrs, readQuorum); err != nil {
+		return MultipartInfo{}, false, false, err
+	}
+	fi, err := pickValidFileInfo(s.ctx, metadata, modTime, etag, readQuorum)
+	if err != nil {
+		return MultipartInfo{}, false, false, err
+	}
+	storedBucket, storedObject, valid := er.multipartIdentity(fi, shaDir)
+	info := multipartUploadInfo(storedBucket, storedObject, uploadUUID, fi.ModTime)
+	if storedBucket == "" || storedObject == "" {
+		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
+	}
+	if bucket != "" && bucket != storedBucket {
+		return MultipartInfo{}, false, false, nil
+	}
+	return info, true, false, nil
+}
+
+func (er erasureObjects) scanMultipartUploads(s *multipartScan, bucket string, poolIdx, setIdx int) ([]MultipartInfo, bool, error) {
+	disks := er.getDisks()
+	report := multipartScanSet{Pool: poolIdx, Set: setIdx, Drives: er.setDriveCount}
+	var resultErr error
+	defer func() {
+		if resultErr != nil {
+			report.Error = resultErr.Error()
+		}
+		s.sets = append(s.sets, report)
+	}()
+	var candidateMu sync.Mutex
+	candidates := make(map[string]StorageAPI)
+	errs := make([]error, len(disks))
+	var wg sync.WaitGroup
+	for i, disk := range disks {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if disk == nil || !disk.IsOnline() {
+				errs[i] = errDiskNotFound
+				return
+			}
+			paths, err := s.listUploadDirs(disk, bucket)
+			errs[i] = err
+			if err != nil {
+				return
+			}
+			candidateMu.Lock()
+			defer candidateMu.Unlock()
+			for _, p := range paths {
+				candidates[p] = disk
+			}
+		}()
+	}
+	wg.Wait()
+	for i, err := range errs {
+		if err == nil {
+			report.ScannedDrives++
+		} else {
+			report.UncoveredDrives = append(report.UncoveredDrives, i)
+			if _, limited := err.(SlowDown); limited {
+				resultErr = err
+			}
+		}
+	}
+	report.Candidates = len(candidates)
+	// A successful scan must intersect every metadata read quorum, including
+	// records left with R copies by a partially failed cancellation.
+	if report.ScannedDrives < er.setDriveCount/2+1 && resultErr == nil {
+		resultErr = toObjectErr(errErasureReadQuorum, bucket)
+	}
+	if resultErr != nil {
+		return nil, false, resultErr
+	}
+	type candidate struct {
+		path   string
+		source StorageAPI
+	}
+	jobs := make(chan candidate)
+	var mu sync.Mutex
+	var uploads []MultipartInfo
+	for range min(16, len(candidates)) {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for c := range jobs {
+				mu.Lock()
+				stopped := resultErr != nil
+				mu.Unlock()
+				if stopped || s.ctx.Err() != nil {
+					continue
+				}
+				upload, found, legacy, err := er.readMultipartUploadCandidate(s, bucket, c.path, c.source)
+				if errors.Is(err, errFileNotFound) || errors.Is(err, errFileVersionNotFound) {
+					continue
+				}
+				mu.Lock()
+				switch {
+				case err != nil && resultErr == nil:
+					resultErr = err
+				case legacy:
+					report.LegacyUploads++
+					if report.OldestLegacy.IsZero() || upload.Initiated.Before(report.OldestLegacy) {
+						report.OldestLegacy = upload.Initiated
+					}
+				case found && !s.preflight:
+					uploads = append(uploads, upload)
+				}
+				mu.Unlock()
+			}
+		}()
+	}
+dispatch:
+	for p, source := range candidates {
+		select {
+		case jobs <- candidate{p, source}:
+		case <-s.ctx.Done():
+			break dispatch
+		}
+	}
+	close(jobs)
+	wg.Wait()
+	if s.ctx.Err() != nil {
+		resultErr = SlowDown{}
+	}
+	return uploads, report.LegacyUploads != 0, 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.
+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"
+	}
+	for p, pool := range z.serverPools {
+		for i, set := range pool.sets {
+			_, _, _ = set.scanMultipartUploads(s, "", p, i)
+		}
+	}
+	report.Sets = s.sets
+	for _, set := range s.sets {
+		report.LegacyUploads += set.LegacyUploads
+		if set.Error != "" || set.ScannedDrives != set.Drives {
+			report.Complete = false
+		}
+	}
+	report.ScannedEntries = multipartScanEntryLimit - s.remaining.Load()
+	report.Ready = report.Complete && report.LegacyUploads == 0
+	return report, nil
+}
+
+// MultipartPreflightHandler is a read-only storage-admin diagnostic. It never
+// accepts an arbitrary deletion path or changes upload lifetime settings.
+func (a adminAPIHandlers) MultipartPreflightHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	obj, _ := validateAdminReq(ctx, w, r, policy.StorageInfoAdminAction)
+	if obj == nil {
+		return
+	}
+	z, ok := obj.(*erasureServerPools)
+	if !ok {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
+		return
+	}
+	report, err := z.multipartPreflight(ctx)
+	if err != nil {
+		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
+		return
+	}
+	b, err := json.Marshal(report)
+	if err != nil {
+		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
+		return
+	}
+	writeSuccessResponseJSON(w, b)
+}

cmd/erasure-multipart-listing_test.go

@@ -0,0 +1,796 @@
+// Copyright (c) 2026 Ruohang Feng
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"encoding/xml"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"slices"
+	"strconv"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/minio/minio/internal/config/storageclass"
+)
+
+// Check the real handler and storage path, including a successful abort between
+// pages. Choose IDs increasing in both initiation time and lexical order so the
+// failure does not depend on differing interpretations of S3 marker ordering.
+func TestMultipartListingAbortBetweenHTTPPages(t *testing.T) {
+	z, _ := consistencyPools(t)
+	bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads", "AbortMultipart"}, MakeBucketOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var firstID, secondID string
+	for attempt := 0; attempt < 32; attempt++ {
+		one, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		two, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if one.UploadID < two.UploadID {
+			firstID, secondID = one.UploadID, two.UploadID
+			break
+		}
+		for _, id := range []string{one.UploadID, two.UploadID} {
+			if err := z.AbortMultipartUpload(t.Context(), bucket, "a", id, ObjectOptions{}); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+	if firstID == "" {
+		t.Fatal("could not construct increasing upload IDs")
+	}
+	if _, err := z.NewMultipartUpload(t.Context(), bucket, "b", ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	request := func(method, u string) *httptest.ResponseRecorder {
+		t.Helper()
+		req, err := newTestSignedRequestV4(method, u, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		rec := httptest.NewRecorder()
+		router.ServeHTTP(rec, req)
+		return rec
+	}
+	list := func(keyMarker, uploadMarker string, limit int) ListMultipartUploadsResponse {
+		t.Helper()
+		rec := request(http.MethodGet, getListMultipartUploadsURLWithParams("", bucket, "", keyMarker, uploadMarker, "", strconv.Itoa(limit)))
+		if rec.Code != http.StatusOK {
+			t.Fatalf("list HTTP %d: %s", rec.Code, rec.Body.String())
+		}
+		var result ListMultipartUploadsResponse
+		if err := xml.Unmarshal(rec.Body.Bytes(), &result); err != nil {
+			t.Fatal(err)
+		}
+		return result
+	}
+	first := list("", "", 1)
+	if len(first.Uploads) != 1 || first.Uploads[0].UploadID != firstID || !first.IsTruncated {
+		t.Fatalf("unexpected first page: %+v", first)
+	}
+	rec := request(http.MethodDelete, getAbortMultipartUploadURL("", bucket, "a", firstID))
+	if rec.Code != http.StatusNoContent {
+		t.Fatalf("abort HTTP %d: %s", rec.Code, rec.Body.String())
+	}
+	rest := list(first.NextKeyMarker, first.NextUploadIDMarker, 10)
+	var keys []string
+	for _, upload := range rest.Uploads {
+		keys = append(keys, upload.Key)
+	}
+	t.Logf("GET first page=200; DELETE marker=204; GET next page=200 keys=%v truncated=%v", keys, rest.IsTruncated)
+	if _, err := z.GetMultipartInfo(t.Context(), bucket, "a", secondID, ObjectOptions{}); err != nil {
+		t.Fatalf("remaining upload is not valid: %v", err)
+	}
+	if len(rest.Uploads) != 2 || rest.Uploads[0].UploadID != secondID {
+		t.Fatalf("valid remaining upload for key a is missing from continuation: %+v", rest.Uploads)
+	}
+}
+
+type multipartListingFaultDisk struct {
+	StorageAPI
+	read   func(context.Context, string, string, string, string, ReadOptions) (FileInfo, error)
+	delete func(context.Context, string, string, DeleteOptions) error
+}
+
+func (d multipartListingFaultDisk) ReadVersion(ctx context.Context, original, volume, object, version string, opts ReadOptions) (FileInfo, error) {
+	if d.read != nil {
+		return d.read(ctx, original, volume, object, version, opts)
+	}
+	return d.StorageAPI.ReadVersion(ctx, original, volume, object, version, opts)
+}
+
+func (d multipartListingFaultDisk) Delete(ctx context.Context, volume, object string, opts DeleteOptions) error {
+	if d.delete != nil {
+		return d.delete(ctx, volume, object, opts)
+	}
+	return d.StorageAPI.Delete(ctx, volume, object, opts)
+}
+
+func TestMultipartListingLegacyPreflight(t *testing.T) {
+	z, set, bucket := multipartListingFixture(t)
+	const other = "multipart-legacy-other"
+	if err := z.MakeBucket(t.Context(), other, MakeBucketOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	old, err := z.NewMultipartUpload(t.Context(), other, "old", ObjectOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	fi, metadata, err := set.checkUploadIDExists(t.Context(), other, "old", old.UploadID, true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for i := range metadata {
+		delete(metadata[i].Metadata, multipartMetaBucket)
+		delete(metadata[i].Metadata, multipartMetaObject)
+	}
+	if _, err = writeAllMetadata(t.Context(), set.getDisks(), other, minioMetaMultipartBucket,
+		set.getUploadIDDir(other, "old", old.UploadID), metadata, fi.WriteQuorum(set.defaultWQuorum())); err != nil {
+		t.Fatal(err)
+	}
+	if _, err = z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	z.mpCache.Clear()
+	_, err = z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
+	if !errors.Is(err, errMultipartListingLegacy) {
+		t.Fatalf("old upload in another bucket: %v", err)
+	}
+	report, err := z.multipartPreflight(t.Context())
+	if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 {
+		t.Fatalf("legacy preflight: %+v %v", report, err)
+	}
+	if err = z.AbortMultipartUpload(t.Context(), other, "old", old.UploadID, ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	report, err = z.multipartPreflight(t.Context())
+	if err != nil || !report.Ready || report.LegacyUploads != 0 {
+		t.Fatalf("drained preflight: %+v %v", report, err)
+	}
+	got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireMultipartUploadKeys(t, got, "a")
+}
+
+func TestMultipartListingIdentityFallback(t *testing.T) {
+	for _, kind := range []string{"old", "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 {
+				t.Fatal(err)
+			}
+			original := set.getDisks
+			t.Cleanup(func() { set.getDisks = original })
+			var first atomic.Bool
+			set.getDisks = func() []StorageAPI {
+				disks := append([]StorageAPI(nil), original()...)
+				for i, d := range disks {
+					disks[i] = multipartListingFaultDisk{StorageAPI: d, read: func(ctx context.Context, b, v, p, version string, opts ReadOptions) (FileInfo, error) {
+						fi, err := d.ReadVersion(ctx, b, v, p, version, opts)
+						if v == minioMetaMultipartBucket && first.CompareAndSwap(false, true) {
+							if kind == "read-failure" {
+								return FileInfo{}, errDiskNotFound
+							}
+							if kind == "corrupt" {
+								return FileInfo{}, errFileCorrupt
+							}
+							fi.Metadata = cloneMSS(fi.Metadata)
+							if kind == "old" {
+								delete(fi.Metadata, multipartMetaBucket)
+							} else {
+								fi.Metadata[multipartMetaBucket] = "another-bucket"
+							}
+						}
+						return fi, err
+					}}
+				}
+				return disks
+			}
+			got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
+			if err != nil {
+				t.Fatal(err)
+			}
+			requireMultipartUploadKeys(t, got, "a")
+		})
+	}
+}
+
+func TestMultipartAbortPoolsAndRetry(t *testing.T) {
+	z, bucket := consistencyPools(t)
+	mp, err := z.serverPools[1].NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	emptySet := z.serverPools[0].getHashedSet("a")
+	original := emptySet.getDisks
+	t.Cleanup(func() { emptySet.getDisks = original })
+	emptySet.getDisks = func() []StorageAPI {
+		disks := append([]StorageAPI(nil), original()...)
+		for i, d := range disks {
+			disks[i] = multipartListingFaultDisk{StorageAPI: d, read: func(context.Context, string, string, string, string, ReadOptions) (FileInfo, error) {
+				return FileInfo{}, errDiskNotFound
+			}}
+		}
+		return disks
+	}
+	err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{})
+	if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 {
+		t.Fatalf("unknown pool must not acknowledge cancellation: %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)
+	}
+	mp, err = z.serverPools[1].NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err = z.serverPools[1].GetMultipartInfo(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err == nil {
+		t.Fatal("empty first pool hid the actual upload")
+	}
+}
+
+func TestMultipartAbortRetryBelowReadQuorum(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()...)
+		disks[3] = nil
+		d := disks[2]
+		disks[2] = multipartListingFaultDisk{StorageAPI: d, delete: func(ctx context.Context, v, p string, opts DeleteOptions) error {
+			if err := d.Delete(ctx, v, p, opts); err != nil {
+				return err
+			}
+			return context.DeadlineExceeded // operation finished, acknowledgement lost
+		}}
+		return disks
+	}
+	if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err == nil {
+		t.Fatal("lost acknowledgement must fail")
+	}
+	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)
+	}
+}
+
+func TestMultipartListingMarkerHTTP(t *testing.T) {
+	z, _ := consistencyPools(t)
+	bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads"}, MakeBucketOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, tc := range []struct {
+		key, marker string
+		status      int
+	}{
+		{"", "not-base64=", 200},
+		{"a", "not-base64=", 404},
+		{"a", base64.RawURLEncoding.EncodeToString([]byte("not-native")), 400},
+		{"a", multipartListingTestID(time.Unix(100, 0), 1), 200},
+	} {
+		u := getListMultipartUploadsURLWithParams("", bucket, "", tc.key, tc.marker, "", "10")
+		req, err := newTestSignedRequestV4(http.MethodGet, u, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		rec := httptest.NewRecorder()
+		router.ServeHTTP(rec, req)
+		if rec.Code != tc.status {
+			t.Fatalf("key=%q marker=%q: %d %s", tc.key, tc.marker, rec.Code, rec.Body.String())
+		}
+	}
+}
+
+func TestMultipartPreflightAdminHTTP(t *testing.T) {
+	bed, err := prepareAdminErasureTestBed(t.Context())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { bed.done(); bed.objLayer.Shutdown(context.Background()); removeRoots(bed.erasureDirs) })
+	const target = "/minio/admin/v3/multipart-preflight"
+	rec := httptest.NewRecorder()
+	bed.router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, target, nil))
+	if rec.Code != 403 {
+		t.Fatalf("anonymous preflight: %d", rec.Code)
+	}
+	req, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	rec = httptest.NewRecorder()
+	bed.router.ServeHTTP(rec, req)
+	if rec.Code != 200 {
+		t.Fatalf("admin preflight: %d %s", rec.Code, rec.Body.String())
+	}
+	var report multipartPreflightReport
+	if err = json.Unmarshal(rec.Body.Bytes(), &report); err != nil || !report.Ready || !report.Complete {
+		t.Fatalf("preflight: %+v %v", report, err)
+	}
+	for range cap(multipartScanSlots) {
+		scan, err := startMultipartScan(t.Context(), false)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer scan.close()
+	}
+	rec = httptest.NewRecorder()
+	bed.router.ServeHTTP(rec, req)
+	if rec.Code != http.StatusServiceUnavailable {
+		t.Fatalf("busy admin preflight: %d %s", rec.Code, rec.Body.String())
+	}
+}
+
+type multipartLateCreateDisk struct {
+	StorageAPI
+	late chan<- func() error
+}
+
+func (d multipartLateCreateDisk) WriteMetadata(_ context.Context, original, volume, object string, fi FileInfo) error {
+	if volume == minioMetaMultipartBucket {
+		d.late <- func() error { return d.StorageAPI.WriteMetadata(context.Background(), original, volume, object, fi) }
+		return context.DeadlineExceeded
+	}
+	return d.StorageAPI.WriteMetadata(context.Background(), original, volume, object, fi)
+}
+
+// This characterization is deliberately NOT an assertion of terminal abort
+// correctness. It preserves an executable example of the separately scoped
+// late-creation-write limitation. Change the expectation when fencing is added.
+func TestMultipartAbortLateCreateBoundary(t *testing.T) {
+	obj, dirs, err := prepareErasure(t.Context(), 16)
+	if err != nil {
+		t.Fatal(err)
+	}
+	z := obj.(*erasureServerPools)
+	t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) })
+	saved := globalStorageClass
+	globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: 8}})
+	t.Cleanup(func() { globalStorageClass.Update(saved) })
+	const bucket = "multipart-late-create"
+	if err = z.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	set := z.serverPools[0].getHashedSet("a")
+	original := set.getDisks
+	t.Cleanup(func() { set.getDisks = original })
+	late := make(chan func() error, 16)
+	set.getDisks = func() []StorageAPI {
+		disks := append([]StorageAPI(nil), original()...)
+		for i := 9; i < 16; i++ {
+			disks[i] = multipartLateCreateDisk{disks[i], late}
+		}
+		return disks
+	}
+	mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(late) != 7 {
+		t.Fatalf("expected seven timed-out creation writes, got %d", len(late))
+	}
+	set.getDisks = func() []StorageAPI {
+		disks := append([]StorageAPI(nil), original()...)
+		for i := 2; i < 9; i++ {
+			disks[i] = nil
+		}
+		return disks
+	}
+	if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	for range 7 {
+		if err = (<-late)(); err != nil {
+			t.Fatal(err)
+		}
+	}
+	set.getDisks = original
+	_, metadata, err := set.checkUploadIDExists(t.Context(), bucket, "a", mp.UploadID, true)
+	if err != nil {
+		t.Fatalf("late creation boundary changed; review and remove the documented limitation: %v", err)
+	}
+	count := 0
+	for _, fi := range metadata {
+		if fi.IsValid() {
+			count++
+		}
+	}
+	if count != 14 {
+		t.Fatalf("expected fourteen 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")
+}
+
+type multipartListingCountingDisk struct {
+	StorageAPI
+	directoryCalls *atomic.Int64
+	metadataCalls  *atomic.Int64
+}
+
+func (d multipartListingCountingDisk) ListDir(ctx context.Context, original, volume, dir string, count int) ([]string, error) {
+	if volume == minioMetaMultipartBucket {
+		d.directoryCalls.Add(1)
+	}
+	return d.StorageAPI.ListDir(ctx, original, volume, dir, count)
+}
+
+func (d multipartListingCountingDisk) ReadVersion(ctx context.Context, original, volume, object, version string, opts ReadOptions) (FileInfo, error) {
+	if volume == minioMetaMultipartBucket {
+		d.metadataCalls.Add(1)
+	}
+	return d.StorageAPI.ReadVersion(ctx, original, volume, object, version, opts)
+}
+
+// Counts storage API calls; this is not a deployment throughput benchmark.
+func TestMultipartListingScanCosts(t *testing.T) {
+	obj, dirs, err := prepareErasure(t.Context(), 4)
+	if err != nil {
+		t.Fatal(err)
+	}
+	z := obj.(*erasureServerPools)
+	t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) })
+	const bucket, otherBucket = "r9-scan-target", "r9-scan-unrelated"
+	for _, name := range []string{bucket, otherBucket} {
+		if err := z.MakeBucket(t.Context(), name, MakeBucketOptions{}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := z.NewMultipartUpload(t.Context(), bucket, "dir/target", ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	var directoryCalls, metadataCalls atomic.Int64
+	for _, pool := range z.serverPools {
+		for _, set := range pool.sets {
+			original := set.getDisks
+			set.getDisks = func() []StorageAPI {
+				disks := original()
+				wrapped := make([]StorageAPI, len(disks))
+				for i, disk := range disks {
+					if disk != nil {
+						wrapped[i] = multipartListingCountingDisk{disk, &directoryCalls, &metadataCalls}
+					}
+				}
+				return wrapped
+			}
+			t.Cleanup(func() { set.getDisks = original })
+		}
+	}
+	measure := func(label string) (int64, int64) {
+		t.Helper()
+		directoryCalls.Store(0)
+		metadataCalls.Store(0)
+		result, err := z.ListMultipartUploads(t.Context(), bucket, "dir/", "", "", "", 1)
+		if err != nil {
+			t.Fatal(err)
+		}
+		requireMultipartUploadKeys(t, result, "dir/target")
+		d, m := directoryCalls.Load(), metadataCalls.Load()
+		t.Logf("%s: max-uploads=1 returned=%d ListDir=%d ReadVersion=%d", label, len(result.Uploads), d, m)
+		return d, m
+	}
+	_, baseline := measure("no unrelated uploads")
+	for n := 0; n < 32; n++ {
+		if _, err := z.NewMultipartUpload(t.Context(), otherBucket, fmt.Sprintf("other/%03d", n), ObjectOptions{}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	_, loaded := measure("32 uploads in another bucket")
+	_, repeated := measure("same query repeated")
+	if loaded != baseline+32 || repeated != loaded {
+		t.Fatalf("unexpected scan accounting: baseline=%d loaded=%d repeated=%d", baseline, loaded, repeated)
+	}
+}
+
+func multipartListingTestID(created time.Time, n int) string {
+	return base64.RawURLEncoding.EncodeToString([]byte(fmt.Sprintf("00000000-0000-4000-8000-000000000000.00000000-0000-4000-8000-%012xx%d", n, created.UnixNano())))
+}
+
+func multipartListingFixture(t *testing.T) (*erasureServerPools, *erasureObjects, string) {
+	t.Helper()
+	obj, dirs, err := prepareErasure(t.Context(), 4)
+	if err != nil {
+		t.Fatal(err)
+	}
+	z := obj.(*erasureServerPools)
+	t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) })
+	const bucket = "multipart-listing-test"
+	if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	return z, z.serverPools[0].getHashedSet("a"), bucket
+}
+
+func TestMultipartListingMissingMarkers(t *testing.T) {
+	base := time.Unix(100, 0)
+	sameTime := []string{multipartListingTestID(base.Add(time.Second), 1), multipartListingTestID(base.Add(time.Second), 2), multipartListingTestID(base.Add(time.Second), 3)}
+	slices.Sort(sameTime)
+	uploads := []MultipartInfo{
+		{Bucket: "bucket", Object: "a", UploadID: multipartListingTestID(base, 1), Initiated: base},
+		{Bucket: "bucket", Object: "a", UploadID: sameTime[1], Initiated: base.Add(time.Second)},
+		{Bucket: "bucket", Object: "b", UploadID: multipartListingTestID(base, 4), Initiated: base},
+	}
+	for _, tc := range []struct {
+		name    string
+		created time.Time
+		id      int
+		want    []string
+	}{
+		{"before", base.Add(-time.Second), 0, []string{"a", "a", "b"}},
+		{"between", base.Add(time.Second / 2), 2, []string{"a", "b"}},
+		{"same-time-before", base.Add(time.Second), 2, []string{"a", "b"}},
+		{"same-time-after", base.Add(time.Second), 4, []string{"b"}},
+		{"after", base.Add(2 * time.Second), 5, []string{"b"}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			marker := multipartListingTestID(tc.created, tc.id)
+			if tc.name == "same-time-before" {
+				marker = sameTime[0]
+			}
+			if tc.name == "same-time-after" {
+				marker = sameTime[2]
+			}
+			if err := checkListMultipartArgs(t.Context(), "bucket", "", "a", marker, ""); err != nil {
+				t.Fatal(err)
+			}
+			got := paginateMultipartUploads(uploads, "", "a", marker, "", 10)
+			if !slices.Equal(multipartUploadKeys(got.Uploads), tc.want) {
+				t.Fatalf("got %v want %v", multipartUploadKeys(got.Uploads), tc.want)
+			}
+		})
+	}
+}
+
+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{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	original := set.getDisks
+	t.Cleanup(func() { set.getDisks = original })
+	disks := original()
+	// Leave a readable two-copy record, simulating a partially failed abort.
+	for _, d := range disks[:2] {
+		if err = d.Delete(t.Context(), minioMetaMultipartBucket, set.getUploadIDDir(bucket, "a", mp.UploadID), DeleteOptions{Recursive: true}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	set.getDisks = func() []StorageAPI { return []StorageAPI{disks[0], disks[1], nil, nil} }
+	_, err = z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
+	if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 {
+		t.Fatalf("incomplete discovery returned success: %v", err)
+	}
+	set.getDisks = func() []StorageAPI { return []StorageAPI{nil, disks[1], disks[2], disks[3]} }
+	got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireMultipartUploadKeys(t, got, "a")
+	report, err := z.multipartPreflight(t.Context())
+	if err != nil || report.Ready || report.Complete || len(report.Sets[0].UncoveredDrives) != 1 {
+		t.Fatalf("offline upgrade preflight: %+v %v", report, err)
+	}
+}
+
+func TestMultipartListingBudgetAndAdmission(t *testing.T) {
+	_, set, bucket := multipartListingFixture(t)
+	if _, err := set.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	scan, err := startMultipartScan(t.Context(), false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer scan.close()
+	scan.remaining.Store(1)
+	_, _, err = set.scanMultipartUploads(scan, bucket, 0, 0)
+	var limited SlowDown
+	if !errors.As(err, &limited) {
+		t.Fatalf("budget: %v", err)
+	}
+	if apiErr := toAPIError(t.Context(), err); apiErr.HTTPStatusCode != http.StatusServiceUnavailable || apiErr.Code != "SlowDown" {
+		t.Fatalf("budget error mapping: %+v", apiErr)
+	}
+	second, err := startMultipartScan(t.Context(), false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer second.close()
+	if third, err := startMultipartScan(t.Context(), false); err == nil {
+		third.close()
+		t.Fatal("third scan admitted")
+	}
+}
+
+func TestMultipartListingAdmissionHTTP(t *testing.T) {
+	z, _, _ := multipartListingFixture(t)
+	bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads"}, MakeBucketOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for range cap(multipartScanSlots) {
+		scan, err := startMultipartScan(t.Context(), false)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer scan.close()
+	}
+	req, err := newTestSignedRequestV4(http.MethodGet,
+		getListMultipartUploadsURLWithParams("", bucket, "", "", "", "", "1"),
+		0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	rec := httptest.NewRecorder()
+	router.ServeHTTP(rec, req)
+	var response APIErrorResponse
+	if err := xml.Unmarshal(rec.Body.Bytes(), &response); err != nil {
+		t.Fatal(err)
+	}
+	if rec.Code != http.StatusServiceUnavailable || response.Code != "SlowDown" {
+		t.Fatalf("admission returned %d %s", rec.Code, rec.Body.String())
+	}
+}
+
+func TestMultipartListingPreflightMinorityLegacy(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)
+	disks := set.getDisks()
+	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.Complete || report.LegacyUploads != 1 {
+		t.Fatalf("minority legacy copy must prevent readiness: %+v %v", report, err)
+	}
+}
+
+func TestMultipartListingCancellationRetainsAdmission(t *testing.T) {
+	z, set, bucket := multipartListingFixture(t)
+	for i := range 40 {
+		if _, err := z.NewMultipartUpload(t.Context(), bucket, fmt.Sprintf("key-%02d", i), ObjectOptions{}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	original := set.getDisks
+	t.Cleanup(func() { set.getDisks = original })
+	entered := make(chan struct{}, 40)
+	release := make(chan struct{})
+	var once sync.Once
+	defer once.Do(func() { close(release) })
+	var reads atomic.Int32
+	set.getDisks = func() []StorageAPI {
+		disks := append([]StorageAPI(nil), original()...)
+		for i, d := range disks {
+			disks[i] = multipartListingFaultDisk{StorageAPI: d, read: func(ctx context.Context, b, v, p, version string, opts ReadOptions) (FileInfo, error) {
+				if v != minioMetaMultipartBucket {
+					return d.ReadVersion(ctx, b, v, p, version, opts)
+				}
+				reads.Add(1)
+				entered <- struct{}{}
+				// Model an RPC which does not return immediately on cancellation.
+				<-release
+				return FileInfo{}, ctx.Err()
+			}}
+		}
+		return disks
+	}
+	second, err := startMultipartScan(t.Context(), false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer second.close()
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	done := make(chan error, 1)
+	go func() {
+		_, err := z.ListMultipartUploads(ctx, bucket, "", "", "", "", 10)
+		done <- err
+	}()
+	for range 16 {
+		select {
+		case <-entered:
+		case <-time.After(5 * time.Second):
+			t.Fatal("identity workers did not start")
+		}
+	}
+	cancel()
+	if extra, err := startMultipartScan(t.Context(), false); err == nil {
+		extra.close()
+		t.Fatal("canceled request released admission while RPCs were still running")
+	}
+	start := time.Now()
+	once.Do(func() { close(release) })
+	select {
+	case err := <-done:
+		if err == nil {
+			t.Fatal("canceled scan returned a successful partial list")
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("scan did not stop after blocked RPCs returned")
+	}
+	if got := reads.Load(); got != 16 {
+		t.Fatalf("scheduled more identity/metadata reads after cancellation: %d", got)
+	}
+	t.Logf("16 identity RPCs bounded; admission held until return; cancellation settled in %s", time.Since(start))
+}

cmd/erasure-multipart.go

@@ -31,6 +31,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/klauspost/readahead"
 	"github.com/minio/minio-go/v7/pkg/set"
 	"github.com/minio/minio/internal/config/storageclass"
@@ -44,6 +45,15 @@ import (
 	"github.com/pgsty/silo-pkg/v3/sync/errgroup"
 )
 
+const (
+	multipartMetaBucket = ReservedMetadataPrefixLower + "multipart-v1-bucket"
+	multipartMetaObject = ReservedMetadataPrefixLower + "multipart-v1-object"
+
+	// ponytail: keep scan concurrency fixed until the multipart-list benchmark
+	// establishes a better adaptive limit.
+	multipartMetadataScanConcurrency = 4
+)
+
 func (er erasureObjects) getUploadIDDir(bucket, object, uploadID string) string {
 	uploadUUID := uploadID
 	uploadBytes, err := base64.RawURLEncoding.DecodeString(uploadID)
@@ -251,14 +261,152 @@ func (er erasureObjects) cleanupStaleUploadsOnDisk(ctx context.Context, disk Sto
 	})
 }
 
-// ListMultipartUploads - lists all the pending multipart
-// uploads for a particular object in a bucket.
-//
-// Implements minimal S3 compatible ListMultipartUploads API. We do
-// not support prefix based listing, this is a deliberate attempt
-// towards simplification of multipart APIs.
-// The resulting ListMultipartsInfo structure is unmarshalled directly as XML.
-func (er erasureObjects) ListMultipartUploads(ctx context.Context, bucket, object, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (result ListMultipartsInfo, err error) {
+func multipartUploadInfo(bucket, object, uploadUUID string, fallback time.Time) MultipartInfo {
+	initiated := fallback
+	if parsed, ok := multipartUploadTime(uploadUUID); ok {
+		initiated = parsed
+	}
+	return MultipartInfo{
+		Bucket:    bucket,
+		Object:    object,
+		UploadID:  base64.RawURLEncoding.EncodeToString(fmt.Appendf(nil, "%s.%s", globalDeploymentID(), uploadUUID)),
+		Initiated: initiated,
+	}
+}
+
+// multipartUploadTime is shared by stored records and continuation markers.
+// The time is part of the immutable upload ID, so a removed marker still
+// identifies the same ordering boundary.
+func multipartUploadTime(uploadUUID string) (time.Time, bool) {
+	if len(uploadUUID) < 38 || uploadUUID[36] != 'x' {
+		return time.Time{}, false
+	}
+	if _, err := uuid.Parse(uploadUUID[:36]); err != nil {
+		return time.Time{}, false
+	}
+	ns, err := strconv.ParseInt(uploadUUID[37:], 10, 64)
+	if err != nil || ns <= 0 || strconv.FormatInt(ns, 10) != uploadUUID[37:] {
+		return time.Time{}, false
+	}
+	return time.Unix(0, ns), true
+}
+
+func multipartMarkerTime(uploadID string) (time.Time, bool) {
+	b, err := base64.RawURLEncoding.DecodeString(uploadID)
+	if err != nil {
+		return time.Time{}, false
+	}
+	_, uploadUUID, ok := strings.Cut(string(b), ".")
+	if !ok {
+		return time.Time{}, false
+	}
+	return multipartUploadTime(uploadUUID)
+}
+
+type multipartListEntry struct {
+	upload       *MultipartInfo
+	commonPrefix string
+}
+
+// paginateMultipartUploads applies the S3 ordering, prefix, delimiter, marker,
+// and page rules exactly once after all pools and sets have been merged.
+func paginateMultipartUploads(uploads []MultipartInfo, prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int) ListMultipartsInfo {
+	if maxUploads > maxUploadsList {
+		maxUploads = maxUploadsList
+	}
+	result := ListMultipartsInfo{
+		MaxUploads:     maxUploads,
+		KeyMarker:      keyMarker,
+		UploadIDMarker: uploadIDMarker,
+		Prefix:         prefix,
+		Delimiter:      delimiter,
+	}
+
+	deduplicated := make([]MultipartInfo, 0, len(uploads))
+	seenUploads := make(map[string]struct{}, len(uploads))
+	for _, upload := range uploads {
+		identity := upload.Bucket + "\x00" + upload.Object + "\x00" + upload.UploadID
+		if _, ok := seenUploads[identity]; ok {
+			continue
+		}
+		seenUploads[identity] = struct{}{}
+		deduplicated = append(deduplicated, upload)
+	}
+	sort.Slice(deduplicated, func(i, j int) bool {
+		if deduplicated[i].Object != deduplicated[j].Object {
+			return deduplicated[i].Object < deduplicated[j].Object
+		}
+		if !deduplicated[i].Initiated.Equal(deduplicated[j].Initiated) {
+			return deduplicated[i].Initiated.Before(deduplicated[j].Initiated)
+		}
+		return deduplicated[i].UploadID < deduplicated[j].UploadID
+	})
+
+	markerTime, _ := multipartMarkerTime(uploadIDMarker)
+	seenPrefixes := make(map[string]struct{})
+	entries := make([]multipartListEntry, 0, len(deduplicated))
+	for i := range deduplicated {
+		upload := &deduplicated[i]
+		if !strings.HasPrefix(upload.Object, prefix) {
+			continue
+		}
+		if keyMarker != "" {
+			switch strings.Compare(upload.Object, keyMarker) {
+			case -1:
+				continue
+			case 0:
+				if uploadIDMarker == "" || upload.Initiated.Before(markerTime) ||
+					(upload.Initiated.Equal(markerTime) && upload.UploadID <= uploadIDMarker) {
+					continue
+				}
+			}
+		}
+
+		if delimiter != "" {
+			remainder := strings.TrimPrefix(upload.Object, prefix)
+			if i := strings.Index(remainder, delimiter); i >= 0 {
+				commonPrefix := prefix + remainder[:i+len(delimiter)]
+				if keyMarker != "" && commonPrefix <= keyMarker {
+					continue
+				}
+				if _, ok := seenPrefixes[commonPrefix]; ok {
+					continue
+				}
+				seenPrefixes[commonPrefix] = struct{}{}
+				entries = append(entries, multipartListEntry{commonPrefix: commonPrefix})
+				continue
+			}
+		}
+		entries = append(entries, multipartListEntry{upload: upload})
+	}
+
+	if maxUploads <= 0 {
+		return result
+	}
+	pageSize := min(maxUploads, len(entries))
+	for _, entry := range entries[:pageSize] {
+		if entry.upload != nil {
+			result.Uploads = append(result.Uploads, *entry.upload)
+			continue
+		}
+		result.CommonPrefixes = append(result.CommonPrefixes, entry.commonPrefix)
+	}
+	result.IsTruncated = pageSize < len(entries)
+	if result.IsTruncated && pageSize > 0 {
+		last := entries[pageSize-1]
+		if last.upload != nil {
+			result.NextKeyMarker = last.upload.Object
+			result.NextUploadIDMarker = last.upload.UploadID
+		} else {
+			result.NextKeyMarker = last.commonPrefix
+		}
+	}
+	return result
+}
+
+// listMultipartUploadsExact preserves the hashed exact-object lookup used by
+// multipart write placement and by rolling-upgrade legacy mode.
+func (er erasureObjects) listMultipartUploadsExact(ctx context.Context, bucket, object, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (result ListMultipartsInfo, err error) {
 	auditObjectErasureSet(ctx, "ListMultipartUploads", object, &er)
 
 	result.MaxUploads = maxUploads
@@ -311,20 +459,16 @@ func (er erasureObjects) ListMultipartUploads(ctx context.Context, bucket, objec
 		if populatedUploadIDs.Contains(uploadID) {
 			continue
 		}
-		// If present, use time stored in ID.
-		startTime := time.Now()
-		if split := strings.Split(uploadID, "x"); len(split) == 2 {
-			t, err := strconv.ParseInt(split[1], 10, 64)
-			if err == nil {
-				startTime = time.Unix(0, t)
+		var fallback time.Time
+		if _, ok := multipartUploadTime(uploadID); !ok {
+			fi, err := disk.ReadVersion(ctx, bucket, minioMetaMultipartBucket,
+				pathJoin(er.getMultipartSHADir(bucket, object), uploadID), "", ReadOptions{})
+			if err != nil {
+				return result, toObjectErr(err, bucket, object)
 			}
+			fallback = fi.ModTime
 		}
-		uploads = append(uploads, MultipartInfo{
-			Bucket:    bucket,
-			Object:    object,
-			UploadID:  base64.RawURLEncoding.EncodeToString(fmt.Appendf(nil, "%s.%s", globalDeploymentID(), uploadID)),
-			Initiated: startTime,
-		})
+		uploads = append(uploads, multipartUploadInfo(bucket, object, uploadID, fallback))
 		populatedUploadIDs.Add(uploadID)
 	}
 
@@ -365,6 +509,25 @@ func (er erasureObjects) ListMultipartUploads(ctx context.Context, bucket, objec
 	return result, nil
 }
 
+func (er erasureObjects) ListMultipartUploads(ctx context.Context, bucket, prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (ListMultipartsInfo, error) {
+	if err := checkListMultipartArgs(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter); err != nil {
+		return ListMultipartsInfo{}, err
+	}
+	scan, err := startMultipartScan(ctx, false)
+	if err != nil {
+		return ListMultipartsInfo{}, err
+	}
+	defer scan.close()
+	uploads, legacy, 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
+}
+
 // newMultipartUpload - wrapper for initializing a new multipart
 // request; returns a unique upload id.
 //
@@ -402,6 +565,8 @@ func (er erasureObjects) newMultipartUpload(ctx context.Context, bucket string,
 	}
 
 	userDefined := cloneMSS(opts.UserDefined)
+	userDefined[multipartMetaBucket] = bucket
+	userDefined[multipartMetaObject] = object
 	if opts.PreserveETag != "" {
 		userDefined["etag"] = opts.PreserveETag
 	}
@@ -1459,6 +1624,8 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 	// Remove superfluous internal headers.
 	delete(fi.Metadata, hash.MinIOMultipartChecksum)
 	delete(fi.Metadata, hash.MinIOMultipartChecksumType)
+	delete(fi.Metadata, multipartMetaBucket)
+	delete(fi.Metadata, multipartMetaObject)
 
 	// Save the final object size and modtime.
 	fi.Size = objectSize
@@ -1586,22 +1753,66 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 	return fi.ToObjectInfo(bucket, object, opts.Versioned || opts.VersionSuspended), nil
 }
 
-// AbortMultipartUpload - aborts an ongoing multipart operation
-// signified by the input uploadID. This is an atomic operation
-// doesn't require clients to initiate multiple such requests.
-//
-// All parts are purged from all disks and reference to the uploadID
-// would be removed from the system, rollback is not possible on this
-// operation.
-func (er erasureObjects) AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) (err error) {
+// 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) {
 	if !opts.NoAuditLog {
 		auditObjectErasureSet(ctx, "AbortMultipartUpload", object, &er)
 	}
-
-	// Cleanup all uploaded parts.
-	defer er.deleteAll(ctx, minioMetaMultipartBucket, er.getUploadIDDir(bucket, object, uploadID))
-
-	// Validates if upload ID exists.
-	_, _, err = er.checkUploadIDExists(ctx, bucket, object, uploadID, false)
-	return toObjectErr(err, bucket, object, uploadID)
+	b, err := base64.RawURLEncoding.DecodeString(uploadID)
+	if err != nil {
+		return false, MalformedUploadID{UploadID: uploadID}
+	}
+	_, internalID, ok := strings.Cut(string(b), ".")
+	if !ok || internalID == "" || internalID == "." || internalID == ".." || strings.ContainsAny(internalID, "/\\") {
+		return false, InvalidUploadID{Bucket: bucket, Object: object, UploadID: uploadID}
+	}
+	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 {
+		switch {
+		case err == nil, errors.Is(err, errFileCorrupt):
+			found = true
+		case errors.Is(err, errFileNotFound), errors.Is(err, errFileVersionNotFound):
+			absent++
+		}
+	}
+	if absent >= quorum {
+		return found, 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)
+	}
+	return true, toObjectErr(reduceWriteQuorumErrs(ctx, g.Wait(), nil, quorum), bucket, object, uploadID)
+}
+
+// AbortMultipartUpload confirms logical cancellation. 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)
+	if err != nil {
+		return err
+	}
+	if !found {
+		return InvalidUploadID{Bucket: bucket, Object: object, UploadID: uploadID}
+	}
+	return nil
 }

cmd/erasure-server-pool.go

@@ -1891,7 +1891,41 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p
 	if err := checkListMultipartArgs(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter); err != nil {
 		return ListMultipartsInfo{}, err
 	}
+	if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil {
+		return ListMultipartsInfo{}, toObjectErr(err, bucket)
+	}
+	if globalAPIConfig.getMultipartListingLegacy() {
+		return z.listMultipartUploadsLegacy(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
+	}
+	scan, err := startMultipartScan(ctx, false)
+	if err != nil {
+		return ListMultipartsInfo{}, err
+	}
+	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)
+		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
+	}
+	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
@@ -1917,15 +1951,14 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p
 	}
 
 	if z.SinglePool() {
-		return z.serverPools[0].ListMultipartUploads(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
+		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.ListMultipartUploads(ctx, bucket, prefix, keyMarker, uploadIDMarker,
-			delimiter, maxUploads)
+		result, err := pool.getHashedSet(prefix).listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
 		if err != nil {
 			return result, err
 		}
@@ -1961,7 +1994,7 @@ func (z *erasureServerPools) NewMultipartUpload(ctx context.Context, bucket, obj
 			continue
 		}
 
-		result, err := pool.ListMultipartUploads(ctx, bucket, object, "", "", "", maxUploadsList)
+		result, err := pool.listMultipartUploadsExact(ctx, bucket, object)
 		if err != nil {
 			return nil, err
 		}
@@ -2123,9 +2156,13 @@ func (z *erasureServerPools) AbortMultipartUpload(ctx context.Context, bucket, o
 	if err := checkAbortMultipartArgs(ctx, bucket, object, uploadID); err != nil {
 		return err
 	}
+	if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil {
+		return toObjectErr(err, bucket)
+	}
 
 	defer func() {
-		if err == nil {
+		_, absent := err.(InvalidUploadID)
+		if err == nil || absent {
 			z.mpCache.Delete(uploadID)
 			globalNotificationSys.DeleteUploadID(ctx, uploadID)
 		}
@@ -2139,23 +2176,23 @@ func (z *erasureServerPools) AbortMultipartUpload(ctx context.Context, bucket, o
 	ctx = lkctx.Context()
 	defer lk.Unlock(lkctx)
 
-	if z.SinglePool() {
-		return z.serverPools[0].AbortMultipartUpload(ctx, bucket, object, uploadID, opts)
-	}
-
+	found := false
+	var firstErr error
 	for idx, pool := range z.serverPools {
 		if z.IsSuspended(idx) {
 			continue
 		}
-		err := pool.AbortMultipartUpload(ctx, bucket, object, uploadID, opts)
-		if err == nil {
-			return nil
+		poolFound, err := pool.getHashedSet(object).abortMultipartUpload(ctx, bucket, object, uploadID, opts)
+		found = found || poolFound
+		if err != nil && firstErr == nil {
+			firstErr = err
 		}
-		if _, ok := err.(InvalidUploadID); ok {
-			// upload id not found move to next pool
-			continue
-		}
-		return err
+	}
+	if firstErr != nil {
+		return firstErr
+	}
+	if found {
+		return nil
 	}
 	return InvalidUploadID{
 		Bucket:   bucket,

cmd/erasure-sets.go

@@ -880,10 +880,40 @@ func (s *erasureSets) CopyObject(ctx context.Context, srcBucket, srcObject, dstB
 }
 
 func (s *erasureSets) ListMultipartUploads(ctx context.Context, bucket, prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (result ListMultipartsInfo, err error) {
-	// In list multipart uploads we are going to treat input prefix as the object,
-	// this means that we are not supporting directory navigation.
-	set := s.getHashedSet(prefix)
-	return set.ListMultipartUploads(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
+	if err := checkListMultipartArgs(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter); err != nil {
+		return ListMultipartsInfo{}, err
+	}
+	scan, err := startMultipartScan(ctx, false)
+	if err != nil {
+		return ListMultipartsInfo{}, err
+	}
+	defer scan.close()
+	uploads, legacy, 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) {
+	var uploads []MultipartInfo
+	var keyless bool
+	for i, set := range s.sets {
+		setUploads, setKeyless, err := set.scanMultipartUploads(scan, bucket, poolIdx, i)
+		if err != nil {
+			return nil, false, err
+		}
+		uploads = append(uploads, setUploads...)
+		keyless = keyless || setKeyless
+	}
+	return uploads, keyless, nil
+}
+
+func (s *erasureSets) listMultipartUploadsExact(ctx context.Context, bucket, object string) (ListMultipartsInfo, error) {
+	return s.getHashedSet(object).listMultipartUploadsExact(ctx, bucket, object, "", "", "", maxUploadsList)
 }
 
 // Initiate a new multipart upload on a hashedSet based on object name.

cmd/handler-api.go

@@ -50,6 +50,7 @@ type apiConfig struct {
 	transitionWorkers      int
 
 	staleUploadsExpiry          time.Duration
+	multipartListingLegacy      bool
 	staleUploadsCleanupInterval time.Duration
 	deleteCleanupInterval       time.Duration
 	enableODirect               bool
@@ -181,6 +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.deleteCleanupInterval = cfg.DeleteCleanupInterval
 	t.enableODirect = cfg.EnableODirect
 	t.gzipObjects = cfg.GzipObjects
@@ -206,6 +208,12 @@ func (t *apiConfig) odirectEnabled() bool {
 	return t.enableODirect
 }
 
+func (t *apiConfig) getMultipartListingLegacy() bool {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+	return t.multipartListingLegacy
+}
+
 func (t *apiConfig) shouldGzipObjects() bool {
 	t.mu.RLock()
 	defer t.mu.RUnlock()

cmd/list-multipart-uploads-compat_test.go

@@ -0,0 +1,298 @@
+// Copyright (c) 2026 mr javad seydi
+//
+// This file is part of Silo Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"slices"
+	"testing"
+	"time"
+)
+
+func multipartUploadKeys(uploads []MultipartInfo) []string {
+	keys := make([]string, len(uploads))
+	for i := range uploads {
+		keys[i] = uploads[i].Object
+	}
+	return keys
+}
+
+func requireMultipartUploadKeys(t *testing.T, got ListMultipartsInfo, want ...string) {
+	t.Helper()
+	if keys := multipartUploadKeys(got.Uploads); !slices.Equal(keys, want) {
+		t.Fatalf("uploads = %v, want %v", keys, want)
+	}
+}
+
+func TestListMultipartUploadsS3Compatibility(t *testing.T) {
+	obj, dirs, err := prepareErasureSets32(t.Context())
+	if err != nil {
+		t.Fatal(err)
+	}
+	z := obj.(*erasureServerPools)
+	t.Cleanup(func() {
+		z.Shutdown(t.Context())
+		removeRoots(dirs)
+	})
+
+	const bucket = "multipart-list-compat"
+	if err = z.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil {
+		t.Fatal(err)
+	}
+
+	objects := []string{"t/a_b/p1", "t/a_b/p2", "t/c_d/p1", "u/x"}
+	sets := z.serverPools[0]
+	firstSet := sets.getHashedSetIndex(objects[0])
+	if !slices.ContainsFunc(objects[1:], func(object string) bool {
+		return sets.getHashedSetIndex(object) != firstSet
+	}) {
+		for n := 0; ; n++ {
+			object := fmt.Sprintf("v/cross-set-%d", n)
+			if sets.getHashedSetIndex(object) != firstSet {
+				objects = append(objects, object)
+				break
+			}
+		}
+	}
+
+	uploadIDs := make(map[string]string, len(objects))
+	for _, object := range objects {
+		mp, err := z.NewMultipartUpload(t.Context(), bucket, object, ObjectOptions{})
+		if err != nil {
+			t.Fatalf("NewMultipartUpload(%q): %v", object, err)
+		}
+		uploadIDs[object] = mp.UploadID
+	}
+
+	// Durable multipart metadata, rather than this node-local cache, must be
+	// authoritative after a restart or when another node handles the request.
+	z.mpCache.Range(func(uploadID string, _ MultipartInfo) bool {
+		z.mpCache.Delete(uploadID)
+		return true
+	})
+
+	all, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 100)
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireMultipartUploadKeys(t, all, objects...)
+	if all.IsTruncated || all.NextKeyMarker != "" || all.NextUploadIDMarker != "" {
+		t.Fatalf("complete listing has truncation state: %+v", all)
+	}
+
+	first, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireMultipartUploadKeys(t, first, objects[0])
+	if !first.IsTruncated || first.NextKeyMarker != objects[0] || first.NextUploadIDMarker != uploadIDs[objects[0]] {
+		t.Fatalf("first page markers = (%q, %q, %t), want (%q, %q, true)",
+			first.NextKeyMarker, first.NextUploadIDMarker, first.IsTruncated,
+			objects[0], uploadIDs[objects[0]])
+	}
+
+	rest, err := z.ListMultipartUploads(t.Context(), bucket, "", first.NextKeyMarker, first.NextUploadIDMarker, "", 100)
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireMultipartUploadKeys(t, rest, objects[1:]...)
+
+	afterKey, err := z.ListMultipartUploads(t.Context(), bucket, "", objects[1], "", "", 100)
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireMultipartUploadKeys(t, afterKey, objects[2:]...)
+
+	prefixed, err := z.ListMultipartUploads(t.Context(), bucket, "t/", "", "", "", 100)
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireMultipartUploadKeys(t, prefixed, objects[:3]...)
+
+	nested, err := z.ListMultipartUploads(t.Context(), bucket, "t/a_b/", "", "", "", 100)
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireMultipartUploadKeys(t, nested, objects[:2]...)
+
+	grouped, err := z.ListMultipartUploads(t.Context(), bucket, "t/", "", "", SlashSeparator, 100)
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireMultipartUploadKeys(t, grouped)
+	if want := []string{"t/a_b/", "t/c_d/"}; !slices.Equal(grouped.CommonPrefixes, want) {
+		t.Fatalf("common prefixes = %v, want %v", grouped.CommonPrefixes, want)
+	}
+
+	groupPage, err := z.ListMultipartUploads(t.Context(), bucket, "t/", "", "", SlashSeparator, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if want := []string{"t/a_b/"}; !slices.Equal(groupPage.CommonPrefixes, want) {
+		t.Fatalf("first common-prefix page = %v, want %v", groupPage.CommonPrefixes, want)
+	}
+	if !groupPage.IsTruncated || groupPage.NextKeyMarker != "t/a_b/" || groupPage.NextUploadIDMarker != "" {
+		t.Fatalf("common-prefix page markers = (%q, %q, %t)",
+			groupPage.NextKeyMarker, groupPage.NextUploadIDMarker, groupPage.IsTruncated)
+	}
+
+	groupRest, err := z.ListMultipartUploads(t.Context(), bucket, "t/", groupPage.NextKeyMarker, "", SlashSeparator, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if want := []string{"t/c_d/"}; !slices.Equal(groupRest.CommonPrefixes, want) {
+		t.Fatalf("second common-prefix page = %v, want %v", groupRest.CommonPrefixes, want)
+	}
+	if groupRest.IsTruncated {
+		t.Fatalf("last common-prefix page is truncated: %+v", groupRest)
+	}
+
+	part, err := z.PutObjectPart(t.Context(), bucket, objects[0], uploadIDs[objects[0]], 1,
+		mustGetPutObjReader(t, bytes.NewBufferString("part"), 4, "", ""), ObjectOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	completed, err := z.CompleteMultipartUpload(t.Context(), bucket, objects[0], uploadIDs[objects[0]],
+		[]CompletePart{{PartNumber: 1, ETag: part.ETag}}, ObjectOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, key := range []string{multipartMetaBucket, multipartMetaObject} {
+		if _, ok := completed.UserDefined[key]; ok {
+			t.Errorf("completed object retained upload-only metadata %q", key)
+		}
+	}
+
+	// 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.
+	legacyObject := objects[1]
+	er := sets.getHashedSet(legacyObject)
+	fi, metadata, err := er.checkUploadIDExists(t.Context(), bucket, legacyObject, uploadIDs[legacyObject], true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for i := range metadata {
+		delete(metadata[i].Metadata, multipartMetaBucket)
+		delete(metadata[i].Metadata, multipartMetaObject)
+	}
+	if _, err = writeAllMetadata(t.Context(), er.getDisks(), bucket, minioMetaMultipartBucket,
+		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)
+	}
+	globalAPIConfig.mu.Lock()
+	oldLegacy := globalAPIConfig.multipartListingLegacy
+	globalAPIConfig.multipartListingLegacy = true
+	globalAPIConfig.mu.Unlock()
+	t.Cleanup(func() {
+		globalAPIConfig.mu.Lock()
+		globalAPIConfig.multipartListingLegacy = oldLegacy
+		globalAPIConfig.mu.Unlock()
+	})
+	legacy, err := z.ListMultipartUploads(t.Context(), bucket, legacyObject, "", "", "", 100)
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireMultipartUploadKeys(t, legacy, legacyObject)
+}
+
+func TestPaginateMultipartUploads(t *testing.T) {
+	base := time.Unix(100, 0)
+	id1, id2, id3 := multipartListingTestID(base, 1), multipartListingTestID(base.Add(time.Second), 2), multipartListingTestID(base, 3)
+	uploads := []MultipartInfo{
+		{Bucket: "bucket", Object: "b", UploadID: id3, Initiated: base},
+		{Bucket: "bucket", Object: "a", UploadID: id2, Initiated: base.Add(time.Second)},
+		{Bucket: "bucket", Object: "a", UploadID: id1, Initiated: base},
+		{Bucket: "bucket", Object: "a", UploadID: id1, Initiated: base}, // duplicate discovery
+	}
+
+	first := paginateMultipartUploads(uploads, "", "", "", "", 1)
+	requireMultipartUploadKeys(t, first, "a")
+	if !first.IsTruncated || first.NextKeyMarker != "a" || first.NextUploadIDMarker != id1 {
+		t.Fatalf("first page = %+v", first)
+	}
+
+	second := paginateMultipartUploads(uploads, "", first.NextKeyMarker, first.NextUploadIDMarker, "", 1)
+	if len(second.Uploads) != 1 || second.Uploads[0].Object != "a" || second.Uploads[0].UploadID != id2 {
+		t.Fatalf("second page uploads = %+v", second.Uploads)
+	}
+	if !second.IsTruncated || second.NextKeyMarker != "a" || second.NextUploadIDMarker != id2 {
+		t.Fatalf("second page = %+v", second)
+	}
+
+	last := paginateMultipartUploads(uploads, "", second.NextKeyMarker, second.NextUploadIDMarker, "", 1)
+	requireMultipartUploadKeys(t, last, "b")
+	if last.IsTruncated || last.NextKeyMarker != "" || last.NextUploadIDMarker != "" {
+		t.Fatalf("last page = %+v", last)
+	}
+
+	missingUploadMarker := paginateMultipartUploads(uploads, "", "a", multipartListingTestID(base.Add(2*time.Second), 4), "", 10)
+	requireMultipartUploadKeys(t, missingUploadMarker, "b")
+
+	if err := checkListMultipartArgs(t.Context(), "bucket", "", "", "not-base64=", ""); err != nil {
+		t.Fatalf("upload-id-marker without key-marker must be ignored: %v", err)
+	}
+
+	overLimit := make([]MultipartInfo, maxUploadsList+1)
+	for i := range overLimit {
+		overLimit[i] = MultipartInfo{Bucket: "bucket", Object: fmt.Sprintf("%04d", i), UploadID: fmt.Sprint(i)}
+	}
+	capped := paginateMultipartUploads(overLimit, "", "", "", "", maxUploadsList+1)
+	if capped.MaxUploads != maxUploadsList || len(capped.Uploads) != maxUploadsList || !capped.IsTruncated {
+		t.Fatalf("over-limit page = MaxUploads %d, uploads %d, truncated %t",
+			capped.MaxUploads, len(capped.Uploads), capped.IsTruncated)
+	}
+}
+
+func TestListMultipartUploadsGlobalPageAcrossPools(t *testing.T) {
+	z, bucket := consistencyPools(t)
+	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 {
+			t.Fatalf("NewMultipartUpload(%q): %v", object, err)
+		}
+	}
+	z.mpCache.Range(func(uploadID string, _ MultipartInfo) bool {
+		z.mpCache.Delete(uploadID)
+		return true
+	})
+
+	page, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireMultipartUploadKeys(t, page, objects[:2]...)
+	if !page.IsTruncated || page.NextKeyMarker != objects[1] {
+		t.Fatalf("first global page = %+v", page)
+	}
+
+	rest, err := z.ListMultipartUploads(t.Context(), bucket, "", page.NextKeyMarker, page.NextUploadIDMarker, "", 2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireMultipartUploadKeys(t, rest, objects[2:]...)
+	if rest.IsTruncated {
+		t.Fatalf("last global page is truncated: %+v", rest)
+	}
+}

cmd/object-api-input-checks.go

@@ -20,6 +20,7 @@ package cmd
 import (
 	"context"
 	"encoding/base64"
+	"errors"
 	"runtime"
 	"strings"
 
@@ -83,7 +84,8 @@ func checkListMultipartArgs(ctx context.Context, bucket, prefix, keyMarker, uplo
 	if err := checkListObjsArgs(ctx, bucket, prefix, keyMarker); err != nil {
 		return err
 	}
-	if uploadIDMarker != "" {
+	// S3 ignores upload-id-marker when key-marker is absent.
+	if uploadIDMarker != "" && keyMarker != "" {
 		if HasSuffix(keyMarker, SlashSeparator) {
 			return InvalidUploadIDKeyCombination{
 				UploadIDMarker: uploadIDMarker,
@@ -96,6 +98,9 @@ func checkListMultipartArgs(ctx context.Context, bucket, prefix, keyMarker, uplo
 				UploadID: uploadIDMarker,
 			}
 		}
+		if _, ok := multipartMarkerTime(uploadIDMarker); !ok {
+			return InvalidArgument{Bucket: bucket, Object: keyMarker, Err: errors.New("upload-id-marker must contain a native multipart upload ID")}
+		}
 	}
 	return nil
 }

cmd/storage-rest_test.go

@@ -109,6 +109,15 @@ func testStorageAPIListDir(t *testing.T, storage StorageAPI) {
 			}
 		}
 	}
+	for _, name := range []string{"one", "two", "three"} {
+		if err := storage.AppendFile(t.Context(), "foo", "bounded/"+name, []byte("x")); err != nil {
+			t.Fatal(err)
+		}
+	}
+	entries, err := storage.ListDir(t.Context(), "", "foo", "bounded", 2)
+	if err != nil || len(entries) != 2 {
+		t.Fatalf("ListDir count was lost in storage/RPC path: %v %v", entries, err)
+	}
 }
 
 func testStorageAPIReadAll(t *testing.T, storage StorageAPI) {

internal/config/api/api.go

@@ -45,6 +45,7 @@ const (
 	apiTransitionWorkers           = "transition_workers"
 	apiStaleUploadsCleanupInterval = "stale_uploads_cleanup_interval"
 	apiStaleUploadsExpiry          = "stale_uploads_expiry"
+	apiMultipartListing            = "multipart_listing"
 	apiDeleteCleanupInterval       = "delete_cleanup_interval"
 	apiDisableODirect              = "disable_odirect"
 	apiODirect                     = "odirect"
@@ -67,6 +68,7 @@ const (
 
 	EnvAPIStaleUploadsCleanupInterval = "MINIO_API_STALE_UPLOADS_CLEANUP_INTERVAL"
 	EnvAPIStaleUploadsExpiry          = "MINIO_API_STALE_UPLOADS_EXPIRY"
+	EnvAPIMultipartListing            = "MINIO_API_MULTIPART_LISTING"
 	EnvAPIDeleteCleanupInterval       = "MINIO_API_DELETE_CLEANUP_INTERVAL"
 	EnvDeleteCleanupInterval          = "MINIO_DELETE_CLEANUP_INTERVAL"
 	EnvAPIODirect                     = "MINIO_API_ODIRECT"
@@ -133,6 +135,7 @@ var (
 			Key:   apiStaleUploadsExpiry,
 			Value: "24h",
 		},
+		config.KV{Key: apiMultipartListing, Value: "strict"},
 		config.KV{
 			Key:   apiDeleteCleanupInterval,
 			Value: "5m",
@@ -178,6 +181,7 @@ 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"`
@@ -320,6 +324,10 @@ 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
 

internal/config/api/api_test.go

@@ -0,0 +1,41 @@
+// Copyright (c) 2026 Ruohang Feng
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package api
+
+import (
+	"testing"
+
+	"github.com/minio/minio/internal/config"
+)
+
+func TestMultipartListingMigrationMode(t *testing.T) {
+	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},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Setenv(EnvAPIMultipartListing, tc.override)
+			var kvs config.KVS
+			if tc.stored != "" {
+				kvs = config.KVS{{Key: apiMultipartListing, Value: tc.stored}}
+			}
+			cfg, err := LookupConfig(kvs)
+			if tc.invalid {
+				if err == nil {
+					t.Fatal("invalid migration mode silently accepted")
+				}
+				return
+			}
+			if err != nil || cfg.MultipartListing != tc.want {
+				t.Fatalf("mode=%q err=%v, want %q", cfg.MultipartListing, err, tc.want)
+			}
+		})
+	}
+}

internal/config/api/help.go

@@ -26,6 +26,12 @@ 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)`,