fix(listing): confirm current groups and bound recovery lock waits

Distinguish all-drive native absence from read uncertainty and confirm current pool targets before omitting a recovered name. Cover both existing single-pool writer lock domains, release before encoding or output, and preserve cache terminal errors without blocking the HTTP error response on remote publication.

Keep ordinary entries free of recovery I/O and locks. Bound the selective short lock-conflict wait, preserve existing retry intervals and quorum, and make retries cancelable with late-grant cleanup coverage. Restore the original raw listing callback and unchanged scanner/healing/migration callers now that recovery errors originate in the merger.

Add deterministic deletion, pool ordering, cache, lock-routing and cancellation regressions to race CI. Focused cmd/dsync/lsync race suites and verifiers pass; complete current-main integration and bare-binary performance validation are recorded in PR #239.

Refs: https://github.com/pgsty/silo/issues/218
Signed-off-by: Feng Ruohang <rh@vonng.com>

.github/workflows/go.yml

@@ -96,6 +96,12 @@ jobs:
       - name: Run multipart listing and cancellation tests under race detector
         run: go test -race ./cmd -run '^Test(MultipartListing|MultipartAbort|PaginateMultipartUploads|ListMultipartUploads)' -count=1 -timeout=5m
 
+      - name: Run listing metadata recovery, pagination and cancellation tests under race detector
+        run: go test -race ./cmd -run '^Test(ListFallback|ListingCache|ListingMerge|ListingRecovery|ListingPending|ListingEntry|ListingSecondLock|StreamMetadataParts|LocalLockerListing|DistributedListing|MergeXLV2LatestNullCrossParity|CountXLV2MixedECGroups|ListCrossParity)' -count=1 -timeout=5m
+
+      - name: Run lock cancellation and late-grant cleanup tests under race detector
+        run: go test -race ./internal/dsync ./internal/lsync -run '^Test(DRWMutex|LRWMutex)' -count=1 -timeout=5m
+
       - name: Run CPU metrics tests under race detector
         run: go test -race ./cmd -run '^TestLoadCPUMetrics' -count=1 -timeout=5m
 

CHANGELOG.md

@@ -54,35 +54,29 @@ and [complete commit range](https://github.com/pgsty/silo/compare/RELEASE.2026-0
 
 ### Object storage and replication
 
-- Make `ListMultipartUploads` discover quorum-valid uploads from durable state
+- Make durable `ListMultipartUploads` scanning the only listing path (#79):
-  across pools, erasure sets and drives, then apply S3 prefix, delimiter,
+  apply prefix, delimiter, markers, ordering and the 1,000-entry page limit
-  marker, ordering and 1,000-entry pagination semantics globally (#198). New uploads
+  globally across pools and sets, independently of node-local caches. Omit
-  store their canonical bucket and key as reserved fields in the existing
+  readable historical uploads only when both bucket/key identity fields are
-  quorum-written `xl.meta`; completion removes those upload-only fields. Native
+  absent; malformed identities, uncertain coverage and resource limits remain
-  markers remain usable after their upload is completed or canceled. Strict
+  errors. Old uploads still support part upload, completion and cancellation
-  listing returns a diagnostic 503 for legacy uploads or uncertain coverage;
+  with a known bucket/key/upload ID. No backfill or new storage layout is added.
-  the default remains the released exact-key/cache-based `legacy` behavior.
+  `MINIO_API_MULTIPART_LISTING` is retired and ignored with a startup warning;
-  Opt into strict mode only through `MINIO_API_MULTIPART_LISTING=strict`, after
+  historical shared `api multipart_listing` keys remain tolerated. The read-only
-  upgrading every writer, draining old uploads, checking the read-only admin
+  admin `multipart-preflight` reports `mode=persistent`; `ready` diagnoses the
-  `multipart-preflight` report and validating scan capacity. The process-only
+  absence of old records and no longer gates listing. Each page still scans
-  setting is not persisted into shared API configuration. Historical
+  durable state within the existing 100,000-directory-entry, 30-second and
-  `multipart_listing` keys are ignored and can be removed with a targeted
+  per-process concurrency budgets. See the
-  `mcli admin config reset ALIAS api multipart_listing` before rollback. Per-process
+  [design and limits](https://silo.pgsty.com/blog/design/list-multipart-uploads/#implementation).
-  admission, directory-entry, worker and time budgets bound scan scheduling;
+  Thanks to mr javad seydi (@mrjavadseydi) for the original metadata-and-scan
-  each page still scans durable state. See [issue #79](https://github.com/pgsty/silo/issues/79)
+  implementation (#198).
-  and its [design record](https://silo.pgsty.com/blog/design/list-multipart-uploads/).
+- Keep the default read-quorum, best-effort multipart cancellation and
-  Thanks to mr javad seydi (@mrjavadseydi) for the original implementation.
+  first-matching-pool behavior when retiring the listing switch. The optional
-- Retain released read-quorum and best-effort multipart cancellation in default
+  strict cancellation branch is removed. Wrong-target protection, upload-ID
-  legacy mode. Strict mode requires majority deletion acknowledgements and
+  path validation and peer notification with the request context remain.
-  permits retries below read quorum. When most drives were already empty,
+  HTTP 204 for an absent upload remains idempotent and does not prove physical
-  failed deletion of an observed remnant now returns 503 instead of being
+  cleanup. **Known boundary:** delayed creation writes can still restore a
-  masked by empty-drive successes. Wrong-key or wrong-bucket cancellation
+  readable upload after cancellation; this change adds no creation fence.
-  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.

cmd/api-errors.go

@@ -450,7 +450,7 @@ const (
 	ErrAdminNoSecretKey
 
 	ErrIAMNotInitialized
-	ErrMultipartListingLegacy
+	ErrMultipartListingLegacy // Reserved: legacy uploads no longer block listing.
 	ErrMultipartListingIdentity
 	ErrSlowDown
 
@@ -2191,8 +2191,6 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
 	err = unwrapAll(err)
 
 	switch err {
-	case errMultipartListingLegacy:
-		apiErr = ErrMultipartListingLegacy
 	case errMultipartListingIdentity:
 		apiErr = ErrMultipartListingIdentity
 	case errCompleteMultipartChecksumMismatch, errCompleteMultipartChecksumTypeMismatch:

cmd/common-main.go

@@ -56,6 +56,7 @@ import (
 	"github.com/minio/minio/internal/auth"
 	"github.com/minio/minio/internal/color"
 	"github.com/minio/minio/internal/config"
+	"github.com/minio/minio/internal/config/api"
 	"github.com/minio/minio/internal/handlers"
 	"github.com/minio/minio/internal/kms"
 	"github.com/minio/minio/internal/logger"
@@ -881,6 +882,9 @@ func serverHandleEnvVars() {
 		logger.Warning("%s=%s is ignored: Silo in-place updates are permanently disabled", config.EnvUpdate, updateSetting)
 	}
 	globalInplaceUpdateDisabled = true
+	if env.IsSet(api.EnvAPIMultipartListing) {
+		logger.Warning("%s is ignored: multipart listing always uses persistent metadata; uploads without identity fields are omitted", api.EnvAPIMultipartListing)
+	}
 
 	// Check if the supported credential env vars,
 	// "MINIO_ROOT_USER" and "MINIO_ROOT_PASSWORD" are provided

cmd/erasure-multipart-listing-benchmark_test.go

@@ -0,0 +1,85 @@
+// Copyright (c) 2026 Ruohang Feng
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync/atomic"
+	"testing"
+)
+
+// Run explicitly with -benchtime=1x. Fixture creation is excluded from timing;
+// each request still discovers real on-disk metadata. The 16-drive/10k case
+// deliberately exceeds the existing aggregate directory-entry budget.
+func BenchmarkMultipartListingCapacity(b *testing.B) {
+	for _, tc := range []struct {
+		drives, uploads int
+		mixed, limited  bool
+	}{
+		{4, 1_000, false, false},
+		{4, 10_000, false, false},
+		{4, 10_000, true, false},
+		{16, 1_000, false, false},
+		{16, 10_000, false, true},
+	} {
+		b.Run(fmt.Sprintf("drives=%d/uploads=%d/mixed=%t", tc.drives, tc.uploads, tc.mixed), func(b *testing.B) {
+			obj, dirs, err := prepareErasure(b.Context(), tc.drives)
+			if err != nil {
+				b.Fatal(err)
+			}
+			z := obj.(*erasureServerPools)
+			b.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) })
+			const bucket, other = "multipart-capacity", "multipart-unrelated"
+			for _, name := range []string{bucket, other} {
+				if err := z.MakeBucket(b.Context(), name, MakeBucketOptions{}); err != nil {
+					b.Fatal(err)
+				}
+			}
+			for n := range tc.uploads {
+				target := bucket
+				if tc.mixed && n%2 != 0 {
+					target = other
+				}
+				key := fmt.Sprintf("dir/%05d", n)
+				mp, err := z.NewMultipartUpload(b.Context(), target, key, ObjectOptions{})
+				if err != nil {
+					b.Fatal(err)
+				}
+				if tc.mixed && n%10 == 0 {
+					rewriteMultipartIdentity(b, z.serverPools[0].getHashedSet(key), target, key, mp.UploadID, removeMultipartIdentity)
+				}
+			}
+			z.mpCache.Clear()
+			var directoryCalls, metadataCalls atomic.Int64
+			for _, set := range z.serverPools[0].sets {
+				original := set.getDisks
+				set.getDisks = func() []StorageAPI {
+					disks := append([]StorageAPI(nil), original()...)
+					for i, disk := range disks {
+						disks[i] = multipartListingCountingDisk{disk, &directoryCalls, &metadataCalls}
+					}
+					return disks
+				}
+				b.Cleanup(func() { set.getDisks = original })
+			}
+			b.ResetTimer()
+			for range b.N {
+				got, err := z.ListMultipartUploads(b.Context(), bucket, "dir/", "", "", "", 1000)
+				if tc.limited {
+					var limited SlowDown
+					if !errors.As(err, &limited) || len(got.Uploads) != 0 {
+						b.Fatalf("overflow must return SlowDown without a partial page: %+v %v", got, err)
+					}
+				} else if err != nil || len(got.Uploads) != 1000 {
+					b.Fatalf("capacity listing: count=%d err=%v", len(got.Uploads), err)
+				}
+			}
+			b.StopTimer()
+			b.ReportMetric(float64(directoryCalls.Load())/float64(b.N), "ListDir/op")
+			b.ReportMetric(float64(metadataCalls.Load())/float64(b.N), "ReadVersion/op")
+		})
+	}
+}

cmd/erasure-multipart-listing.go

@@ -23,7 +23,6 @@ const multipartScanEntryLimit = 100_000
 
 var (
 	multipartScanSlots          = make(chan struct{}, 2)
-	errMultipartListingLegacy   = errors.New("legacy multipart uploads require a coordinated upgrade and drain")
 	errMultipartListingIdentity = errors.New("multipart upload identity is invalid")
 )
 
@@ -133,6 +132,14 @@ func (er erasureObjects) multipartIdentity(fi FileInfo, shaDir string) (string,
 		IsValidObjectPrefix(object) && er.getMultipartSHADir(bucket, object) == shaDir
 }
 
+// Only the absence of both fields identifies the historical format. Empty,
+// partial or invalid identities must not hide a damaged new upload as legacy.
+func multipartLegacyIdentity(fi FileInfo) bool {
+	_, bucketPresent := fi.Metadata[multipartMetaBucket]
+	_, objectPresent := fi.Metadata[multipartMetaObject]
+	return !bucketPresent && !objectPresent
+}
+
 func (er erasureObjects) readMultipartUploadCandidate(s *multipartScan, bucket, candidate string, source StorageAPI) (MultipartInfo, bool, bool, error) {
 	if s.ctx.Err() != nil {
 		return MultipartInfo{}, false, false, SlowDown{}
@@ -162,8 +169,8 @@ func (er erasureObjects) readMultipartUploadCandidate(s *multipartScan, bucket,
 	disks := er.getDisks()
 	metadata, errs := readAllFileInfo(s.ctx, disks, bucket, minioMetaMultipartBucket, candidate, "", false, false)
 	if s.preflight {
-		// Readiness is stronger than listing liveness: even one readable legacy
+		// The diagnostic counts even a single readable legacy copy. It does not
-		// copy must be drained, and an unreadable copy cannot certify readiness.
+		// gate listing; unreadable copies still prevent a complete diagnosis.
 		var oldest MultipartInfo
 		legacy := false
 		_, nativeID := multipartUploadTime(uploadUUID)
@@ -176,10 +183,11 @@ func (er erasureObjects) readMultipartUploadCandidate(s *multipartScan, bucket,
 			}
 			fi := metadata[i]
 			storedBucket, storedObject, valid := er.multipartIdentity(fi, shaDir)
-			if storedBucket != "" && storedObject != "" && !valid {
+			isLegacy := multipartLegacyIdentity(fi)
+			if !isLegacy && (!valid || !nativeID) {
 				return MultipartInfo{}, false, false, errMultipartListingIdentity
 			}
-			if !valid || !nativeID {
+			if isLegacy {
 				info := multipartUploadInfo(storedBucket, storedObject, uploadUUID, fi.ModTime)
 				if !legacy || info.Initiated.Before(oldest.Initiated) {
 					oldest = info
@@ -203,16 +211,29 @@ func (er erasureObjects) readMultipartUploadCandidate(s *multipartScan, bucket,
 	if err != nil {
 		return MultipartInfo{}, false, false, err
 	}
+	// The generic metadata quorum does not compare these identity fields.
+	// Contradictory readable copies cannot certify a historical upload: picking
+	// a copy with missing fields must not silently hide a new-format record.
+	legacy := multipartLegacyIdentity(fi)
 	storedBucket, storedObject, valid := er.multipartIdentity(fi, shaDir)
+	for i, other := range metadata {
+		if errs[i] != nil || !other.IsValid() {
+			continue
+		}
+		if multipartLegacyIdentity(other) != legacy ||
+			(!legacy && (other.Metadata[multipartMetaBucket] != storedBucket || other.Metadata[multipartMetaObject] != storedObject)) {
+			return MultipartInfo{}, false, false, fmt.Errorf("%w: inconsistent copies at %s", errMultipartListingIdentity, candidate)
+		}
+	}
 	info := multipartUploadInfo(storedBucket, storedObject, uploadUUID, fi.ModTime)
-	if storedBucket == "" || storedObject == "" {
+	if legacy {
 		return info, false, true, nil
 	}
 	if !valid {
 		return MultipartInfo{}, false, false, fmt.Errorf("%w: %s", errMultipartListingIdentity, candidate)
 	}
 	if _, ok := multipartUploadTime(uploadUUID); !ok {
-		return info, false, true, nil
+		return MultipartInfo{}, false, false, fmt.Errorf("%w: %s", errMultipartListingIdentity, candidate)
 	}
 	if bucket != "" && bucket != storedBucket {
 		return MultipartInfo{}, false, false, nil
@@ -220,7 +241,7 @@ func (er erasureObjects) readMultipartUploadCandidate(s *multipartScan, bucket,
 	return info, true, false, nil
 }
 
-func (er erasureObjects) scanMultipartUploads(s *multipartScan, bucket string, poolIdx, setIdx int) ([]MultipartInfo, bool, error) {
+func (er erasureObjects) scanMultipartUploads(s *multipartScan, bucket string, poolIdx, setIdx int) ([]MultipartInfo, error) {
 	disks := er.getDisks()
 	report := multipartScanSet{Pool: poolIdx, Set: setIdx, Drives: er.setDriveCount}
 	var resultErr error
@@ -272,7 +293,7 @@ func (er erasureObjects) scanMultipartUploads(s *multipartScan, bucket string, p
 		resultErr = toObjectErr(errErasureReadQuorum, bucket)
 	}
 	if resultErr != nil {
-		return nil, false, resultErr
+		return nil, resultErr
 	}
 	type candidate struct {
 		path   string
@@ -325,25 +346,23 @@ dispatch:
 	if s.ctx.Err() != nil {
 		resultErr = SlowDown{}
 	}
-	return uploads, report.LegacyUploads != 0, resultErr
+	return uploads, resultErr
 }
 
 // multipartPreflight scans all pools, sets and drives, independent of caches.
-// A majority suffices for normal listing; upgrade readiness requires every
+// A majority suffices for normal listing; a complete diagnosis requires every
-// drive to have been inspected. The operator must also upgrade all writers.
+// drive to have been inspected. Ready describes the absence of legacy copies,
+// not whether listing is enabled.
 func (z *erasureServerPools) multipartPreflight(ctx context.Context) (multipartPreflightReport, error) {
 	s, err := startMultipartScan(ctx, true)
 	if err != nil {
 		return multipartPreflightReport{}, err
 	}
 	defer s.close()
-	report := multipartPreflightReport{Complete: true, Mode: "strict"}
+	report := multipartPreflightReport{Complete: true, Mode: "persistent"}
-	if globalAPIConfig.getMultipartListingLegacy() {
-		report.Mode = "legacy"
-	}
 	for p, pool := range z.serverPools {
 		for i, set := range pool.sets {
-			_, _, _ = set.scanMultipartUploads(s, "", p, i)
+			_, _ = set.scanMultipartUploads(s, "", p, i)
 		}
 	}
 	report.Sets = s.sets

cmd/erasure-multipart-listing_test.go

@@ -31,7 +31,7 @@ func TestMultipartListingAbortBetweenHTTPPages(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	setMultipartListingTestMode(t, false)
+
 	var firstID, secondID string
 	for attempt := 0; attempt < 32; attempt++ {
 		one, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
@@ -148,28 +148,25 @@ func TestMultipartListingLegacyPreflight(t *testing.T) {
 		t.Fatal(err)
 	}
 	z.mpCache.Clear()
-	t.Run("legacy-upgrade", func(t *testing.T) {
+	for _, target := range []string{bucket, other} {
-		setMultipartListingTestMode(t, true)
+		got, err := z.ListMultipartUploads(t.Context(), target, "", "", "", "", 10)
-		exact, err := z.ListMultipartUploads(t.Context(), other, "old", "", "", "", 10)
 		if err != nil {
 			t.Fatal(err)
 		}
-		requireMultipartUploadKeys(t, exact, "old")
+		if target == bucket {
-		const empty = "multipart-legacy-empty"
+			requireMultipartUploadKeys(t, got, "a")
-		if err := z.MakeBucket(t.Context(), empty, MakeBucketOptions{}); err != nil {
+		} else {
-			t.Fatal(err)
+			requireMultipartUploadKeys(t, got)
 		}
-		got, err := z.ListMultipartUploads(t.Context(), empty, "", "", "", "", 10)
-		if err != nil || len(got.Uploads) != 0 {
-			t.Fatalf("old upload in another bucket broke legacy listing: %+v %v", got, err)
-		}
-	})
-	_, err = z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
-	if !errors.Is(err, errMultipartListingLegacy) {
-		t.Fatalf("old upload in another bucket: %v", err)
 	}
+	// Exact-looking prefixes now have the same durable semantics.
+	exact, err := z.ListMultipartUploads(t.Context(), other, "old", "", "", "", 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireMultipartUploadKeys(t, exact)
 	report, err := z.multipartPreflight(t.Context())
-	if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 {
+	if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "persistent" {
 		t.Fatalf("legacy preflight: %+v %v", report, err)
 	}
 	if err = z.AbortMultipartUpload(t.Context(), other, "old", old.UploadID, ObjectOptions{}); err != nil {
@@ -187,7 +184,7 @@ func TestMultipartListingLegacyPreflight(t *testing.T) {
 }
 
 func TestMultipartListingIdentityFallback(t *testing.T) {
-	for _, kind := range []string{"old", "corrupt", "read-failure", "wrong-bucket"} {
+	for _, kind := range []string{"old", "partial", "corrupt", "read-failure", "wrong-bucket"} {
 		t.Run(kind, func(t *testing.T) {
 			z, set, bucket := multipartListingFixture(t)
 			if _, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}); err != nil {
@@ -209,8 +206,11 @@ func TestMultipartListingIdentityFallback(t *testing.T) {
 								return FileInfo{}, errFileCorrupt
 							}
 							fi.Metadata = cloneMSS(fi.Metadata)
-							if kind == "old" {
+							if kind == "old" || kind == "partial" {
 								delete(fi.Metadata, multipartMetaBucket)
+								if kind == "old" {
+									delete(fi.Metadata, multipartMetaObject)
+								}
 							} else {
 								fi.Metadata[multipartMetaBucket] = "another-bucket"
 							}
@@ -231,7 +231,7 @@ func TestMultipartListingIdentityFallback(t *testing.T) {
 
 func TestMultipartAbortPoolsAndRetry(t *testing.T) {
 	z, bucket := consistencyPools(t)
-	setMultipartListingTestMode(t, false)
+
 	mp, err := z.serverPools[1].NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
 	if err != nil {
 		t.Fatal(err)
@@ -252,10 +252,12 @@ func TestMultipartAbortPoolsAndRetry(t *testing.T) {
 	if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 {
 		t.Fatalf("unknown pool must not acknowledge cancellation: %v", err)
 	}
+	if _, err := z.serverPools[1].GetMultipartInfo(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
+		t.Fatalf("earlier pool error changed the later pool: %v", err)
+	}
 	emptySet.getDisks = original
-	err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{})
+	if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
-	if _, ok := err.(InvalidUploadID); !ok {
+		t.Fatalf("retry after pool recovery: %v", err)
-		t.Fatalf("retry after all pools confirm absence: %v", err)
 	}
 	mp, err = z.serverPools[1].NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
 	if err != nil {
@@ -289,12 +291,13 @@ func TestMultipartAbortRetryBelowReadQuorum(t *testing.T) {
 		}}
 		return disks
 	}
-	if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err == nil {
+	if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
-		t.Fatal("lost acknowledgement must fail")
+		t.Fatalf("best-effort cleanup must retain read-quorum availability: %v", err)
 	}
 	set.getDisks = original
-	if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
+	var invalid InvalidUploadID
-		t.Fatalf("one remaining metadata copy prevented retry: %v", err)
+	if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) {
+		t.Fatalf("minority remnant must retain absent-upload result: %v", err)
 	}
 }
 
@@ -304,7 +307,7 @@ func TestMultipartListingMarkerHTTP(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	setMultipartListingTestMode(t, false)
+
 	for _, tc := range []struct {
 		key, marker string
 		status      int
@@ -388,7 +391,7 @@ func TestMultipartAbortLateCreateBoundary(t *testing.T) {
 		t.Fatal(err)
 	}
 	z := obj.(*erasureServerPools)
-	setMultipartListingTestMode(t, false)
+
 	t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) })
 	saved := globalStorageClass
 	globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: 8}})
@@ -417,9 +420,7 @@ func TestMultipartAbortLateCreateBoundary(t *testing.T) {
 	}
 	set.getDisks = func() []StorageAPI {
 		disks := append([]StorageAPI(nil), original()...)
-		for i := 2; i < 9; i++ {
+		disks[8] = nil
-			disks[i] = nil
-		}
 		return disks
 	}
 	if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
@@ -431,7 +432,7 @@ func TestMultipartAbortLateCreateBoundary(t *testing.T) {
 		}
 	}
 	set.getDisks = original
-	_, metadata, err := set.checkUploadIDExists(t.Context(), bucket, "a", mp.UploadID, true)
+	_, metadata, err := set.checkUploadIDExists(t.Context(), bucket, "a", mp.UploadID, false)
 	if err != nil {
 		t.Fatalf("late creation boundary changed; review and remove the documented limitation: %v", err)
 	}
@@ -441,10 +442,10 @@ func TestMultipartAbortLateCreateBoundary(t *testing.T) {
 			count++
 		}
 	}
-	if count != 14 {
+	if count != 8 {
-		t.Fatalf("expected fourteen resurrected copies, got %d", count)
+		t.Fatalf("expected eight resurrected copies, got %d", count)
 	}
-	t.Log("KNOWN UNRESOLVED BOUNDARY: seven delayed creation writes plus seven offline copies restore a writable upload after acknowledged cancellation")
+	t.Log("KNOWN UNRESOLVED BOUNDARY: seven delayed creation writes plus one offline copy restore a readable upload after acknowledged cancellation")
 }
 
 type multipartListingCountingDisk struct {
@@ -474,7 +475,7 @@ func TestMultipartListingScanCosts(t *testing.T) {
 		t.Fatal(err)
 	}
 	z := obj.(*erasureServerPools)
-	setMultipartListingTestMode(t, false)
+
 	t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) })
 	const bucket, otherBucket = "r9-scan-target", "r9-scan-unrelated"
 	for _, name := range []string{bucket, otherBucket} {
@@ -544,7 +545,7 @@ func multipartListingFixture(t *testing.T) (*erasureServerPools, *erasureObjects
 	if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil {
 		t.Fatal(err)
 	}
-	setMultipartListingTestMode(t, false)
+
 	return z, z.serverPools[0].getHashedSet("a"), bucket
 }
 
@@ -588,45 +589,6 @@ func TestMultipartListingMissingMarkers(t *testing.T) {
 	}
 }
 
-func TestMultipartListingAbortRecovery(t *testing.T) {
-	for _, offline := range []int{1, 2} {
-		t.Run(fmt.Sprint(offline), func(t *testing.T) {
-			z, set, bucket := multipartListingFixture(t)
-			mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
-			if err != nil {
-				t.Fatal(err)
-			}
-			original := set.getDisks
-			t.Cleanup(func() { set.getDisks = original })
-			set.getDisks = func() []StorageAPI {
-				disks := append([]StorageAPI(nil), original()...)
-				for i := 0; i < offline; i++ {
-					disks[i] = nil
-				}
-				return disks
-			}
-			err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{})
-			if offline == 1 && err != nil {
-				t.Fatal(err)
-			}
-			if offline == 2 && (err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503) {
-				t.Fatalf("two offline drives must not acknowledge cancellation: %v", err)
-			}
-			set.getDisks = original
-			if offline == 2 {
-				// Retry a partially completed deletion after the original disks return.
-				if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
-					t.Fatal(err)
-				}
-			}
-			got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 100)
-			if err != nil || len(got.Uploads) != 0 {
-				t.Fatalf("canceled upload reappeared: %+v %v", got, err)
-			}
-		})
-	}
-}
-
 func TestMultipartListingCoverage(t *testing.T) {
 	z, set, bucket := multipartListingFixture(t)
 	mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
@@ -670,7 +632,7 @@ func TestMultipartListingBudgetAndAdmission(t *testing.T) {
 	}
 	defer scan.close()
 	scan.remaining.Store(1)
-	_, _, err = set.scanMultipartUploads(scan, bucket, 0, 0)
+	_, err = set.scanMultipartUploads(scan, bucket, 0, 0)
 	var limited SlowDown
 	if !errors.As(err, &limited) {
 		t.Fatalf("budget: %v", err)
@@ -695,7 +657,7 @@ func TestMultipartListingAdmissionHTTP(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	setMultipartListingTestMode(t, false)
+
 	for range cap(multipartScanSlots) {
 		scan, err := startMultipartScan(t.Context(), false)
 		if err != nil {
@@ -743,7 +705,7 @@ func TestMultipartListingPreflightMinorityLegacy(t *testing.T) {
 		}
 	}
 	report, err := z.multipartPreflight(t.Context())
-	if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 {
+	if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "persistent" {
 		t.Fatalf("minority legacy copy must prevent readiness: %+v %v", report, err)
 	}
 }

cmd/erasure-multipart-migration_test.go

@@ -12,41 +12,13 @@ import (
 	"sync/atomic"
 	"testing"
 
-	"github.com/minio/minio/internal/config/api"
 	"github.com/minio/minio/internal/config/storageclass"
 	"github.com/minio/minio/internal/dsync"
 	"github.com/minio/minio/internal/grid"
 	xnet "github.com/pgsty/silo-pkg/v3/net"
 )
 
-func setMultipartListingTestMode(t *testing.T, legacy bool) {
+func TestMultipartAbortDefaultAvailability(t *testing.T) {
-	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) {
@@ -56,7 +28,7 @@ func TestMultipartAbortLegacyAvailability(t *testing.T) {
 				}
 				z := obj.(*erasureServerPools)
 				t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) })
-				setMultipartListingTestMode(t, true)
+
 				previous := globalStorageClass
 				globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: drives / 2}})
 				t.Cleanup(func() { globalStorageClass.Update(previous) })
@@ -91,11 +63,11 @@ func TestMultipartAbortLegacyAvailability(t *testing.T) {
 	}
 }
 
-func TestMultipartAbortLegacyPoolOrder(t *testing.T) {
+func TestMultipartAbortDefaultPoolOrder(t *testing.T) {
 	for _, owner := range []int{0, 1} {
 		t.Run(fmt.Sprintf("owner=%d", owner), func(t *testing.T) {
 			z, bucket := consistencyPools(t)
-			setMultipartListingTestMode(t, true)
+
 			mp, err := z.serverPools[owner].NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
 			if err != nil {
 				t.Fatal(err)
@@ -106,7 +78,7 @@ func TestMultipartAbortLegacyPoolOrder(t *testing.T) {
 			emptySet.getDisks = func() []StorageAPI { return make([]StorageAPI, emptySet.setDriveCount) }
 			err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{})
 			if owner == 0 && err != nil {
-				t.Fatalf("unrelated later pool blocked legacy cancellation: %v", err)
+				t.Fatalf("unrelated later pool blocked cancellation: %v", err)
 			}
 			if owner == 1 {
 				if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 {
@@ -124,123 +96,56 @@ func TestMultipartAbortLegacyPoolOrder(t *testing.T) {
 	}
 }
 
-func TestMultipartAbortMinorityLegacyCleanup(t *testing.T) {
-	for _, failDelete := range []bool{false, true} {
-		t.Run(fmt.Sprintf("delete-fails=%v", failDelete), func(t *testing.T) {
-			z, set, bucket := multipartListingFixture(t)
-			mp, err := z.NewMultipartUpload(t.Context(), bucket, "old", ObjectOptions{})
-			if err != nil {
-				t.Fatal(err)
-			}
-			path := set.getUploadIDDir(bucket, "old", mp.UploadID)
-			original := set.getDisks
-			disks := original()
-			fi, err := disks[0].ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, path, "", ReadOptions{})
-			if err != nil {
-				t.Fatal(err)
-			}
-			delete(fi.Metadata, multipartMetaBucket)
-			delete(fi.Metadata, multipartMetaObject)
-			if err := disks[0].WriteMetadata(t.Context(), bucket, minioMetaMultipartBucket, path, fi); err != nil {
-				t.Fatal(err)
-			}
-			for _, d := range disks[1:] {
-				if err := d.Delete(t.Context(), minioMetaMultipartBucket, path, DeleteOptions{Recursive: true}); err != nil {
-					t.Fatal(err)
-				}
-			}
-			report, err := z.multipartPreflight(t.Context())
-			if err != nil || report.Ready || report.LegacyUploads != 1 {
-				t.Fatalf("invalid legacy fixture: %+v %v", report, err)
-			}
-			if failDelete {
-				set.getDisks = func() []StorageAPI {
-					wrapped := append([]StorageAPI(nil), disks...)
-					wrapped[0] = multipartListingFaultDisk{StorageAPI: disks[0], delete: func(context.Context, string, string, DeleteOptions) error { return errFaultyDisk }}
-					return wrapped
-				}
-				t.Cleanup(func() { set.getDisks = original })
-				err := z.AbortMultipartUpload(t.Context(), bucket, "old", mp.UploadID, ObjectOptions{})
-				if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 {
-					t.Fatalf("empty drives masked failed deletion of the observed replica: %v", err)
-				}
-				if _, present := z.mpCache.Load(mp.UploadID); !present {
-					t.Fatal("failed cancellation evicted upload from cache")
-				}
-				set.getDisks = original
-			}
-			if err := z.AbortMultipartUpload(t.Context(), bucket, "old", mp.UploadID, ObjectOptions{}); err != nil {
-				t.Fatal(err)
-			}
-			report, err = z.multipartPreflight(t.Context())
-			if err != nil || !report.Ready || report.LegacyUploads != 0 {
-				t.Fatalf("acknowledged cleanup left online legacy remnants: %+v %v", report, err)
-			}
-			if _, present := z.mpCache.Load(mp.UploadID); present {
-				t.Fatal("successful cancellation retained cache entry")
-			}
-		})
-	}
-}
-
 func TestMultipartAbortWrongTargetKeepsCache(t *testing.T) {
-	for _, legacy := range []bool{true, false} {
+	z, _, bucket := multipartListingFixture(t)
-		t.Run(fmt.Sprintf("legacy=%v", legacy), func(t *testing.T) {
+
-			z, _, bucket := multipartListingFixture(t)
+	const other = "multipart-other-bucket"
-			setMultipartListingTestMode(t, legacy)
+	if err := z.MakeBucket(t.Context(), other, MakeBucketOptions{}); err != nil {
-			const other = "multipart-other-bucket"
+		t.Fatal(err)
-			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 {
-			mp, err := z.NewMultipartUpload(t.Context(), bucket, "valid-key", ObjectOptions{})
+		t.Fatal(err)
-			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{})
-			for _, target := range [][2]string{{bucket, "wrong-key"}, {other, "valid-key"}} {
+		var invalid InvalidUploadID
-				err := z.AbortMultipartUpload(t.Context(), target[0], target[1], mp.UploadID, ObjectOptions{})
+		if !errors.As(err, &invalid) {
-				var invalid InvalidUploadID
+			t.Fatalf("wrong target result: %v", err)
-				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 _, 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.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)
-			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) {
-			var invalid InvalidUploadID
+		t.Fatalf("repeat abort: %v", err)
-			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} {
+	z, _, bucket := multipartListingFixture(t)
-		t.Run(fmt.Sprintf("legacy=%v", legacy), func(t *testing.T) {
+
-			z, _, bucket := multipartListingFixture(t)
+	mp, err := z.NewMultipartUpload(t.Context(), bucket, "keep", ObjectOptions{})
-			setMultipartListingTestMode(t, legacy)
+	if err != nil {
-			mp, err := z.NewMultipartUpload(t.Context(), bucket, "keep", ObjectOptions{})
+		t.Fatal(err)
-			if err != nil {
+	}
-				t.Fatal(err)
+	for _, suffix := range []string{".", "..", "../other", "a/b", "a\\b", ""} {
-			}
+		id := base64.RawURLEncoding.EncodeToString([]byte("deployment." + suffix))
-			for _, suffix := range []string{".", "..", "../other", "a/b", "a\\b", ""} {
+		err := z.AbortMultipartUpload(t.Context(), bucket, "keep", id, ObjectOptions{})
-				id := base64.RawURLEncoding.EncodeToString([]byte("deployment." + suffix))
+		var invalid InvalidUploadID
-				err := z.AbortMultipartUpload(t.Context(), bucket, "keep", id, ObjectOptions{})
+		if !errors.As(err, &invalid) {
-				var invalid InvalidUploadID
+			t.Fatalf("unsafe ID accepted: suffix=%q err=%v", suffix, err)
-				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)
-				if _, err := z.GetMultipartInfo(t.Context(), bucket, "keep", mp.UploadID, ObjectOptions{}); err != nil {
+		}
-					t.Fatalf("valid upload harmed by rejected ID: %v", err)
-				}
-			}
-		})
 	}
 }
 
@@ -275,34 +180,30 @@ func TestMultipartAbortPeerNotification(t *testing.T) {
 	set.nsMutex = newNSLock(true)
 	set.getLockers = func() ([]dsync.NetLocker, string) { return lockers, "multipart-test" }
 	t.Cleanup(func() { set.nsMutex, set.getLockers = oldMutex, oldLockers })
-	for _, legacy := range []bool{true, false} {
+
-		t.Run(fmt.Sprintf("legacy=%v", legacy), func(t *testing.T) {
+	for range 4 {
-			setMultipartListingTestMode(t, legacy)
+		mp, err := z.NewMultipartUpload(t.Context(), bucket, "valid", ObjectOptions{})
-			for range 4 {
+		if err != nil {
-				mp, err := z.NewMultipartUpload(t.Context(), bucket, "valid", ObjectOptions{})
+			t.Fatal(err)
-				if err != nil {
+		}
-					t.Fatal(err)
+		before := notifications.Load()
-				}
+		var invalid InvalidUploadID
-				before := notifications.Load()
+		if err := z.AbortMultipartUpload(t.Context(), bucket, "wrong", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) {
-				var invalid InvalidUploadID
+			t.Fatalf("wrong target: %v", err)
-				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 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 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")
-				if notifications.Load() != before+1 {
+		}
-					t.Fatal("successful abort lost peer notification after unlocking")
-				}
-			}
-		})
 	}
 }
 
-func TestMultipartAbortStrictMajority(t *testing.T) {
+func TestMultipartAbortBestEffortCleanup(t *testing.T) {
 	z, set, bucket := multipartListingFixture(t)
 	mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
 	if err != nil {
@@ -317,15 +218,19 @@ func TestMultipartAbortStrictMajority(t *testing.T) {
 	}
 	t.Cleanup(func() { set.getDisks = original })
 	if err := z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
-		t.Fatalf("ordinary majority cancellation was tightened: %v", err)
+		t.Fatalf("read-quorum cancellation was tightened: %v", err)
 	}
-	// One failed deletion is tolerated in the ordinary majority path. Retrying
+	// Best-effort deletion may leave a remnant. Retrying after recovery cleans
-	// after recovery must also clean the now-minority remnant.
+	// it, but the read quorum no longer recognizes it as an active upload.
 	if _, err := disks[0].ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, set.getUploadIDDir(bucket, "a", mp.UploadID), "", ReadOptions{}); err != nil {
 		t.Fatalf("fault fixture did not leave a remnant: %v", err)
 	}
 	set.getDisks = original
-	if err := z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
+	var invalid InvalidUploadID
-		t.Fatal(err)
+	if err := z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) {
+		t.Fatalf("minority retry must preserve default absent-upload result: %v", err)
+	}
+	if _, err := disks[0].ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, set.getUploadIDDir(bucket, "a", mp.UploadID), "", ReadOptions{}); !errors.Is(err, errFileNotFound) {
+		t.Fatalf("best-effort retry did not clean remnant: %v", err)
 	}
 }

cmd/erasure-multipart-unified_test.go

@@ -0,0 +1,256 @@
+// Copyright (c) 2026 Ruohang Feng
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"context"
+	"encoding/xml"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"slices"
+	"testing"
+)
+
+func rewriteMultipartIdentity(t testing.TB, set *erasureObjects, bucket, key, id string, change func(map[string]string)) {
+	t.Helper()
+	fi, metadata, err := set.checkUploadIDExists(t.Context(), bucket, key, id, true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for i := range metadata {
+		change(metadata[i].Metadata)
+	}
+	if _, err := writeAllMetadata(t.Context(), set.getDisks(), bucket, minioMetaMultipartBucket,
+		set.getUploadIDDir(bucket, key, id), metadata, fi.WriteQuorum(set.defaultWQuorum())); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func removeMultipartIdentity(metadata map[string]string) {
+	delete(metadata, multipartMetaBucket)
+	delete(metadata, multipartMetaObject)
+}
+
+// The actual S3 handler must use the same semantics for every prefix and page,
+// even with old records and a stale, misleading node-local cache.
+func TestMultipartListingMixedHTTP(t *testing.T) {
+	z, set, _ := multipartListingFixture(t)
+	bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads"}, MakeBucketOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, key := range []string{"dir/old", "dir/a", "dir/sub/b", "outside"} {
+		mp, err := z.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if key == "dir/old" {
+			rewriteMultipartIdentity(t, set, bucket, key, mp.UploadID, removeMultipartIdentity)
+		}
+	}
+	z.mpCache.Clear()
+	z.mpCache.Store("stale", MultipartInfo{Bucket: bucket, Object: "ghost", UploadID: "stale"})
+	list := func(prefix, keyMarker, uploadMarker, delimiter string) ListMultipartUploadsResponse {
+		t.Helper()
+		req, err := newTestSignedRequestV4(http.MethodGet,
+			getListMultipartUploadsURLWithParams("", bucket, prefix, keyMarker, uploadMarker, delimiter, "1"),
+			0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		rec := httptest.NewRecorder()
+		router.ServeHTTP(rec, req)
+		var got ListMultipartUploadsResponse
+		if rec.Code != http.StatusOK {
+			t.Fatalf("list %q: HTTP %d %s", prefix, rec.Code, rec.Body.String())
+		}
+		if err := xml.Unmarshal(rec.Body.Bytes(), &got); err != nil {
+			t.Fatal(err)
+		}
+		return got
+	}
+	var keys []string
+	keyMarker, uploadMarker := "", ""
+	for page := 0; ; page++ {
+		if page >= 4 {
+			t.Fatal("pagination did not terminate")
+		}
+		got := list("", keyMarker, uploadMarker, "")
+		if len(got.Uploads) != 1 {
+			t.Fatalf("page size: %+v", got)
+		}
+		keys = append(keys, got.Uploads[0].Key)
+		if !got.IsTruncated {
+			break
+		}
+		keyMarker, uploadMarker = got.NextKeyMarker, got.NextUploadIDMarker
+	}
+	if !slices.Equal(keys, []string{"dir/a", "dir/sub/b", "outside"}) {
+		t.Fatalf("mixed full listing: %v", keys)
+	}
+	first := list("dir/", "", "", "/")
+	if len(first.Uploads) != 1 || first.Uploads[0].Key != "dir/a" || !first.IsTruncated {
+		t.Fatalf("prefix first page: %+v", first)
+	}
+	rest := list("dir/", first.NextKeyMarker, first.NextUploadIDMarker, "/")
+	if len(rest.Uploads) != 0 || rest.IsTruncated || len(rest.CommonPrefixes) != 1 || rest.CommonPrefixes[0].Prefix != "dir/sub/" {
+		t.Fatalf("prefix continuation: %+v", rest)
+	}
+	if got := list("dir/old", "", "", ""); len(got.Uploads) != 0 || got.IsTruncated {
+		t.Fatalf("historical exact-looking prefix: %+v", got)
+	}
+}
+
+func TestMultipartListingInvalidIdentity(t *testing.T) {
+	for _, kind := range []string{"missing-bucket", "missing-object", "empty-bucket", "empty-object", "empty-both", "wrong-hash", "read-failure"} {
+		t.Run(kind, func(t *testing.T) {
+			z, set, bucket := multipartListingFixture(t)
+			mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			rewriteMultipartIdentity(t, set, bucket, "a", mp.UploadID, func(metadata map[string]string) {
+				switch kind {
+				case "missing-bucket":
+					delete(metadata, multipartMetaBucket)
+				case "missing-object":
+					delete(metadata, multipartMetaObject)
+				case "empty-bucket":
+					metadata[multipartMetaBucket] = ""
+				case "empty-object":
+					metadata[multipartMetaObject] = ""
+				case "empty-both":
+					metadata[multipartMetaBucket], metadata[multipartMetaObject] = "", ""
+				case "wrong-hash":
+					metadata[multipartMetaObject] = "wrong-key"
+				case "read-failure":
+					removeMultipartIdentity(metadata)
+				}
+			})
+			if kind == "read-failure" {
+				original := set.getDisks
+				t.Cleanup(func() { set.getDisks = original })
+				set.getDisks = func() []StorageAPI {
+					disks := append([]StorageAPI(nil), original()...)
+					for i, disk := range disks {
+						disks[i] = multipartListingFaultDisk{StorageAPI: disk, read: func(context.Context, string, string, string, string, ReadOptions) (FileInfo, error) {
+							return FileInfo{}, errFaultyDisk
+						}}
+					}
+					return disks
+				}
+			}
+			got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
+			if err == nil || len(got.Uploads) != 0 || toAPIError(t.Context(), err).HTTPStatusCode != http.StatusServiceUnavailable {
+				t.Fatalf("invalid/unreadable record returned a successful page: %+v %v", got, err)
+			}
+			if kind != "read-failure" && !errors.Is(err, errMultipartListingIdentity) {
+				t.Fatalf("identity error: %v", err)
+			}
+			report, err := z.multipartPreflight(t.Context())
+			if err != nil || report.Complete || report.Ready || report.LegacyUploads != 0 {
+				t.Fatalf("invalid/unreadable record counted as historical: %+v %v", report, err)
+			}
+		})
+	}
+}
+
+func TestMultipartListingRestart(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	obj, dirs, err := prepareErasure(ctx, 4)
+	if err != nil {
+		cancel()
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { cancel(); obj.Shutdown(context.Background()); removeRoots(dirs) })
+	const bucket, key = "multipart-restart", "dir/new"
+	if err := obj.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	mp, err := obj.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	cancel()
+	if err := obj.Shutdown(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	reopened, _, err := initObjectLayer(t.Context(), mustGetPoolEndpoints(0, dirs...))
+	if err != nil {
+		t.Fatal(err)
+	}
+	obj = reopened
+	if _, cached := obj.(*erasureServerPools).mpCache.Load(mp.UploadID); cached {
+		t.Fatal("restart unexpectedly preserved the upload cache")
+	}
+	got, err := obj.ListMultipartUploads(t.Context(), bucket, "dir/", "", "", "", 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireMultipartUploadKeys(t, got, key)
+	if got.Uploads[0].UploadID != mp.UploadID {
+		t.Fatal("restart changed upload identity")
+	}
+}
+
+func TestMultipartListingConflictingIdentityCopies(t *testing.T) {
+	for _, oldCopies := range []int{1, 3} {
+		z, set, bucket := multipartListingFixture(t)
+		mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		path := set.getUploadIDDir(bucket, "a", mp.UploadID)
+		for _, disk := range set.getDisks()[:oldCopies] {
+			fi, err := disk.ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, path, "", ReadOptions{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			removeMultipartIdentity(fi.Metadata)
+			if err := disk.WriteMetadata(t.Context(), bucket, minioMetaMultipartBucket, path, fi); err != nil {
+				t.Fatal(err)
+			}
+		}
+		got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
+		if !errors.Is(err, errMultipartListingIdentity) || len(got.Uploads) != 0 {
+			t.Fatalf("%d contradictory copies returned a successful page: %+v %v", oldCopies, got, err)
+		}
+	}
+}
+
+func TestMultipartListingUnsupportedStoredID(t *testing.T) {
+	for _, historical := range []bool{false, true} {
+		z, set, bucket := multipartListingFixture(t)
+		mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		oldPath := set.getUploadIDDir(bucket, "a", mp.UploadID)
+		unsupportedPath := pathJoin(set.getMultipartSHADir(bucket, "a"), "unsupported-id")
+		for _, disk := range set.getDisks() {
+			fi, err := disk.ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, oldPath, "", ReadOptions{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			if historical {
+				removeMultipartIdentity(fi.Metadata)
+			}
+			if err := disk.WriteMetadata(t.Context(), bucket, minioMetaMultipartBucket, unsupportedPath, fi); err != nil {
+				t.Fatal(err)
+			}
+			if err := disk.Delete(t.Context(), minioMetaMultipartBucket, oldPath, DeleteOptions{Recursive: true}); err != nil {
+				t.Fatal(err)
+			}
+		}
+		got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
+		if historical {
+			if err != nil || len(got.Uploads) != 0 {
+				t.Fatalf("unidentified historical ID: %+v %v", got, err)
+			}
+		} else if !errors.Is(err, errMultipartListingIdentity) {
+			t.Fatalf("invalid new ID silently treated as historical: %+v %v", got, err)
+		}
+	}
+}

cmd/erasure-multipart.go

@@ -414,7 +414,7 @@ func paginateMultipartUploads(uploads []MultipartInfo, prefix, keyMarker, upload
 }
 
 // listMultipartUploadsExact preserves the hashed exact-object lookup used by
-// multipart write placement and by rolling-upgrade legacy mode.
+// multipart write placement. Public listing always uses the durable scan.
 func (er erasureObjects) listMultipartUploadsExact(ctx context.Context, bucket, object, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (result ListMultipartsInfo, err error) {
 	auditObjectErasureSet(ctx, "ListMultipartUploads", object, &er)
 
@@ -527,13 +527,10 @@ func (er erasureObjects) ListMultipartUploads(ctx context.Context, bucket, prefi
 		return ListMultipartsInfo{}, err
 	}
 	defer scan.close()
-	uploads, legacy, err := er.scanMultipartUploads(scan, bucket, 0, 0)
+	uploads, err := er.scanMultipartUploads(scan, bucket, 0, 0)
 	if err != nil {
 		return ListMultipartsInfo{}, err
 	}
-	if legacy {
-		return ListMultipartsInfo{}, errMultipartListingLegacy
-	}
 	return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil
 }
 
@@ -1788,11 +1785,9 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 	return fi.ToObjectInfo(bucket, object, opts.Versioned || opts.VersionSuspended), nil
 }
 
-// abortMultipartUpload retains read-quorum validation and best-effort cleanup
+// abortMultipartUpload retains read-quorum validation and best-effort cleanup.
-// in legacy mode. Strict mode requires majority deletion acknowledgements and
+// It does not fence creation writes still executing after a storage timeout.
-// permits retrying remnants below read quorum. Neither mode fences creation
+func (er erasureObjects) abortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) (bool, error) {
-// 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)
 	}
@@ -1804,70 +1799,19 @@ func (er erasureObjects) abortMultipartUpload(ctx context.Context, bucket, objec
 	if !ok || internalID == "" || internalID == "." || internalID == ".." || strings.ContainsAny(internalID, "/\\") {
 		return false, InvalidUploadID{Bucket: bucket, Object: object, UploadID: uploadID}
 	}
-	if legacy {
+	defer er.deleteAll(ctx, minioMetaMultipartBucket, er.getUploadIDDir(bucket, object, uploadID))
-		// Keep the released read-quorum and best-effort cleanup behavior.
+	_, _, err = er.checkUploadIDExists(ctx, bucket, object, uploadID, false)
-		// The upload ID safety check above applies to both modes.
+	err = toObjectErr(err, bucket, object, uploadID)
-		defer er.deleteAll(ctx, minioMetaMultipartBucket, er.getUploadIDDir(bucket, object, uploadID))
+	if _, absent := err.(InvalidUploadID); absent {
-		_, _, err := er.checkUploadIDExists(ctx, bucket, object, uploadID, false)
-		err = toObjectErr(err, bucket, object, uploadID)
-		if _, absent := err.(InvalidUploadID); absent {
-			return false, nil
-		}
-		return err == nil, err
-	}
-	disks := er.getDisks()
-	uploadPath := er.getUploadIDDir(bucket, object, uploadID)
-	_, errs := readAllFileInfo(ctx, disks, bucket, minioMetaMultipartBucket, uploadPath, "", false, false)
-	quorum := er.setDriveCount/2 + 1
-	found, absent := false, 0
-	observed := make([]bool, len(disks))
-	for i, err := range errs {
-		switch {
-		case err == nil, errors.Is(err, errFileCorrupt):
-			found = true
-			observed[i] = true
-		case errors.Is(err, errFileNotFound), errors.Is(err, errFileVersionNotFound):
-			absent++
-		}
-	}
-	if absent >= quorum && !found {
 		return false, nil
 	}
-	if !found {
+	return err == nil, err
-		return false, toObjectErr(errErasureReadQuorum, bucket, object, uploadID)
-	}
-	g := errgroup.WithNErrs(len(disks))
-	for i, disk := range disks {
-		g.Go(func() error {
-			if disk == nil {
-				return errDiskNotFound
-			}
-			err := disk.Delete(ctx, minioMetaMultipartBucket, uploadPath, DeleteOptions{Recursive: true, Immediate: false})
-			if errors.Is(err, errFileNotFound) || errors.Is(err, errFileVersionNotFound) {
-				return nil
-			}
-			return err
-		}, i)
-	}
-	deleteErrs := g.Wait()
-	if err := reduceWriteQuorumErrs(ctx, deleteErrs, nil, quorum); err != nil {
-		return true, toObjectErr(err, bucket, object, uploadID)
-	}
-	if absent >= quorum {
-		// Missing disks must not mask a failed cleanup of known remnants.
-		for i, found := range observed {
-			if found && deleteErrs[i] != nil {
-				return true, toObjectErr(errErasureWriteQuorum, bucket, object, uploadID)
-			}
-		}
-	}
-	return true, nil
 }
 
-// AbortMultipartUpload cancels an upload using the configured mode. Offline
+// AbortMultipartUpload cancels an upload using read-quorum validation.
-// part data may still need stale-upload cleanup after its drives return.
+// Offline part data may still need stale-upload cleanup after its drives return.
 func (er erasureObjects) AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) (err error) {
-	found, err := er.abortMultipartUpload(ctx, bucket, object, uploadID, opts, globalAPIConfig.getMultipartListingLegacy())
+	found, err := er.abortMultipartUpload(ctx, bucket, object, uploadID, opts)
 	if err != nil {
 		return err
 	}

cmd/erasure-object.go

@@ -729,6 +729,57 @@ func readAllXL(ctx context.Context, disks []StorageAPI, bucket, object string, r
 	return pickLatestQuorumFilesInfo(ctx, rawFileInfos, errs, bucket, object, readData, inclFreeVers)
 }
 
+// readListingFileInfo is the read-only, latest-version metadata path for listing
+// recovery. Ordinary GET keeps its fast-read, healing and dangling semantics.
+// Absence is proved before decoding can replace a native storage error: every
+// configured slot must have returned FileNotFound in this read.
+func (er erasureObjects) readListingFileInfo(ctx context.Context, bucket, object string) (FileInfo, bool, error) {
+	disks := er.getDisks()
+	if len(disks) == 0 || len(disks) != er.setDriveCount {
+		return FileInfo{}, false, listingReadQuorumError(bucket, object)
+	}
+	// An offline slot is not evidence of absence, even if a test or transport
+	// implementation could return a cached not-found response for it.
+	online := append([]StorageAPI(nil), disks...)
+	for i, disk := range online {
+		if disk != nil && !disk.IsOnline() {
+			online[i] = nil
+		}
+	}
+	raw, errs := readAllRawFileInfo(ctx, online, bucket, object, false)
+	if err := ctx.Err(); err != nil {
+		return FileInfo{}, false, err
+	}
+	absent := true
+	for _, err := range errs {
+		absent = absent && err == errFileNotFound
+	}
+	if absent {
+		return FileInfo{}, true, nil
+	}
+
+	// Keep the same latest-version and actual object parity quorum as GET,
+	// without its opportunistic dangling deletion or generic NotFound mapping.
+	meta, errs := pickLatestQuorumFilesInfo(ctx, raw, errs, bucket, object, false, false)
+	quorum, _, err := objectQuorumFromMeta(ctx, meta, errs, er.defaultParityCount)
+	if err == nil {
+		err = reduceReadQuorumErrs(ctx, errs, objectOpIgnoredErrs, quorum)
+	}
+	if err != nil {
+		return FileInfo{}, false, listingReadQuorumError(bucket, object)
+	}
+	_, modTime, etag := listOnlineDisks(disks, meta, errs, quorum)
+	fi, err := pickValidFileInfo(ctx, meta, modTime, etag, quorum)
+	if err != nil || (!fi.Deleted && len(fi.Erasure.Distribution) != len(disks)) {
+		return FileInfo{}, false, listingReadQuorumError(bucket, object)
+	}
+	return fi, false, nil
+}
+
+func listingReadQuorumError(bucket, object string) error {
+	return InsufficientReadQuorum{Bucket: bucket, Object: decodeDirObject(object), Err: errErasureReadQuorum, Type: RQInconsistentMeta}
+}
+
 func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object string, opts ObjectOptions, readData bool) (FileInfo, []FileInfo, []StorageAPI, error) {
 	rawArr := make([]RawFileInfo, er.setDriveCount)
 	metaArr := make([]FileInfo, er.setDriveCount)

cmd/erasure-server-pool.go

@@ -1916,9 +1916,6 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p
 	if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil {
 		return ListMultipartsInfo{}, toObjectErr(err, bucket)
 	}
-	if globalAPIConfig.getMultipartListingLegacy() {
-		return z.listMultipartUploadsLegacy(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
-	}
 	scan, err := startMultipartScan(ctx, false)
 	if err != nil {
 		return ListMultipartsInfo{}, err
@@ -1926,70 +1923,22 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p
 	defer scan.close()
 
 	var uploads []MultipartInfo
-	var keyless bool
 	for idx, pool := range z.serverPools {
 		if z.IsSuspended(idx) {
 			continue
 		}
-		poolUploads, poolKeyless, err := pool.scanMultipartUploads(scan, bucket, idx)
+		poolUploads, err := pool.scanMultipartUploads(scan, bucket, idx)
 		if err != nil {
 			return ListMultipartsInfo{}, err
 		}
 		uploads = append(uploads, poolUploads...)
-		keyless = keyless || poolKeyless
 	}
 
-	// The old format cannot be enumerated authoritatively. Migration mode is
+	// Confirmed historical records lack a recoverable bucket/key and are
-	// explicit: another bucket must never silently change this API's semantics.
+	// omitted by the scan. They do not block listing identifiable uploads.
-	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
-	poolResult.Prefix = prefix
-	poolResult.Delimiter = delimiter
-
-	// if no prefix provided, return the list from cache
-	if prefix == "" {
-		if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil {
-			return ListMultipartsInfo{}, toObjectErr(err, bucket)
-		}
-
-		z.mpCache.Range(func(_ string, mp MultipartInfo) bool {
-			if mp.Bucket == bucket {
-				poolResult.Uploads = append(poolResult.Uploads, mp)
-			}
-			return true
-		})
-		sort.Slice(poolResult.Uploads, func(i int, j int) bool {
-			return poolResult.Uploads[i].Initiated.Before(poolResult.Uploads[j].Initiated)
-		})
-		return poolResult, nil
-	}
-
-	if z.SinglePool() {
-		return z.serverPools[0].getHashedSet(prefix).listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
-	}
-
-	for idx, pool := range z.serverPools {
-		if z.IsSuspended(idx) {
-			continue
-		}
-		result, err := pool.getHashedSet(prefix).listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
-		if err != nil {
-			return result, err
-		}
-		poolResult.Uploads = append(poolResult.Uploads, result.Uploads...)
-	}
-
-	return poolResult, nil
-}
-
 // Initiate a new multipart upload on a hashedSet based on object name.
 func (z *erasureServerPools) NewMultipartUpload(ctx context.Context, bucket, object string, opts ObjectOptions) (mp *NewMultipartUploadResult, err error) {
 	if err := checkNewMultipartArgs(ctx, bucket, object); err != nil {
@@ -2199,28 +2148,15 @@ func (z *erasureServerPools) AbortMultipartUpload(ctx context.Context, bucket, o
 	ctx = lkctx.Context()
 	defer lk.Unlock(lkctx)
 
-	legacy := globalAPIConfig.getMultipartListingLegacy()
-	found := false
-	var firstErr error
 	for idx, pool := range z.serverPools {
 		if z.IsSuspended(idx) {
 			continue
 		}
-		poolFound, err := pool.getHashedSet(object).abortMultipartUpload(ctx, bucket, object, uploadID, opts, legacy)
+		found, err := pool.getHashedSet(object).abortMultipartUpload(ctx, bucket, object, uploadID, opts)
-		if legacy && (poolFound || err != nil) {
+		if found || err != nil {
-			// Match the released first-matching-pool behavior.
+			// Preserve the default first-matching-pool behavior.
 			return err
 		}
-		found = found || poolFound
-		if err != nil && firstErr == nil {
-			firstErr = err
-		}
-	}
-	if firstErr != nil {
-		return firstErr
-	}
-	if found {
-		return nil
 	}
 	return InvalidUploadID{
 		Bucket:   bucket,
@@ -2651,9 +2587,9 @@ func (z *erasureServerPools) Walk(ctx context.Context, bucket, prefix string, re
 					minDisks:       listingQuorum,
 					reportNotFound: false,
 					agreed:         send,
-					partial: func(entries metaCacheEntries, _ []error) {
+					partial: func(entries metaCacheEntries, errs []error) {
-						entry, ok := entries.resolve(&resolver)
+						entry := set.resolveListEntry(entries, errs, &resolver)
-						if ok {
+						if entry != nil {
 							send(*entry)
 						}
 					},
@@ -2755,7 +2691,11 @@ func (z *erasureServerPools) Walk(ctx context.Context, bucket, prefix string, re
 		// Merge all entries from all disks.
 		// We leave quorum at 1, since entries are already resolved to have the desired quorum.
 		// mergeEntryChannels will close 'merged' channel upon completion or cancellation.
-		errCh <- mergeEntryChannels(ctx, entries, merged, 1)
+		errCh <- mergeEntryChannelsWithResolver(ctx, entries, merged, 1, func(int) error {
+			return context.Cause(ctx)
+		}, func(group []*metaCacheEntry) (*metaCacheEntry, error) {
+			return z.resolveListingGroup(ctx, bucket, group)
+		})
 	}()
 
 	return nil

cmd/erasure-sets.go

@@ -894,28 +894,23 @@ func (s *erasureSets) ListMultipartUploads(ctx context.Context, bucket, prefix,
 		return ListMultipartsInfo{}, err
 	}
 	defer scan.close()
-	uploads, legacy, err := s.scanMultipartUploads(scan, bucket, 0)
+	uploads, err := s.scanMultipartUploads(scan, bucket, 0)
 	if err != nil {
 		return ListMultipartsInfo{}, err
 	}
-	if legacy {
-		return ListMultipartsInfo{}, errMultipartListingLegacy
-	}
 	return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil
 }
 
-func (s *erasureSets) scanMultipartUploads(scan *multipartScan, bucket string, poolIdx int) ([]MultipartInfo, bool, error) {
+func (s *erasureSets) scanMultipartUploads(scan *multipartScan, bucket string, poolIdx int) ([]MultipartInfo, error) {
 	var uploads []MultipartInfo
-	var keyless bool
 	for i, set := range s.sets {
-		setUploads, setKeyless, err := set.scanMultipartUploads(scan, bucket, poolIdx, i)
+		setUploads, err := set.scanMultipartUploads(scan, bucket, poolIdx, i)
 		if err != nil {
-			return nil, false, err
+			return nil, err
 		}
 		uploads = append(uploads, setUploads...)
-		keyless = keyless || setKeyless
 	}
-	return uploads, keyless, nil
+	return uploads, nil
 }
 
 func (s *erasureSets) listMultipartUploadsExact(ctx context.Context, bucket, object string) (ListMultipartsInfo, error) {

cmd/globals.go

@@ -319,6 +319,8 @@ var (
 
 	globalOperationTimeout       = newDynamicTimeout(10*time.Minute, 5*time.Minute) // default timeout for general ops
 	globalDeleteOperationTimeout = newDynamicTimeout(5*time.Minute, 1*time.Minute)  // default time for delete ops
+	// Listing recovery has its own failure history and a five-second caller budget.
+	globalListingOperationTimeout = newDynamicTimeout(5*time.Second, 5*time.Second)
 
 	globalBucketObjectLockSys *BucketObjectLockSys
 	globalBucketQuotaSys      *BucketQuotaSys

cmd/handler-api.go

@@ -50,7 +50,6 @@ type apiConfig struct {
 	transitionWorkers      int
 
 	staleUploadsExpiry          time.Duration
-	multipartListingStrict      bool
 	staleUploadsCleanupInterval time.Duration
 	deleteCleanupInterval       time.Duration
 	enableODirect               bool
@@ -182,7 +181,6 @@ func (t *apiConfig) init(cfg api.Config, setDriveCounts []int, legacy bool) {
 	t.transitionWorkers = cfg.TransitionWorkers
 
 	t.staleUploadsExpiry = cfg.StaleUploadsExpiry
-	t.multipartListingStrict = cfg.MultipartListing == "strict"
 	t.deleteCleanupInterval = cfg.DeleteCleanupInterval
 	t.enableODirect = cfg.EnableODirect
 	t.gzipObjects = cfg.GzipObjects
@@ -208,12 +206,6 @@ func (t *apiConfig) odirectEnabled() bool {
 	return t.enableODirect
 }
 
-func (t *apiConfig) getMultipartListingLegacy() bool {
-	t.mu.RLock()
-	defer t.mu.RUnlock()
-	return !t.multipartListingStrict
-}
-
 func (t *apiConfig) shouldGzipObjects() bool {
 	t.mu.RLock()
 	defer t.mu.RUnlock()

cmd/list-multipart-uploads-compat_test.go

@@ -19,7 +19,6 @@ package cmd
 
 import (
 	"bytes"
-	"errors"
 	"fmt"
 	"slices"
 	"testing"
@@ -47,7 +46,7 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
 		t.Fatal(err)
 	}
 	z := obj.(*erasureServerPools)
-	setMultipartListingTestMode(t, false)
+
 	t.Cleanup(func() {
 		z.Shutdown(t.Context())
 		removeRoots(dirs)
@@ -182,8 +181,8 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
 	}
 
 	// Simulate an upload written by a pre-upgrade server. Its key cannot be
-	// recovered by scanning the hashed namespace. Strict listing must fail;
+	// recovered by scanning the hashed namespace. Omit only that record, and
-	// the old path is available only through an explicit migration setting.
+	// preserve normal operations when its bucket, key and upload ID are known.
 	legacyObject := objects[1]
 	er := sets.getHashedSet(legacyObject)
 	fi, metadata, err := er.checkUploadIDExists(t.Context(), bucket, legacyObject, uploadIDs[legacyObject], true)
@@ -198,16 +197,25 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
 		er.getUploadIDDir(bucket, legacyObject, uploadIDs[legacyObject]), metadata, fi.WriteQuorum(er.defaultWQuorum())); err != nil {
 		t.Fatal(err)
 	}
-	_, err = z.ListMultipartUploads(t.Context(), bucket, legacyObject, "", "", "", 100)
-	if !errors.Is(err, errMultipartListingLegacy) {
-		t.Fatalf("legacy strict listing: %v", err)
-	}
-	setMultipartListingTestMode(t, true)
 	legacy, err := z.ListMultipartUploads(t.Context(), bucket, legacyObject, "", "", "", 100)
 	if err != nil {
 		t.Fatal(err)
 	}
-	requireMultipartUploadKeys(t, legacy, legacyObject)
+	requireMultipartUploadKeys(t, legacy)
+	remaining, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 100)
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireMultipartUploadKeys(t, remaining, objects[2:]...)
+	part, err = z.PutObjectPart(t.Context(), bucket, legacyObject, uploadIDs[legacyObject], 1,
+		mustGetPutObjReader(t, bytes.NewBufferString("old part"), 8, "", ""), ObjectOptions{})
+	if err != nil {
+		t.Fatalf("continue historical upload: %v", err)
+	}
+	if _, err = z.CompleteMultipartUpload(t.Context(), bucket, legacyObject, uploadIDs[legacyObject],
+		[]CompletePart{{PartNumber: 1, ETag: part.ETag}}, ObjectOptions{}); err != nil {
+		t.Fatalf("complete historical upload: %v", err)
+	}
 }
 
 func TestPaginateMultipartUploads(t *testing.T) {
@@ -260,7 +268,7 @@ func TestPaginateMultipartUploads(t *testing.T) {
 
 func TestListMultipartUploadsGlobalPageAcrossPools(t *testing.T) {
 	z, bucket := consistencyPools(t)
-	setMultipartListingTestMode(t, false)
+
 	objects := []string{"a/one", "b/two", "c/three", "d/four"}
 	for i, object := range objects {
 		if _, err := z.serverPools[i%len(z.serverPools)].NewMultipartUpload(t.Context(), bucket, object, ObjectOptions{}); err != nil {

cmd/local-locker-listing-reverse_test.go

@@ -0,0 +1,412 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// 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 (
+	"context"
+	"fmt"
+	"sync/atomic"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"github.com/minio/minio/internal/dsync"
+)
+
+func listingLockerTestWrite(t *testing.T, l *localLocker, args dsync.LockArgs) {
+	t.Helper()
+	if ok, err := l.Lock(t.Context(), args); !ok || err != nil {
+		t.Fatalf("Lock = %v, %v", ok, err)
+	}
+}
+
+func listingLockerTestUnlock(t *testing.T, l *localLocker, args dsync.LockArgs) {
+	t.Helper()
+	if ok, err := l.Unlock(t.Context(), args); !ok || err != nil {
+		t.Fatalf("Unlock = %v, %v", ok, err)
+	}
+}
+
+func TestLocalLockerListingReverseWaitRechecks(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		change func(*testing.T, *localLocker, dsync.LockArgs)
+		want   bool
+	}{
+		{"writer-released", func(t *testing.T, l *localLocker, w dsync.LockArgs) { listingLockerTestUnlock(t, l, w) }, true},
+		{"writer-still-held", func(*testing.T, *localLocker, dsync.LockArgs) {}, false},
+		{"replacement-writer", func(t *testing.T, l *localLocker, w dsync.LockArgs) {
+			listingLockerTestUnlock(t, l, w)
+			listingLockerTestWrite(t, l, listingLockerTestArgs("new-writer", "key", "writer"))
+		}, false},
+		{"ordinary-reader-arrives", func(t *testing.T, l *localLocker, w dsync.LockArgs) {
+			listingLockerTestUnlock(t, l, w)
+			listingLockerTestRead(t, l, listingLockerTestArgs("ordinary-reader", "key", "ordinary-reader"))
+		}, true},
+		{"listing-reader-arrives", func(t *testing.T, l *localLocker, w dsync.LockArgs) {
+			listingLockerTestUnlock(t, l, w)
+			listingLockerTestRead(t, l, listingLockerTestArgs("other-listing", "key", listingLockerTestSource))
+		}, true},
+		{"force-unlock", func(t *testing.T, l *localLocker, w dsync.LockArgs) {
+			if ok, err := l.ForceUnlock(t.Context(), w); !ok || err != nil {
+				t.Fatalf("ForceUnlock = %v, %v", ok, err)
+			}
+		}, true},
+		{"expiry", func(_ *testing.T, l *localLocker, _ dsync.LockArgs) { l.expireOldLocks(-time.Minute) }, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				l := newLocker()
+				writer := listingLockerTestArgs("writer", "key", "writer")
+				reader := listingLockerTestArgs("listing", "key", listingLockerTestSource)
+				listingLockerTestWrite(t, l, writer)
+				var got bool
+				var err error
+				start := time.Now()
+				go func() { got, err = l.RLock(t.Context(), reader) }()
+				synctest.Wait()
+				if state := listingLockerTestSnapshot(l); state.waiting != 1 {
+					t.Fatalf("waiting = %d, want 1", state.waiting)
+				}
+				listingLockerTestWrite(t, l, listingLockerTestArgs("unrelated", "other-key", "writer"))
+				if elapsed := time.Since(start); elapsed != 0 {
+					t.Fatalf("unrelated key waited %v", elapsed)
+				}
+				time.Sleep(time.Millisecond)
+				tc.change(t, l, writer)
+				time.Sleep(time.Millisecond)
+				synctest.Wait()
+				if state := listingLockerTestSnapshot(l); got != tc.want || err != nil || state.waiting != 0 {
+					t.Fatalf("RLock = %v, %v, waiting = %d; want %v, nil, 0", got, err, state.waiting, tc.want)
+				}
+				if elapsed := time.Since(start); elapsed != 2*time.Millisecond {
+					t.Fatalf("one attempt elapsed %v, want 2ms", elapsed)
+				}
+				found := false
+				for _, holder := range l.DupLockMap()["key"] {
+					if holder.UID == reader.UID {
+						found = true
+						if holder.Writer || holder.Timestamp != UTCNow().UnixNano() || holder.TimeLastRefresh != holder.Timestamp {
+							t.Fatalf("invalid granted reader: %+v", holder)
+						}
+					}
+				}
+				if found != got {
+					t.Fatalf("reader present = %v, granted = %v", found, got)
+				}
+			})
+		})
+	}
+}
+
+func TestLocalLockerListingReverseWaitCancellation(t *testing.T) {
+	for _, when := range []string{"before-call", "during-wait", "deadline"} {
+		t.Run(when, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				l := newLocker()
+				writer := listingLockerTestArgs("writer", "key", "writer")
+				listingLockerTestWrite(t, l, writer)
+				ctx, cancel := context.WithCancel(t.Context())
+				if when == "deadline" {
+					cancel()
+					ctx, cancel = context.WithTimeout(t.Context(), time.Millisecond)
+				}
+				defer cancel()
+				if when == "before-call" {
+					cancel()
+				}
+				var got bool
+				var err error
+				start := time.Now()
+				go func() { got, err = l.RLock(ctx, listingLockerTestArgs("listing", "key", listingLockerTestSource)) }()
+				synctest.Wait()
+				if when != "before-call" {
+					if state := listingLockerTestSnapshot(l); state.waiting != 1 {
+						t.Fatalf("waiting = %d, want 1", state.waiting)
+					}
+					time.Sleep(time.Millisecond)
+					// Removal must not turn the canceled request into a grant.
+					listingLockerTestUnlock(t, l, writer)
+					cancel()
+				} else {
+					listingLockerTestUnlock(t, l, writer)
+				}
+				synctest.Wait()
+				state := listingLockerTestSnapshot(l)
+				if got || err == nil || err != ctx.Err() || state != (listingLockerTestState{}) {
+					t.Fatalf("canceled RLock = %v, %v; context %v; state %+v", got, err, ctx.Err(), state)
+				}
+				if elapsed := time.Since(start); elapsed >= 2*time.Millisecond {
+					t.Fatalf("canceled request waited for full timer: %v", elapsed)
+				}
+			})
+		})
+	}
+}
+
+func TestLocalLockerListingReverseWaitExcluded(t *testing.T) {
+	for _, source := range []string{"ordinary-reader", "", "[<unknown>:0:()]", "listingRLock", "[metacache-server-pool.go:0:(*erasureServerPools).listingRLock()]"} {
+		t.Run(source, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				l := newLocker()
+				listingLockerTestWrite(t, l, listingLockerTestArgs("writer", "key", "writer"))
+				start := time.Now()
+				if ok, err := l.RLock(t.Context(), listingLockerTestArgs("reader", "key", source)); ok || err != nil {
+					t.Fatalf("excluded RLock = %v, %v", ok, err)
+				}
+				if state := listingLockerTestSnapshot(l); time.Since(start) != 0 || state.waiting != 0 || state.uids != 1 {
+					t.Fatalf("excluded RLock changed timing or state: %+v", state)
+				}
+			})
+		})
+	}
+}
+
+func TestLocalLockerListingReverseWaitFastPaths(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		l := newLocker()
+		start := time.Now()
+		for i, source := range []string{listingLockerTestSource, "ordinary-reader", ""} {
+			listingLockerTestRead(t, l, listingLockerTestArgs(fmt.Sprintf("shared-%d", i), "shared", source))
+			listingLockerTestRead(t, l, listingLockerTestArgs(fmt.Sprintf("free-%d", i), fmt.Sprintf("free-%d", i), source))
+		}
+		if state := listingLockerTestSnapshot(l); time.Since(start) != 0 || state.waiting != 0 || state.uids != 6 {
+			t.Fatalf("uncontended readers changed timing or state: %+v", state)
+		}
+	})
+}
+
+func TestLocalLockerListingReverseWaitOverloaded(t *testing.T) {
+	for _, when := range []string{"before-call", "during-wait"} {
+		t.Run(when, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				l := newLocker()
+				writer := listingLockerTestArgs("writer", "key", "writer")
+				listingLockerTestWrite(t, l, writer)
+				defer l.waitMutex.Store(0)
+				if when == "before-call" {
+					l.waitMutex.Store(lockMutexWaitLimit + 1)
+				}
+				var got bool
+				var err error
+				go func() {
+					got, err = l.RLock(t.Context(), listingLockerTestArgs("listing", "key", listingLockerTestSource))
+				}()
+				synctest.Wait()
+				if when == "during-wait" {
+					if state := listingLockerTestSnapshot(l); state.waiting != 1 {
+						t.Fatalf("waiting = %d, want 1", state.waiting)
+					}
+					listingLockerTestUnlock(t, l, writer)
+					l.waitMutex.Store(lockMutexWaitLimit + 1)
+					time.Sleep(2 * time.Millisecond)
+					synctest.Wait()
+				} else {
+					listingLockerTestUnlock(t, l, writer)
+				}
+				if state := listingLockerTestSnapshot(l); got || err != nil || state != (listingLockerTestState{}) || l.locksOverloaded.Load() != 1 {
+					t.Fatalf("overloaded RLock = %v, %v; state %+v; overloads %d", got, err, state, l.locksOverloaded.Load())
+				}
+			})
+		})
+	}
+}
+
+func TestLocalLockerListingWaitSharedAdmission(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		l := newLocker()
+		listingLockerTestRead(t, l, listingLockerTestArgs("reader", "read-key", listingLockerTestSource))
+		listingLockerTestWrite(t, l, listingLockerTestArgs("writer", "write-key", "writer"))
+		ctx, cancel := context.WithCancel(t.Context())
+		defer cancel()
+		results := make(chan error, listingLockWaitLimit)
+		for i := range listingLockWaitLimit {
+			go func() {
+				var ok bool
+				var err error
+				if i%2 == 0 {
+					ok, err = l.Lock(ctx, listingLockerTestArgs(fmt.Sprintf("writer-%d", i), "read-key", "writer"))
+				} else {
+					ok, err = l.RLock(ctx, listingLockerTestArgs(fmt.Sprintf("listing-%d", i), "write-key", listingLockerTestSource))
+				}
+				if ok {
+					err = fmt.Errorf("granted over a conflicting holder")
+				}
+				results <- err
+			}()
+		}
+		synctest.Wait()
+		if state := listingLockerTestSnapshot(l); state.waiting != listingLockWaitLimit {
+			t.Fatalf("combined waiting = %d, want %d", state.waiting, listingLockWaitLimit)
+		}
+		start := time.Now()
+		if ok, err := l.Lock(ctx, listingLockerTestArgs("excess-writer", "read-key", "writer")); ok || err != nil {
+			t.Fatalf("excess Lock = %v, %v", ok, err)
+		}
+		if ok, err := l.RLock(ctx, listingLockerTestArgs("excess-listing", "write-key", listingLockerTestSource)); ok || err != nil {
+			t.Fatalf("excess RLock = %v, %v", ok, err)
+		}
+		listingLockerTestWrite(t, l, listingLockerTestArgs("unrelated-writer", "unrelated-write", "writer"))
+		listingLockerTestRead(t, l, listingLockerTestArgs("unrelated-listing", "unrelated-read", listingLockerTestSource))
+		if state := listingLockerTestSnapshot(l); time.Since(start) != 0 || state.waiting != listingLockWaitLimit || state.uids != 4 {
+			t.Fatalf("combined cap changed immediate paths: %+v", state)
+		}
+		cancel()
+		synctest.Wait()
+		for range listingLockWaitLimit {
+			if err := <-results; err != context.Canceled {
+				t.Fatalf("canceled waiter: %v", err)
+			}
+		}
+		if state := listingLockerTestSnapshot(l); state.waiting != 0 || state.uids != 4 {
+			t.Fatalf("combined wait leaked: %+v", state)
+		}
+	})
+}
+
+type listingReverseCountLocker struct {
+	*localLocker
+	calls  atomic.Int32
+	grants atomic.Int32
+	reject bool
+}
+
+func (l *listingReverseCountLocker) RLock(ctx context.Context, args dsync.LockArgs) (bool, error) {
+	l.calls.Add(1)
+	if l.reject {
+		return false, nil
+	}
+	ok, err := l.localLocker.RLock(ctx, args)
+	if ok {
+		l.grants.Add(1)
+	}
+	return ok, err
+}
+
+func TestDistributedListingReverseWait(t *testing.T) {
+	for _, tc := range []struct {
+		name             string
+		free, held       int
+		immediateRejects int
+		cancel           bool
+		want             bool
+	}{
+		{name: "all-writers-release", want: true},
+		{name: "two-writers-remain", held: 2, want: true},
+		{name: "three-writers-remain", held: 3},
+		{name: "partial-grant-then-quorum", free: 1, want: true},
+		{name: "partial-grant-without-quorum", free: 1, held: 3},
+		{name: "late-grant-after-rejection", immediateRejects: 3},
+		{name: "partial-and-late-grants-after-cancel", free: 1, cancel: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				clients := make([]dsync.NetLocker, 4)
+				writer := listingLockerTestArgs("writer", "key", "writer")
+				for i := range clients {
+					l := &listingReverseCountLocker{localLocker: newLocker(), reject: i < tc.immediateRejects}
+					clients[i] = l
+					if i >= tc.free {
+						listingLockerTestWrite(t, l.localLocker, writer)
+					}
+				}
+				ds := &dsync.Dsync{GetLockers: func() ([]dsync.NetLocker, string) { return clients, writer.Owner }, Timeouts: dsync.DefaultTimeouts}
+				dm := dsync.NewDRWMutex(ds, "key")
+				ctx, cancel := context.WithCancel(t.Context())
+				defer cancel()
+				result := make(chan bool, 1)
+				go func() {
+					result <- dm.GetRLock(ctx, nil, "listing", listingLockerTestSource, dsync.Options{Timeout: time.Second, RetryInterval: -1})
+				}()
+				synctest.Wait()
+				for i, c := range clients {
+					l := c.(*listingReverseCountLocker)
+					if i < tc.free && l.grants.Load() != 1 {
+						t.Fatal("uncontended partial grant did not arrive immediately")
+					}
+					if i >= tc.free && !l.reject && listingLockerTestSnapshot(l.localLocker).waiting != 1 {
+						t.Fatal("conflicting reader did not wait in its first RPC")
+					}
+				}
+				earlyResult := tc.immediateRejects == 3 || tc.cancel
+				if tc.cancel {
+					cancel()
+					synctest.Wait()
+				}
+				if earlyResult {
+					select {
+					case got := <-result:
+						if got {
+							t.Fatal("attempt succeeded before delayed replies")
+						}
+					default:
+						t.Fatal("attempt did not return before delayed replies")
+					}
+					for i := range tc.free {
+						if state := listingLockerTestSnapshot(clients[i].(*listingReverseCountLocker).localLocker); state.uids != 0 {
+							t.Fatal("early partial grant was not released")
+						}
+					}
+				}
+				time.Sleep(time.Millisecond)
+				for i, c := range clients {
+					if i >= tc.free+tc.held {
+						listingLockerTestUnlock(t, c.(*listingReverseCountLocker).localLocker, writer)
+					}
+				}
+				time.Sleep(time.Millisecond)
+				synctest.Wait()
+				if !earlyResult {
+					if got := <-result; got != tc.want {
+						t.Fatalf("GetRLock = %v, want %v", got, tc.want)
+					}
+				}
+				for i, c := range clients {
+					l := c.(*listingReverseCountLocker)
+					if l.calls.Load() != 1 || listingLockerTestSnapshot(l.localLocker).waiting != 0 {
+						t.Fatalf("peer %d calls = %d or leaked admission", i, l.calls.Load())
+					}
+					wantGrant := !l.reject && (i < tc.free || i >= tc.free+tc.held)
+					if (l.grants.Load() == 1) != wantGrant {
+						t.Fatalf("peer %d grants = %d, want granted = %v", i, l.grants.Load(), wantGrant)
+					}
+					if !tc.want {
+						for _, holder := range l.DupLockMap()["key"] {
+							if holder.UID == "listing" {
+								t.Fatalf("peer %d leaked partial or late read grant", i)
+							}
+						}
+					}
+				}
+				if tc.want {
+					dm.RUnlock(t.Context())
+					synctest.Wait()
+				}
+				for i, c := range clients {
+					l := c.(*listingReverseCountLocker)
+					if i >= tc.free && i < tc.free+tc.held {
+						listingLockerTestUnlock(t, l.localLocker, writer)
+					}
+					if state := listingLockerTestSnapshot(l.localLocker); state != (listingLockerTestState{}) {
+						t.Fatalf("peer %d final state is not empty: %+v", i, state)
+					}
+				}
+			})
+		})
+	}
+}

cmd/local-locker-listing_test.go

@@ -0,0 +1,432 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// 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 (
+	"context"
+	"fmt"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"github.com/minio/minio/internal/dsync"
+)
+
+const listingLockerTestSource = "[metacache-server-pool.go:274:(*erasureServerPools).listingRLock()]"
+
+func listingLockerTestArgs(uid, resource, source string) dsync.LockArgs {
+	quorum := 3
+	return dsync.LockArgs{UID: uid, Resources: []string{resource}, Source: source, Owner: "listing-wait-test", Quorum: &quorum}
+}
+
+func listingLockerTestRead(t *testing.T, l *localLocker, args dsync.LockArgs) {
+	t.Helper()
+	if ok, err := l.RLock(t.Context(), args); !ok || err != nil {
+		t.Fatalf("RLock = %v, %v", ok, err)
+	}
+}
+
+func listingLockerTestRUnlock(t *testing.T, l *localLocker, args dsync.LockArgs) {
+	t.Helper()
+	if ok, err := l.RUnlock(t.Context(), args); !ok || err != nil {
+		t.Fatalf("RUnlock = %v, %v", ok, err)
+	}
+}
+
+type listingLockerTestState struct {
+	waiting, resources, uids int
+}
+
+func listingLockerTestSnapshot(l *localLocker) listingLockerTestState {
+	defer l.getMutex()()
+	return listingLockerTestState{waiting: l.listingWaiters, resources: len(l.lockMap), uids: len(l.lockUID)}
+}
+
+func TestLocalLockerListingSource(t *testing.T) {
+	for _, source := range []string{
+		"", "listingRLock", "[<unknown>:0:()]",
+		strings.Replace(listingLockerTestSource, ":274:", ":", 1),
+		strings.Replace(listingLockerTestSource, "274", "", 1),
+		strings.Replace(listingLockerTestSource, "274", "0", 1),
+		strings.Replace(listingLockerTestSource, "274", "1x", 1),
+		strings.Replace(listingLockerTestSource, "274", "-1", 1),
+		strings.Replace(listingLockerTestSource, "listingRLock", "GetObjectNInfo", 1),
+		strings.Replace(listingLockerTestSource, "listingRLock", "listingRLock.func1", 1),
+		strings.Replace(listingLockerTestSource, "metacache-server-pool.go", "other.go", 1),
+		listingLockerTestSource + "trailing",
+	} {
+		if isListingReadLock([]lockRequesterInfo{{Source: source}}) {
+			t.Errorf("unrecognized source accepted: %q", source)
+		}
+	}
+	if isListingReadLock(nil) || isListingReadLock([]lockRequesterInfo{{Source: listingLockerTestSource, Writer: true}}) {
+		t.Fatal("empty or write holders accepted")
+	}
+	if isListingReadLock([]lockRequesterInfo{{Source: listingLockerTestSource}, {Source: "ordinary-reader"}}) {
+		t.Fatal("mixed readers accepted")
+	}
+
+	// Check the real runtime-generated source, not only a hand-written label.
+	synctest.Test(t, func(t *testing.T) {
+		lockers := []dsync.NetLocker{newLocker(), newLocker(), newLocker(), newLocker()}
+		set := &erasureObjects{nsMutex: newNSLock(true), getLockers: func() ([]dsync.NetLocker, string) { return lockers, "listing-source-test" }}
+		z := &erasureServerPools{serverPools: []*erasureSets{{sets: []*erasureObjects{set}}, {sets: []*erasureObjects{set}}}}
+		_, unlock, err := z.listingRLock(t.Context(), "bucket", "object", set)
+		if err != nil {
+			t.Fatal(err)
+		}
+		for _, c := range lockers {
+			l := c.(*localLocker)
+			for _, holders := range l.DupLockMap() {
+				if !isListingReadLock(holders) {
+					t.Errorf("actual listingRLock source was not recognized: %#v", holders)
+				}
+			}
+		}
+		unlock()
+		synctest.Wait()
+	})
+}
+
+func FuzzListingLockSource(f *testing.F) {
+	f.Add(listingLockerTestSource)
+	f.Add("[metacache-server-pool.go:(*erasureServerPools).listingRLock()]")
+	f.Add("")
+	f.Fuzz(func(t *testing.T, source string) {
+		// Source is diagnostic text from a peer, including older versions. No
+		// malformed string may panic or make a writer eligible for this wait.
+		isListingReadLock([]lockRequesterInfo{{Source: source}})
+		if isListingReadLock([]lockRequesterInfo{{Source: source, Writer: true}}) {
+			t.Fatal("writer was accepted as a listing reader")
+		}
+	})
+}
+
+func TestLocalLockerListingWaitRechecks(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		change func(*testing.T, *localLocker, dsync.LockArgs)
+		want   bool
+	}{
+		{"reader-released", func(t *testing.T, l *localLocker, r dsync.LockArgs) { listingLockerTestRUnlock(t, l, r) }, true},
+		{"reader-still-held", func(*testing.T, *localLocker, dsync.LockArgs) {}, false},
+		{"one-of-two-readers-remains", func(t *testing.T, l *localLocker, r dsync.LockArgs) {
+			other := r
+			other.UID = "second-reader"
+			listingLockerTestRead(t, l, other)
+			listingLockerTestRUnlock(t, l, r)
+		}, false},
+		{"ordinary-reader-arrives", func(t *testing.T, l *localLocker, r dsync.LockArgs) {
+			listingLockerTestRUnlock(t, l, r)
+			listingLockerTestRead(t, l, listingLockerTestArgs("new-reader", "key", "ordinary-reader"))
+		}, false},
+		{"other-writer-wins", func(t *testing.T, l *localLocker, r dsync.LockArgs) {
+			listingLockerTestRUnlock(t, l, r)
+			if ok, err := l.Lock(t.Context(), listingLockerTestArgs("other-writer", "key", "writer")); !ok || err != nil {
+				t.Fatalf("competing Lock = %v, %v", ok, err)
+			}
+		}, false},
+		{"force-unlock", func(t *testing.T, l *localLocker, r dsync.LockArgs) {
+			if ok, err := l.ForceUnlock(t.Context(), r); !ok || err != nil {
+				t.Fatalf("ForceUnlock = %v, %v", ok, err)
+			}
+		}, true},
+		{"expiry", func(_ *testing.T, l *localLocker, _ dsync.LockArgs) { l.expireOldLocks(-time.Minute) }, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				l := newLocker()
+				reader := listingLockerTestArgs("reader", "key", listingLockerTestSource)
+				listingLockerTestRead(t, l, reader)
+				var got bool
+				var err error
+				start := time.Now()
+				go func() { got, err = l.Lock(t.Context(), listingLockerTestArgs("writer", "key", "writer")) }()
+				synctest.Wait()
+				if state := listingLockerTestSnapshot(l); state.waiting != 1 {
+					t.Fatalf("waiting = %d, want 1", state.waiting)
+				}
+				// The global mutex must be released throughout the timer wait.
+				if ok, err := l.Lock(t.Context(), listingLockerTestArgs("unrelated", "other-key", "writer")); !ok || err != nil {
+					t.Fatalf("unrelated key blocked: %v, %v", ok, err)
+				}
+				if elapsed := time.Since(start); elapsed != 0 {
+					t.Fatalf("unrelated key waited %v", elapsed)
+				}
+				time.Sleep(time.Millisecond)
+				tc.change(t, l, reader)
+				time.Sleep(time.Millisecond)
+				synctest.Wait()
+				if state := listingLockerTestSnapshot(l); got != tc.want || err != nil || state.waiting != 0 {
+					t.Fatalf("Lock = %v, %v, waiting = %d; want %v, nil, 0", got, err, state.waiting, tc.want)
+				}
+				if elapsed := time.Since(start); elapsed != 2*time.Millisecond {
+					t.Fatalf("one attempt elapsed %v, want 2ms", elapsed)
+				}
+			})
+		})
+	}
+}
+
+func TestLocalLockerListingWaitCancellation(t *testing.T) {
+	for _, when := range []string{"before-call", "during-wait", "deadline"} {
+		t.Run(when, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				l := newLocker()
+				reader := listingLockerTestArgs("reader", "key", listingLockerTestSource)
+				listingLockerTestRead(t, l, reader)
+				ctx, cancel := context.WithCancel(t.Context())
+				if when == "deadline" {
+					cancel()
+					ctx, cancel = context.WithTimeout(t.Context(), time.Millisecond)
+				}
+				defer cancel()
+				if when == "before-call" {
+					cancel()
+				}
+				var got bool
+				var err error
+				go func() { got, err = l.Lock(ctx, listingLockerTestArgs("writer", "key", "writer")) }()
+				synctest.Wait()
+				if when != "before-call" {
+					if state := listingLockerTestSnapshot(l); state.waiting != 1 {
+						t.Fatalf("waiting = %d", state.waiting)
+					}
+					time.Sleep(time.Millisecond)
+					cancel()
+				}
+				synctest.Wait()
+				if state := listingLockerTestSnapshot(l); got || err != ctx.Err() || err == nil || state.waiting != 0 {
+					t.Fatalf("canceled Lock = %v, %v; context %v; waiting = %d", got, err, ctx.Err(), state.waiting)
+				}
+				listingLockerTestRUnlock(t, l, reader)
+				if state := listingLockerTestSnapshot(l); state.resources != 0 || state.uids != 0 {
+					t.Fatal("canceled wait left a lock")
+				}
+			})
+		})
+	}
+}
+
+func TestLocalLockerListingWaitExcluded(t *testing.T) {
+	for _, kind := range []string{"ordinary-reader", "unknown-reader", "mixed-readers", "writer", "bulk", "overloaded"} {
+		t.Run(kind, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				l := newLocker()
+				reader := listingLockerTestArgs("holder", "key", listingLockerTestSource)
+				if kind == "ordinary-reader" {
+					reader.Source = "[erasure-object.go:100:(*erasureObjects).GetObjectNInfo()]"
+				}
+				if kind == "unknown-reader" {
+					reader.Source = ""
+				}
+				if kind == "writer" {
+					if ok, err := l.Lock(t.Context(), reader); !ok || err != nil {
+						t.Fatalf("initial write = %v, %v", ok, err)
+					}
+				} else {
+					listingLockerTestRead(t, l, reader)
+				}
+				if kind == "mixed-readers" {
+					listingLockerTestRead(t, l, listingLockerTestArgs("ordinary", "key", "ordinary-reader"))
+				}
+				writer := listingLockerTestArgs("writer", "key", "writer")
+				if kind == "bulk" {
+					writer.Resources = []string{"key", "other-key"}
+				}
+				if kind == "overloaded" {
+					l.waitMutex.Store(lockMutexWaitLimit + 1)
+				}
+				start := time.Now()
+				if ok, err := l.Lock(t.Context(), writer); ok || err != nil {
+					t.Fatalf("conflicting Lock = %v, %v", ok, err)
+				}
+				state := listingLockerTestSnapshot(l)
+				if elapsed := time.Since(start); elapsed != 0 || state.waiting != 0 {
+					t.Fatalf("excluded path waited %v; waiters = %d", elapsed, state.waiting)
+				}
+				if _, ok := l.DupLockMap()["other-key"]; ok {
+					t.Fatal("bulk request partially acquired an unrelated key")
+				}
+			})
+		})
+	}
+}
+
+func TestLocalLockerListingWaitAdmission(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		l := newLocker()
+		reader := listingLockerTestArgs("reader", "key", listingLockerTestSource)
+		listingLockerTestRead(t, l, reader)
+		ctx, cancel := context.WithCancel(t.Context())
+		defer cancel()
+		results := make(chan error, listingLockWaitLimit)
+		for i := range listingLockWaitLimit {
+			go func() {
+				ok, err := l.Lock(ctx, listingLockerTestArgs(fmt.Sprintf("writer-%d", i), "key", "writer"))
+				if ok {
+					err = fmt.Errorf("granted over a live reader")
+				}
+				results <- err
+			}()
+		}
+		synctest.Wait()
+		if state := listingLockerTestSnapshot(l); state.waiting != listingLockWaitLimit {
+			t.Fatalf("waiting = %d, want %d", state.waiting, listingLockWaitLimit)
+		}
+		start := time.Now()
+		if ok, err := l.Lock(ctx, listingLockerTestArgs("excess", "key", "writer")); ok || err != nil {
+			t.Fatalf("excess waiter = %v, %v", ok, err)
+		}
+		if state := listingLockerTestSnapshot(l); time.Since(start) != 0 || state.waiting != listingLockWaitLimit {
+			t.Fatal("excess waiter consumed time or admission")
+		}
+		if ok, err := l.Lock(ctx, listingLockerTestArgs("unrelated", "other-key", "writer")); !ok || err != nil {
+			t.Fatalf("admission limit rejected an uncontended key: %v, %v", ok, err)
+		}
+		cancel()
+		synctest.Wait()
+		for range listingLockWaitLimit {
+			if err := <-results; err != context.Canceled {
+				t.Fatalf("canceled waiter: %v", err)
+			}
+		}
+		if state := listingLockerTestSnapshot(l); state.waiting != 0 {
+			t.Fatalf("admission leaked: %d", state.waiting)
+		}
+	})
+}
+
+func TestLocalLockerListingWaitContended(t *testing.T) {
+	for _, n := range []int{2, 8, 32} {
+		t.Run(fmt.Sprintf("%d-writers", n), func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				l := newLocker()
+				reader := listingLockerTestArgs("reader", "key", listingLockerTestSource)
+				listingLockerTestRead(t, l, reader)
+				results := make(chan bool, n)
+				for i := range n {
+					go func() {
+						ok, err := l.Lock(t.Context(), listingLockerTestArgs(fmt.Sprintf("writer-%d", i), "key", "writer"))
+						if err != nil {
+							t.Errorf("Lock: %v", err)
+						}
+						results <- ok
+					}()
+				}
+				synctest.Wait()
+				if state := listingLockerTestSnapshot(l); state.waiting != n {
+					t.Fatalf("waiting = %d, want %d", state.waiting, n)
+				}
+				time.Sleep(time.Millisecond)
+				listingLockerTestRUnlock(t, l, reader)
+				time.Sleep(time.Millisecond)
+				synctest.Wait()
+				granted := 0
+				for range n {
+					if <-results {
+						granted++
+					}
+				}
+				if state := listingLockerTestSnapshot(l); granted != 1 || state.uids != 1 || state.waiting != 0 {
+					t.Fatalf("grants = %d, indexed = %d, waiting = %d", granted, state.uids, state.waiting)
+				}
+			})
+		})
+	}
+}
+
+type listingCountLocker struct {
+	*localLocker
+	calls atomic.Int32
+}
+
+func (l *listingCountLocker) Lock(ctx context.Context, args dsync.LockArgs) (bool, error) {
+	l.calls.Add(1)
+	return l.localLocker.Lock(ctx, args)
+}
+
+// Exercise real DRWMutex quorum and late-response cleanup. The mixed case with
+// two immediate rejections fails before the upgraded lockers finish waiting.
+func TestDistributedListingWait(t *testing.T) {
+	for _, ordinaryReaders := range []int{0, 1, 2} {
+		t.Run(fmt.Sprintf("%d-immediate-rejections", ordinaryReaders), func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				clients := make([]dsync.NetLocker, 4)
+				reader := listingLockerTestArgs("reader", "key", listingLockerTestSource)
+				for i := range clients {
+					l := &listingCountLocker{localLocker: newLocker()}
+					clients[i] = l
+					r := reader
+					if i < ordinaryReaders {
+						r.Source = "old-node-unknown-source"
+					}
+					listingLockerTestRead(t, l.localLocker, r)
+				}
+				ds := &dsync.Dsync{GetLockers: func() ([]dsync.NetLocker, string) { return clients, reader.Owner }, Timeouts: dsync.DefaultTimeouts}
+				dm := dsync.NewDRWMutex(ds, "key")
+				var got bool
+				go func() {
+					got = dm.GetLock(t.Context(), nil, "writer", "writer", dsync.Options{Timeout: time.Second, RetryInterval: -1})
+				}()
+				synctest.Wait()
+				time.Sleep(time.Millisecond)
+				for _, c := range clients {
+					listingLockerTestRUnlock(t, c.(*listingCountLocker).localLocker, reader)
+				}
+				time.Sleep(time.Millisecond)
+				synctest.Wait()
+				if got != (ordinaryReaders < 2) {
+					t.Fatalf("GetLock = %v", got)
+				}
+				granted := 0
+				for _, c := range clients {
+					l := c.(*listingCountLocker)
+					if l.calls.Load() != 1 {
+						t.Fatalf("lock calls = %d, want one round", l.calls.Load())
+					}
+					state := listingLockerTestSnapshot(l.localLocker)
+					if state.resources != 0 {
+						granted++
+					}
+					if state.waiting != 0 {
+						t.Fatal("wait admission leaked")
+					}
+				}
+				if got && granted < 3 {
+					t.Fatalf("writer accepted without quorum: %d", granted)
+				}
+				if !got && granted != 0 {
+					t.Fatalf("late grants leaked after rejected attempt: %d", granted)
+				}
+				if got {
+					dm.Unlock(t.Context())
+					synctest.Wait()
+				}
+				for _, c := range clients {
+					l := c.(*listingCountLocker)
+					if state := listingLockerTestSnapshot(l.localLocker); state.resources != 0 || state.uids != 0 {
+						t.Fatal("final lock state not empty")
+					}
+				}
+			})
+		})
+	}
+}

cmd/local-locker.go

@@ -23,6 +23,7 @@ import (
 	"context"
 	"fmt"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -38,6 +39,14 @@ import (
 // Rejected clients are expected to retry.
 const lockMutexWaitLimit = 1000
 
+// A short listing recovery read or conflicting writer may finish before the
+// rejected request retries. Recheck once in the same request, sharing a bound
+// for timers outside the mutex. This is not a queue and gives neither priority.
+const (
+	listingLockWait      = 2 * time.Millisecond
+	listingLockWaitLimit = 64
+)
+
 // lockRequesterInfo stores various info from the client for each lock that is requested.
 type lockRequesterInfo struct {
 	Name            string // name of the resource lock was requested for
@@ -65,6 +74,8 @@ type localLocker struct {
 	waitMutex atomic.Int32
 	lockMap   map[string][]lockRequesterInfo
 	lockUID   map[string]string // UUID -> resource map.
+	// Protected by mutex. These waiters are not included in waitMutex.
+	listingWaiters int
 
 	// the following are updated on every cleanup defined in lockValidityDuration
 	readers         atomic.Int32
@@ -96,6 +107,32 @@ func (l *localLocker) canTakeLock(resources ...string) bool {
 	return true
 }
 
+// isListingReadLock recognizes only the existing diagnostic source emitted by
+// listingRLock. Unknown or changed source formats retain immediate rejection;
+// Source is not a new wire contract or proof that the read will finish quickly.
+func isListingReadLock(holders []lockRequesterInfo) bool {
+	if len(holders) == 0 {
+		return false
+	}
+	const prefix = "[metacache-server-pool.go:"
+	const suffix = ":(*erasureServerPools).listingRLock()]"
+	for _, holder := range holders {
+		if holder.Writer || len(holder.Source) <= len(prefix)+len(suffix) || !strings.HasPrefix(holder.Source, prefix) || !strings.HasSuffix(holder.Source, suffix) {
+			return false
+		}
+		line := holder.Source[len(prefix) : len(holder.Source)-len(suffix)]
+		if len(line) == 0 || line[0] == '0' {
+			return false
+		}
+		for _, c := range line {
+			if c < '0' || c > '9' {
+				return false
+			}
+		}
+	}
+	return true
+}
+
 func (l *localLocker) Lock(ctx context.Context, args dsync.LockArgs) (reply bool, err error) {
 	if len(args.Resources) > maxDeleteList {
 		return false, fmt.Errorf("internal error: localLocker.Lock called with more than %d resources", maxDeleteList)
@@ -112,9 +149,33 @@ func (l *localLocker) Lock(ctx context.Context, args dsync.LockArgs) (reply bool
 		return false, ctx.Err()
 	}
 	if !l.canTakeLock(args.Resources...) {
-		// Not all locks can be taken on resources,
+		// Only a single-resource write blocked exclusively by listing recovery
-		// reject it completely.
+		// readers gets a short, bounded opportunity to succeed in this attempt.
-		return false, nil
+		if len(args.Resources) != 1 || l.listingWaiters >= listingLockWaitLimit || !isListingReadLock(l.lockMap[args.Resources[0]]) {
+			return false, nil
+		}
+		l.listingWaiters++
+		l.mutex.Unlock()
+		timer := time.NewTimer(listingLockWait)
+		select {
+		case <-ctx.Done():
+		case <-timer.C:
+		}
+		timer.Stop()
+		// Restore the mutex before returning to the existing deferred unlock.
+		// Readers or another writer may have entered while it was released.
+		l.getMutex()
+		l.listingWaiters--
+		if ctx.Err() != nil {
+			return false, ctx.Err()
+		}
+		if l.waitMutex.Load() > lockMutexWaitLimit {
+			l.locksOverloaded.Add(1)
+			return false, nil
+		}
+		if !l.canTakeLock(args.Resources...) {
+			return false, nil
+		}
 	}
 
 	// No locks held on the all resources, so claim write
@@ -206,6 +267,31 @@ func (l *localLocker) RLock(ctx context.Context, args dsync.LockArgs) (reply boo
 		return false, ctx.Err()
 	}
 	resource := args.Resources[0]
+	lri, ok := l.lockMap[resource]
+	if ok && isWriteLock(lri) && l.listingWaiters < listingLockWaitLimit && isListingReadLock([]lockRequesterInfo{{Source: args.Source}}) {
+		// Only a positively identified listing read blocked by a writer may
+		// wait once. A long or replacement writer still rejects normally.
+		l.listingWaiters++
+		l.mutex.Unlock()
+		timer := time.NewTimer(listingLockWait)
+		select {
+		case <-ctx.Done():
+		case <-timer.C:
+		}
+		timer.Stop()
+		// Restore the mutex before the existing deferred unlock, then use
+		// the current holders in the normal read-lock grant path below.
+		l.getMutex()
+		l.listingWaiters--
+		if ctx.Err() != nil {
+			return false, ctx.Err()
+		}
+		if l.waitMutex.Load() > lockMutexWaitLimit {
+			l.locksOverloaded.Add(1)
+			return false, nil
+		}
+		lri, ok = l.lockMap[resource]
+	}
 	now := UTCNow()
 	lrInfo := lockRequesterInfo{
 		Name:            resource,
@@ -217,7 +303,6 @@ func (l *localLocker) RLock(ctx context.Context, args dsync.LockArgs) (reply boo
 		TimeLastRefresh: now.UnixNano(),
 		Quorum:          *args.Quorum,
 	}
-	lri, ok := l.lockMap[resource]
 	if ok {
 		if reply = !isWriteLock(lri); reply {
 			// Unless there is a write lock

cmd/metacache-entries.go

@@ -37,6 +37,10 @@ type metaCacheEntry struct {
 	// Entries without metadata will only be present in non-recursive scans.
 	metadata []byte
 
+	// listingSet marks an eligible unresolved walk entry. It is consumed by the
+	// pool merger before any cache serialization or downstream filtering.
+	listingSet *erasureObjects
+
 	// cached contains the metadata if decoded.
 	cached *xlMetaV2
 
@@ -46,17 +50,17 @@ type metaCacheEntry struct {
 
 // isDir returns if the entry is representing a prefix directory.
 func (e metaCacheEntry) isDir() bool {
-	return len(e.metadata) == 0 && strings.HasSuffix(e.name, slashSeparator)
+	return e.listingSet == nil && len(e.metadata) == 0 && strings.HasSuffix(e.name, slashSeparator)
 }
 
 // isObject returns if the entry is representing an object.
 func (e metaCacheEntry) isObject() bool {
-	return len(e.metadata) > 0
+	return e.listingSet != nil || len(e.metadata) > 0
 }
 
 // isObjectDir returns if the entry is representing an object/
 func (e metaCacheEntry) isObjectDir() bool {
-	return len(e.metadata) > 0 && strings.HasSuffix(e.name, slashSeparator)
+	return (e.listingSet != nil || len(e.metadata) > 0) && strings.HasSuffix(e.name, slashSeparator)
 }
 
 // hasPrefix returns whether an entry has a specific prefix
@@ -670,26 +674,82 @@ func (m *metaCacheEntriesSorted) forwardPast(s string) {
 // If the context is canceled the function will return the error,
 // otherwise the function will return nil.
 func mergeEntryChannels(ctx context.Context, in []chan metaCacheEntry, out chan<- metaCacheEntry, readQuorum int) error {
+	return mergeEntryChannelsWithErrors(ctx, in, out, readQuorum, nil)
+}
+
+// terminalErr is inspected after each input closes and before the merger
+// advances beyond it. Callers must publish the error before closing the input.
+func mergeEntryChannelsWithErrors(ctx context.Context, in []chan metaCacheEntry, out chan<- metaCacheEntry, readQuorum int, terminalErr func(int) error) error {
+	return mergeEntryChannelsWithResolver(ctx, in, out, readQuorum, terminalErr, nil)
+}
+
+// resolve is called only for groups containing a pending listing entry. All
+// selected inputs are advanced (and their terminal errors checked) before it is
+// called, so recovery never holds an object lock while waiting for input.
+func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntry, out chan<- metaCacheEntry, readQuorum int, terminalErr func(int) error, resolve func([]*metaCacheEntry) (*metaCacheEntry, error)) error {
 	defer xioutil.SafeClose(out)
 	top := make([]*metaCacheEntry, len(in))
 	nDone := 0
 	ctxDone := ctx.Done()
 
-	// Use simpler forwarder.
+	// Keep the single-input normal path allocation-free. Only a pending entry
+	// requires looking ahead to observe its stream's terminal error first.
 	if len(in) == 1 {
+		var buffered metaCacheEntry
+		haveBuffered, exhausted := false, false
 		for {
-			select {
+			var v metaCacheEntry
-			case <-ctxDone:
+			if haveBuffered {
-				return ctx.Err()
+				v, haveBuffered = buffered, false
-			case v, ok := <-in[0]:
+			} else {
-				if !ok {
+				select {
-					return nil
+				case <-ctxDone:
+					return ctx.Err()
+				case entry, ok := <-in[0]:
+					if !ok {
+						if terminalErr != nil {
+							return terminalErr(0)
+						}
+						return nil
+					}
+					v = entry
+				}
+			}
+			if v.listingSet != nil {
+				if resolve == nil {
+					return errInvalidArgument
 				}
 				select {
 				case <-ctxDone:
 					return ctx.Err()
-				case out <- v:
+				case entry, ok := <-in[0]:
+					buffered, haveBuffered, exhausted = entry, ok, !ok
+					if !ok && terminalErr != nil {
+						if err := terminalErr(0); err != nil {
+							return err
+						}
+					}
 				}
+				pending := v
+				resolved, err := resolve([]*metaCacheEntry{&pending})
+				if err != nil {
+					return err
+				}
+				if resolved == nil {
+					if exhausted {
+						return nil
+					}
+					continue
+				}
+				v = *resolved
+			}
+			select {
+			case <-ctxDone:
+				return ctx.Err()
+			case out <- v:
+			}
+			if exhausted {
+				return nil
 			}
 		}
 	}
@@ -700,6 +760,11 @@ func mergeEntryChannels(ctx context.Context, in []chan metaCacheEntry, out chan<
 			return ctx.Err()
 		case entry, ok := <-in[idx]:
 			if !ok {
+				if terminalErr != nil {
+					if err := terminalErr(idx); err != nil {
+						return err
+					}
+				}
 				top[idx] = nil
 				nDone++
 			} else {
@@ -778,6 +843,40 @@ func mergeEntryChannels(ctx context.Context, in []chan metaCacheEntry, out chan<
 			}
 		}
 
+		pending := best.listingSet != nil
+		for _, idx := range toMerge {
+			pending = pending || top[idx].listingSet != nil
+		}
+		if pending {
+			if resolve == nil {
+				return errInvalidArgument
+			}
+			group := make([]*metaCacheEntry, 1, len(toMerge)+1)
+			group[0] = best
+			for _, idx := range toMerge {
+				group = append(group, top[idx])
+				if err := selectFrom(idx); err != nil {
+					return err
+				}
+			}
+			if err := selectFrom(bestIdx); err != nil {
+				return err
+			}
+			resolved, err := resolve(group)
+			if err != nil {
+				return err
+			}
+			if resolved != nil && resolved.name > last {
+				select {
+				case <-ctxDone:
+					return ctx.Err()
+				case out <- *resolved:
+					last = resolved.name
+				}
+			}
+			continue
+		}
+
 		// Merge any unmerged
 		if len(toMerge) > 0 {
 			versions := make([][]xlMetaV2ShallowVersion, 0, len(toMerge)+1)

cmd/metacache-listing-cache_test.go

@@ -0,0 +1,403 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// 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"
+	"context"
+	"encoding/xml"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"slices"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/minio/minio/internal/grid"
+)
+
+type listingReadFailDisk struct {
+	StorageAPI
+	object string
+	reads  *atomic.Int64
+}
+
+func (d *listingReadFailDisk) ReadXL(ctx context.Context, bucket, object string, readData bool) (RawFileInfo, error) {
+	if object == d.object {
+		d.reads.Add(1)
+		return RawFileInfo{}, errDiskNotFound
+	}
+	return d.StorageAPI.ReadXL(ctx, bucket, object, readData)
+}
+
+// Install before issuing requests: cache writers can outlive their HTTP
+// response, so disk faults must be switched without replacing getDisks.
+// Each published slice must remain immutable.
+func installListingDiskSwitch(set *erasureObjects) ([]StorageAPI, func([]StorageAPI)) {
+	disks := set.getDisks()
+	var current atomic.Value
+	current.Store(disks)
+	set.getDisks = func() []StorageAPI { return current.Load().([]StorageAPI) }
+	return disks, func(disks []StorageAPI) { current.Store(disks) }
+}
+
+func TestListingCachePreservesFallbackError(t *testing.T) {
+	for _, continuation := range []bool{false, true} {
+		name := "cold"
+		if continuation {
+			name = "continuation"
+		}
+		t.Run(name, func(t *testing.T) {
+			testListingCacheFallbackError(t, continuation, false)
+		})
+	}
+}
+
+func TestListingCacheCleanupKeepsOriginalID(t *testing.T) {
+	testListingCacheFallbackError(t, true, true)
+}
+
+// Use actual peer RPCs to verify the abandoned cache is updated after the
+// caller clears its own listing ID. The local-only path cannot observe this.
+func recordListingPeerCleanup(t *testing.T) func() {
+	t.Helper()
+	tg, err := grid.SetupTestGrid(2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	previous := globalNotificationSys
+	t.Cleanup(func() {
+		tg.Cleanup()
+		globalNotificationSys = previous
+	})
+	getIDs := make(chan string, 8)
+	type cleanupResult struct {
+		id  string
+		err *grid.RemoteErr
+	}
+	updated := make(chan cleanupResult, 1)
+	server := &peerRESTServer{}
+	if err := getMetacacheListingRPC.Register(tg.Managers[1], func(o *listPathOptions) (*metacache, *grid.RemoteErr) {
+		getIDs <- o.ID
+		return server.GetMetacacheListingHandler(o)
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := updateMetacacheListingRPC.Register(tg.Managers[1], func(m *metacache) (*metacache, *grid.RemoteErr) {
+		result, err := server.UpdateMetacacheListingHandler(m)
+		if m.error == "no longer used" {
+			updated <- cleanupResult{id: m.id, err: err}
+		}
+		return result, err
+	}); err != nil {
+		t.Fatal(err)
+	}
+	client := &peerRESTClient{gridConn: func() *grid.Connection { return tg.Managers[0].Connection(tg.Hosts[1]) }}
+	globalNotificationSys = &NotificationSys{peerClients: []*peerRESTClient{client}, allPeerClients: []*peerRESTClient{client}}
+	return func() {
+		ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+		defer cancel()
+		var ids []string
+		for range 2 {
+			select {
+			case id := <-getIDs:
+				ids = append(ids, id)
+			case <-ctx.Done():
+				t.Fatal("peer did not receive the failed cache cleanup lookup")
+			}
+		}
+		if ids[0] == "" || ids[1] != ids[0] {
+			t.Fatalf("cleanup lookup lost the original cache ID: %q", ids)
+		}
+		select {
+		case result := <-updated:
+			if result.err != nil || result.id != ids[0] {
+				t.Fatalf("cleanup update: id=%q err=%v, want id=%q", result.id, result.err, ids[0])
+			}
+			t.Logf("peer cleanup Get/Update retained cache ID %q", result.id)
+		case <-ctx.Done():
+			t.Fatal("peer did not finish the failed cache cleanup update")
+		}
+	}
+}
+
+func testListingCacheFallbackError(t *testing.T, continuation, peer bool) {
+	t.Helper()
+	z, bucket, router := nullQuorumBackend(t)
+	var waitPeerCleanup func()
+	if peer {
+		waitPeerCleanup = recordListingPeerCleanup(t)
+	}
+	set := z.serverPools[0].sets[0]
+	disks, switchDisks := installListingDiskSwitch(set)
+	body := bytes.Repeat([]byte{'a'}, 8192)
+	mtime := time.Now().UTC().Add(-time.Hour)
+	for _, object := range []string{"a", "b", "z"} {
+		nullQuorumPut(t, z, bucket, object, body, ObjectOptions{MTime: mtime})
+	}
+	old := make([][]byte, 3)
+	for i := range old {
+		old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, "z")
+	}
+	nullQuorumPut(t, z, bucket, "z", body, ObjectOptions{MTime: mtime.Add(time.Minute)})
+	var token string
+	if continuation {
+		rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1", "", "", ""))
+		var list ListObjectsV2Response
+		if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil || !list.IsTruncated || list.NextContinuationToken == "" {
+			t.Fatalf("first page: %d %s", rec.Code, rec.Body.String())
+		}
+		token = list.NextContinuationToken
+	}
+	for i := range old {
+		if err := disks[i].WriteAll(t.Context(), bucket, "z/"+xlStorageFormatFile, old[i]); err != nil {
+			t.Fatal(err)
+		}
+	}
+	var reads atomic.Int64
+	walkers := slices.Clone(disks)
+	for i := range walkers {
+		walkers[i] = &listingReadFailDisk{StorageAPI: disks[i], object: "z", reads: &reads}
+		if i >= 8 {
+			walkers[i] = &failedMetadataWalkDisk{StorageAPI: walkers[i]}
+		}
+	}
+	switchDisks(walkers)
+	target := getListObjectsV2URL("", bucket, "", "1000", "", "", "")
+	if token != "" {
+		target += "&continuation-token=" + url.QueryEscape(token)
+	}
+	rec := nullQuorumRequest(t, router, http.MethodGet, target)
+	t.Logf("LIST=%d failed metadata reads=%d body=%s", rec.Code, reads.Load(), rec.Body.String())
+	if reads.Load() == 0 {
+		t.Fatal("fallback not exercised")
+	}
+	var apiErr APIErrorResponse
+	if rec.Code != http.StatusServiceUnavailable || xml.Unmarshal(rec.Body.Bytes(), &apiErr) != nil || apiErr.Code != "SlowDownRead" {
+		t.Fatalf("fallback failure lost its read-quorum error: %d %s", rec.Code, rec.Body.String())
+	}
+	if waitPeerCleanup != nil {
+		waitPeerCleanup()
+	}
+	if continuation {
+		caches, _ := localMetacacheMgr.getBucket(t.Context(), bucket).cloneCaches()
+		if len(caches) == 0 {
+			t.Fatal("continuation did not create a cache")
+		}
+		for _, cache := range caches {
+			if cache.status != scanStateError {
+				t.Fatalf("incomplete cache was not marked failed: %+v", cache)
+			}
+		}
+	}
+	switchDisks(disks)
+	retry := nullQuorumRequest(t, router, http.MethodGet, target)
+	var recovered ListObjectsV2Response
+	if retry.Code != http.StatusOK || xml.Unmarshal(retry.Body.Bytes(), &recovered) != nil {
+		t.Fatalf("retry after metadata recovers: %d %s", retry.Code, retry.Body.String())
+	}
+	var keys []string
+	for _, entry := range recovered.Contents {
+		keys = append(keys, entry.Key)
+	}
+	want := []string{"a", "b", "z"}
+	if continuation {
+		want = []string{"b", "z"}
+	}
+	if !slices.Equal(keys, want) || recovered.IsTruncated {
+		t.Fatalf("retry returned incomplete result: %+v", recovered)
+	}
+}
+
+func TestListingCachePagination(t *testing.T) {
+	z, bucket, router := nullQuorumBackend(t)
+	for _, key := range []string{"a", "b", "c", "d"} {
+		nullQuorumPut(t, z, bucket, key, []byte(key), ObjectOptions{})
+	}
+	var keys []string
+	var token string
+	var finished bool
+	for page := 0; page < 8; page++ {
+		target := getListObjectsV2URL("", bucket, "", "1", "", "", "")
+		if token != "" {
+			target += "&continuation-token=" + url.QueryEscape(token)
+		}
+		rec := nullQuorumRequest(t, router, http.MethodGet, target)
+		var list ListObjectsV2Response
+		if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil {
+			t.Fatalf("page %d: %d %s", page, rec.Code, rec.Body.String())
+		}
+		for _, entry := range list.Contents {
+			keys = append(keys, entry.Key)
+		}
+		if !list.IsTruncated {
+			finished = true
+			break
+		}
+		if list.NextContinuationToken == "" || list.NextContinuationToken == token {
+			t.Fatal("pagination did not advance")
+		}
+		token = list.NextContinuationToken
+	}
+	if !finished || !slices.Equal(keys, []string{"a", "b", "c", "d"}) {
+		t.Fatalf("pagination returned %v, finished=%v", keys, finished)
+	}
+}
+
+// A cached full page must remain streamable even if a later key is slow. The
+// completion barrier belongs at EOF, not before every page response.
+func TestListingCacheFullPageDoesNotWaitForTail(t *testing.T) {
+	z, bucket, router := nullQuorumBackend(t)
+	disks, switchDisks := installListingDiskSwitch(z.serverPools[0].sets[0])
+	for _, key := range []string{"a", "b", "c", "d", "e", "zz"} {
+		nullQuorumPut(t, z, bucket, key, []byte(key), ObjectOptions{})
+	}
+	first := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1", "", "", ""))
+	var page ListObjectsV2Response
+	if first.Code != http.StatusOK || xml.Unmarshal(first.Body.Bytes(), &page) != nil || !page.IsTruncated {
+		t.Fatalf("first page: %d %s", first.Code, first.Body.String())
+	}
+	resume := make(chan struct{})
+	release := sync.OnceFunc(func() { close(resume) })
+	defer release()
+	var blocked, active atomic.Int64
+	walkers := slices.Clone(disks)
+	for i, disk := range walkers {
+		walkers[i] = &listingPausedTailDisk{StorageAPI: disk, bucket: bucket, tail: "zz", resume: resume, blocked: &blocked, active: &active}
+	}
+	switchDisks(walkers)
+	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+	defer cancel()
+	target := getListObjectsV2URL("", bucket, "", "1", "", "", "") + "&continuation-token=" + url.QueryEscape(page.NextContinuationToken)
+	req, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	response := make(chan *httptest.ResponseRecorder, 1)
+	go func() {
+		rec := httptest.NewRecorder()
+		router.ServeHTTP(rec, req.WithContext(ctx))
+		response <- rec
+	}()
+	select {
+	case rec := <-response:
+		page = ListObjectsV2Response{}
+		if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &page) != nil || !page.IsTruncated || len(page.Contents) != 1 || page.Contents[0].Key != "b" || page.NextContinuationToken == "" {
+			t.Fatalf("full page before tail: %d %s", rec.Code, rec.Body.String())
+		}
+		if blocked.Load() == 0 || active.Load() == 0 {
+			t.Fatal("full page did not overlap the paused tail")
+		}
+	case <-ctx.Done():
+		t.Fatal("full page waited for the paused tail")
+	}
+	release()
+	waitForListingCondition(t, "disk walkers did not finish", func() bool { return active.Load() == 0 })
+	target = getListObjectsV2URL("", bucket, "", "1000", "", "", "") + "&continuation-token=" + url.QueryEscape(page.NextContinuationToken)
+	rec := nullQuorumRequest(t, router, http.MethodGet, target)
+	page = ListObjectsV2Response{}
+	if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &page) != nil || page.IsTruncated {
+		t.Fatalf("page after tail resumes: %d %s", rec.Code, rec.Body.String())
+	}
+	var keys []string
+	for _, entry := range page.Contents {
+		keys = append(keys, entry.Key)
+	}
+	if !slices.Equal(keys, []string{"c", "d", "e", "zz"}) {
+		t.Fatalf("cached remainder lost keys: %v", keys)
+	}
+}
+
+// Keep the error update pending at the actual peer RPC boundary. Other cache
+// operations, including readers and abandoned-cache cleanup, remain live.
+func listingCacheErrorPeer(t *testing.T, fail bool) (*peerRESTClient, <-chan struct{}, func()) {
+	t.Helper()
+	tg, err := grid.SetupTestGrid(2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ready, resume := make(chan struct{}), make(chan struct{})
+	release := sync.OnceFunc(func() { close(resume) })
+	t.Cleanup(func() {
+		release()
+		tg.Cleanup()
+	})
+	var entered sync.Once
+	server := &peerRESTServer{}
+	if err := getMetacacheListingRPC.Register(tg.Managers[1], server.GetMetacacheListingHandler); err != nil {
+		t.Fatal(err)
+	}
+	if err := updateMetacacheListingRPC.Register(tg.Managers[1], func(m *metacache) (*metacache, *grid.RemoteErr) {
+		if m.error != "" && m.error != "no longer used" {
+			entered.Do(func() { close(ready) })
+			select {
+			case <-resume:
+			case <-t.Context().Done():
+				return nil, grid.NewRemoteErr(t.Context().Err())
+			}
+			if fail {
+				return nil, grid.NewRemoteErr(errors.New("cache error update failed"))
+			}
+		}
+		return server.UpdateMetacacheListingHandler(m)
+	}); err != nil {
+		t.Fatal(err)
+	}
+	return &peerRESTClient{gridConn: func() *grid.Connection { return tg.Managers[0].Connection(tg.Hosts[1]) }}, ready, release
+}
+
+func TestListingCacheRetainsErrorAfterPeerFailure(t *testing.T) {
+	client, ready, release := listingCacheErrorPeer(t, true)
+	o := listPathOptions{ID: mustGetUUID(), Bucket: "bucket"}
+	m := o.newMetacache()
+	mc := metaCacheRPC{meta: &m, rpc: client, o: o}
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		mc.setErr("terminal listing failure")
+	}()
+	t.Cleanup(func() {
+		release()
+		select {
+		case <-done:
+		case <-time.After(5 * time.Second):
+			t.Error("setErr worker did not finish before grid cleanup")
+		}
+	})
+	select {
+	case <-ready:
+	case <-time.After(5 * time.Second):
+		t.Fatal("error update did not reach peer")
+	}
+	release()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("failed peer update did not return")
+	}
+	if m.id != o.ID || m.bucket != o.Bucket || m.status != scanStateError || m.error != "terminal listing failure" || mc.getErr() == nil {
+		t.Fatalf("peer failure erased local failure: %+v", m)
+	}
+	mc.setErr("later failure")
+	if m.error != "terminal listing failure" {
+		t.Fatalf("later failure replaced the terminal error: %+v", m)
+	}
+}

cmd/metacache-listing-delete_test.go

@@ -0,0 +1,133 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// 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"
+	"encoding/xml"
+	"fmt"
+	"net/http"
+	"slices"
+	"testing"
+	"time"
+)
+
+func TestListFallbackPreservesPoolDeleteMarker(t *testing.T) {
+	for _, markerPool := range []int{0, 1} {
+		t.Run(fmt.Sprint(markerPool), func(t *testing.T) {
+			z, bucket, router := listingPoolsBackend(t)
+			ctx := t.Context()
+			live, deleted := z.serverPools[1-markerPool].sets[0], z.serverPools[markerPool].sets[0]
+			disks, switchDisks := installListingDiskSwitch(deleted)
+			const object = "gone"
+			body := bytes.Repeat([]byte{'a'}, 8192)
+			oldTime := time.Now().UTC().Add(-time.Hour)
+			for _, set := range []*erasureObjects{live, deleted} {
+				_, err := set.PutObject(ctx, bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{MTime: oldTime, VersionSuspended: true})
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+			old := make([][]byte, 3)
+			for i := range old {
+				old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, object)
+			}
+			_, err := deleted.DeleteObject(ctx, bucket, object, ObjectOptions{VersionSuspended: true, DeleteMarker: true, MTime: oldTime.Add(time.Minute)})
+			if err != nil {
+				t.Fatal(err)
+			}
+			get := nullQuorumRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object))
+			if get.Code != http.StatusNotFound {
+				t.Fatalf("expected deleted GET, got %d %s", get.Code, get.Body.String())
+			}
+			first := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
+			var page ListObjectsV2Response
+			if first.Code != http.StatusOK || xml.Unmarshal(first.Body.Bytes(), &page) != nil || len(page.Contents) != 0 {
+				t.Fatalf("normal LIST marker: %d %s", first.Code, first.Body.String())
+			}
+			for i := range old {
+				if err := disks[i].WriteAll(ctx, bucket, object+"/"+xlStorageFormatFile, old[i]); err != nil {
+					t.Fatal(err)
+				}
+			}
+			walkers := slices.Clone(disks)
+			for i := 8; i < len(walkers); i++ {
+				walkers[i] = &failedMetadataWalkDisk{StorageAPI: disks[i]}
+			}
+			switchDisks(walkers)
+			fi, _, _, err := deleted.getObjectFileInfo(ctx, bucket, object, ObjectOptions{}, false)
+			if err != nil || !fi.Deleted {
+				t.Fatalf("read fallback marker: %+v %v", fi, err)
+			}
+			rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
+			page = ListObjectsV2Response{}
+			if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &page) != nil || len(page.Contents) != 0 {
+				t.Fatalf("fallback resurrected old pool copy: %d %s", rec.Code, rec.Body.String())
+			}
+		})
+	}
+}
+
+// The regular metadata reader can translate a persistent generation split to
+// ObjectNotFound even though every drive still has valid metadata. LIST must
+// preserve that uncertainty instead of treating the key as deleted.
+func TestListFallbackRejectsSplitGenerations(t *testing.T) {
+	z, bucket, _ := nullQuorumBackend(t)
+	set := z.serverPools[0].sets[0]
+	disks := set.getDisks()
+	const object = "object"
+	mtime := time.Now().UTC().Add(-time.Hour)
+	var generations [3][][]byte
+	for generation := range generations {
+		body := bytes.Repeat([]byte{byte('a' + generation)}, 8192)
+		nullQuorumPut(t, z, bucket, object, body, ObjectOptions{MTime: mtime.Add(time.Duration(generation) * time.Minute)})
+		generations[generation] = make([][]byte, len(disks))
+		for i, disk := range disks {
+			generations[generation][i] = mustReadNullQuorumMeta(t, disk, bucket, object)
+		}
+	}
+	// All sixteen drives are online, but the 7/5/4 split leaves no generation
+	// with the eight matching snapshots required by the listing resolver.
+	for i, disk := range disks {
+		generation := 0
+		if i >= 7 {
+			generation = 1
+		}
+		if i >= 12 {
+			generation = 2
+		}
+		if err := disk.WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, generations[generation][i]); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, _, _, err := set.getObjectFileInfo(t.Context(), bucket, object, ObjectOptions{}, false); !isErrObjectNotFound(err) {
+		t.Fatalf("split metadata did not exercise ambiguous not-found: %v", err)
+	}
+	entries := make(metaCacheEntries, len(disks))
+	for i, disk := range disks {
+		entries[i] = metaCacheEntry{name: object, metadata: mustReadNullQuorumMeta(t, disk, bucket, object)}
+		if _, err := entries[i].xlmeta(); err != nil {
+			t.Fatalf("drive %d no longer has valid metadata: %v", i, err)
+		}
+	}
+	resolver := metadataResolutionParams{dirQuorum: 8, objQuorum: 8, bucket: bucket, requestedVersions: 1}
+	entry, err := z.resolveListingGroup(t.Context(), bucket, []*metaCacheEntry{set.resolveListEntry(entries, nil, &resolver)})
+	if entry != nil || !isErrReadQuorum(err) {
+		t.Fatalf("split live metadata became absence: entry=%v err=%v", entry, err)
+	}
+}

cmd/metacache-listing-http_test.go

@@ -0,0 +1,101 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// 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"
+	"encoding/xml"
+	"net/http"
+	"net/http/httptest"
+	"slices"
+	"testing"
+	"time"
+)
+
+// Exercise the actual LIST pipeline with the disputed entry in pool 1.
+func TestListFallbackMultiPoolHTTPReadLock(t *testing.T) {
+	for _, object := range []string{"object", "dir/"} {
+		t.Run(object, func(t *testing.T) {
+			z, bucket, router := listingPoolsBackend(t)
+			ctx := t.Context()
+			set := z.serverPools[1].sets[0]
+			disks := set.getDisks()
+			encoded := encodeDirObject(object)
+			body := bytes.Repeat([]byte{'a'}, 8192)
+			if object == "dir/" {
+				body = nil
+			}
+			mtime := time.Now().UTC().Add(-time.Hour)
+			put := func(tm time.Time) {
+				_, err := set.PutObject(ctx, bucket, encoded, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{MTime: tm})
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+			put(mtime)
+			old := make([][]byte, 3)
+			for i := range old {
+				old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, encoded)
+			}
+			put(mtime.Add(time.Minute))
+			for i := range old {
+				if err := disks[i].WriteAll(ctx, bucket, encoded+"/"+xlStorageFormatFile, old[i]); err != nil {
+					t.Fatal(err)
+				}
+			}
+			walkers := slices.Clone(disks)
+			for i := 8; i < len(walkers); i++ {
+				walkers[i] = &failedMetadataWalkDisk{StorageAPI: disks[i]}
+			}
+			set.getDisks = func() []StorageAPI { return walkers }
+			lock := z.NewNSLock(bucket, encoded)
+			lkctx, err := lock.GetLock(ctx, globalOperationTimeout)
+			if err != nil {
+				t.Fatal(err)
+			}
+			locked := true
+			defer func() {
+				if locked {
+					lock.Unlock(lkctx)
+				}
+			}()
+			target := getListObjectsV2URL("", bucket, "", "1000", "", "", "")
+			req, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			rec := httptest.NewRecorder()
+			done := make(chan struct{})
+			go func() {
+				defer close(done)
+				router.ServeHTTP(rec, req.WithContext(ctx))
+			}()
+			waitForListingReadLock(t, z.serverPools[0].sets[0].nsMutex, bucket, encoded, done)
+			lock.Unlock(lkctx)
+			locked = false
+			<-done
+			var list ListObjectsV2Response
+			if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil {
+				t.Fatalf("LIST HTTP=%d body=%s", rec.Code, rec.Body.String())
+			}
+			if len(list.Contents) != 1 || list.Contents[0].Key != object {
+				t.Fatalf("LIST lost key: %+v", list)
+			}
+		})
+	}
+}

cmd/metacache-listing-lock-routes_test.go

@@ -0,0 +1,131 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// 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 (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/minio/minio/internal/dsync"
+)
+
+// Use the real distributed-lock algorithm with in-process lock servers.
+// This is a protocol-level check, not a network or rolling-restart test.
+func TestListFallbackDistributedLockRoute(t *testing.T) {
+	for _, tc := range []struct {
+		name                     string
+		multiPool, sharedLockers bool
+	}{
+		{"multi-pool-shared-lockers", true, true},
+		{"multi-pool-disjoint-lockers", true, false},
+		{"single-pool-target-set-lockers", false, false},
+		{"single-pool-first-set-lockers", false, false},
+		{"single-pool-shared-ns-distinct-lockers", false, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			a := []dsync.NetLocker{newLocker(), newLocker()}
+			b := []dsync.NetLocker{newLocker(), newLocker()}
+			if tc.sharedLockers {
+				b = a
+			}
+			set0 := &erasureObjects{nsMutex: newNSLock(true), getLockers: func() ([]dsync.NetLocker, string) { return a, "listing-test" }}
+			target := &erasureObjects{nsMutex: newNSLock(true), getLockers: func() ([]dsync.NetLocker, string) { return b, "listing-test" }}
+			if tc.name == "single-pool-shared-ns-distinct-lockers" {
+				target.nsMutex = set0.nsMutex
+			}
+			z := &erasureServerPools{serverPools: []*erasureSets{{sets: []*erasureObjects{set0, target}}}}
+			if tc.multiPool {
+				z.serverPools = []*erasureSets{{sets: []*erasureObjects{set0}}, {sets: []*erasureObjects{target}}}
+			}
+			writer := target.NewNSLock("bucket", "object")
+			if tc.multiPool || tc.name == "single-pool-first-set-lockers" {
+				writer = z.NewNSLock("bucket", "object")
+			}
+			wctx, err := writer.GetLock(t.Context(), globalOperationTimeout)
+			if err != nil {
+				t.Fatal(err)
+			}
+			locked := true
+			defer func() {
+				if locked {
+					writer.Unlock(wctx)
+				}
+			}()
+			ctx, cancel := context.WithTimeout(t.Context(), 80*time.Millisecond)
+			defer cancel()
+			_, unlock, err := z.listingRLock(ctx, "bucket", "object", target)
+			if err == nil {
+				unlock()
+				t.Fatal("fallback read lock acquired while ordinary writer lock is held")
+			}
+			writer.Unlock(wctx)
+			locked = false
+			ctx, cancel = context.WithTimeout(t.Context(), 5*time.Second)
+			defer cancel()
+			_, unlock, err = z.listingRLock(ctx, "bucket", "object", target)
+			if err != nil {
+				t.Fatalf("fallback read lock failed after writer unlock: %v", err)
+			}
+			unlock()
+		})
+	}
+}
+
+// Cancellation while waiting for the second domain must release the first.
+func TestListingSecondLockCancellationReleasesFirst(t *testing.T) {
+	first := &erasureObjects{nsMutex: newNSLock(false)}
+	target := &erasureObjects{nsMutex: newNSLock(false)}
+	z := &erasureServerPools{serverPools: []*erasureSets{{sets: []*erasureObjects{first, target}}}}
+	writer := target.NewNSLock("bucket", "object")
+	wctx, err := writer.GetLock(t.Context(), globalOperationTimeout)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer writer.Unlock(wctx)
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	done := make(chan struct{})
+	var lockErr error
+	go func() {
+		defer close(done)
+		_, unlock, err := z.listingRLock(ctx, "bucket", "object", target)
+		lockErr = err
+		if err == nil {
+			unlock()
+		}
+	}()
+	waitForListingReadLock(t, target.nsMutex, "bucket", "object", done)
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("second lock did not cancel")
+	}
+	if lockErr == nil {
+		t.Fatal("canceled second lock succeeded")
+	}
+	poolWriter := z.NewNSLock("bucket", "object")
+	ctx, cancel = context.WithTimeout(t.Context(), time.Second)
+	defer cancel()
+	pctx, err := poolWriter.GetLock(ctx, globalOperationTimeout)
+	if err != nil {
+		t.Fatalf("first lock leaked after second failed: %v", err)
+	}
+	poolWriter.Unlock(pctx)
+}

cmd/metacache-listing-locks_test.go

@@ -0,0 +1,137 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// 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"
+	"context"
+	"net/http"
+	"testing"
+	"time"
+)
+
+// The fallback on pool 1 must contend with the
+// pools-layer lock used by ordinary multi-pool writers (anchored at pool 0).
+func TestListFallbackMultiPoolReadLock(t *testing.T) {
+	z, bucket, _ := listingPoolsBackend(t)
+	ctx := t.Context()
+	set := z.serverPools[1].sets[0]
+	disks := set.getDisks()
+	const object = "object"
+	body := bytes.Repeat([]byte{'a'}, 8192)
+	mtime := time.Now().UTC().Add(-time.Hour)
+	put := func(tm time.Time) {
+		_, err := set.PutObject(ctx, bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{MTime: tm})
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	put(mtime)
+	old := make([][]byte, len(disks))
+	for i := range old {
+		old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, object)
+	}
+	put(mtime.Add(time.Minute))
+	entries := make(metaCacheEntries, len(disks))
+	errs := make([]error, len(disks))
+	for i := range 3 {
+		entries[i] = metaCacheEntry{name: object, metadata: old[i]}
+	}
+	for i := 3; i < 8; i++ {
+		entries[i] = metaCacheEntry{name: object, metadata: mustReadNullQuorumMeta(t, disks[i], bucket, object)}
+	}
+	for i := 8; i < len(disks); i++ {
+		errs[i] = errDiskNotFound
+	}
+	resolver := metadataResolutionParams{dirQuorum: 8, objQuorum: 8, bucket: bucket, requestedVersions: 1}
+	lock := z.NewNSLock(bucket, object)
+	lkctx, err := lock.GetLock(ctx, globalOperationTimeout)
+	if err != nil {
+		t.Fatal(err)
+	}
+	locked := true
+	defer func() {
+		if locked {
+			lock.Unlock(lkctx)
+		}
+	}()
+	done := make(chan struct{})
+	var entry *metaCacheEntry
+	var readErr error
+	go func() {
+		defer close(done)
+		entry, readErr = z.resolveListingGroup(ctx, bucket, []*metaCacheEntry{set.resolveListEntry(entries, errs, &resolver)})
+	}()
+	waitForListingReadLock(t, z.serverPools[0].sets[0].nsMutex, bucket, object, done)
+	lock.Unlock(lkctx)
+	locked = false
+	<-done
+	if readErr != nil || entry == nil || entry.name != object {
+		t.Fatalf("fallback after writer unlock: entry=%v err=%v", entry, readErr)
+	}
+}
+
+// Observe the reader registered on the writer's actual namespace lock before
+// releasing the writer. This tests mutual exclusion without a timed sleep.
+func waitForListingReadLock(t *testing.T, mutex *nsLockMap, bucket, object string, done <-chan struct{}) {
+	t.Helper()
+	timer := time.NewTimer(5 * time.Second)
+	defer timer.Stop()
+	tick := time.NewTicker(time.Millisecond)
+	defer tick.Stop()
+	for {
+		mutex.lockMapMutex.Lock()
+		lock := mutex.lockMap[pathJoin(bucket, object)]
+		waiting := lock != nil && lock.ref > 1
+		mutex.lockMapMutex.Unlock()
+		if waiting {
+			return
+		}
+		select {
+		case <-done:
+			t.Fatal("listing completed while the writer held its namespace lock")
+		case <-timer.C:
+			t.Fatal("listing did not wait on the writer's namespace lock")
+		case <-tick.C:
+		}
+	}
+}
+
+func listingPoolsBackend(t *testing.T) (*erasureServerPools, string, http.Handler) {
+	t.Helper()
+	ctx, cancel := context.WithCancel(t.Context())
+	obj, dirs, err := prepareErasurePoolsWithContext(ctx)
+	if err != nil {
+		cancel()
+		t.Fatal(err)
+	}
+	z := obj.(*erasureServerPools)
+	previous := newObjectLayerFn()
+	setObjectLayer(z)
+	t.Cleanup(func() {
+		cancel()
+		z.Shutdown(context.Background())
+		removeRoots(dirs)
+		setObjectLayer(previous)
+	})
+	bucket, router, err := initAPIHandlerTest(ctx, z, nil, MakeBucketOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return z, bucket, router
+}

cmd/metacache-listing-merge_test.go

@@ -0,0 +1,244 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// 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"
+	"context"
+	"encoding/xml"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"slices"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/klauspost/compress/s2"
+	"github.com/tinylib/msgp/msgp"
+)
+
+// Replay real disk-walk metadata in its existing wire format, but pause before
+// the final key. This keeps the healthy pool's listing alive after the other
+// pool has already failed, without forging entries or listing errors.
+type listingPausedTailDisk struct {
+	StorageAPI
+	bucket  string
+	tail    string
+	resume  <-chan struct{}
+	blocked *atomic.Int64
+	active  *atomic.Int64
+}
+
+func (d *listingPausedTailDisk) WalkDir(ctx context.Context, opts WalkDirOptions, out io.Writer) (result error) {
+	if opts.Bucket != d.bucket {
+		return d.StorageAPI.WalkDir(ctx, opts, out)
+	}
+	if d.active != nil {
+		d.active.Add(1)
+		defer d.active.Add(-1)
+	}
+	var source bytes.Buffer
+	if err := d.StorageAPI.WalkDir(ctx, opts, &source); err != nil {
+		return err
+	}
+	r := newMetacacheReader(bytes.NewReader(source.Bytes()))
+	defer r.Close()
+	compressed := s2.NewWriter(out)
+	defer func() {
+		if err := compressed.Close(); result == nil {
+			result = err
+		}
+	}()
+	w := msgp.NewWriter(compressed)
+	if err := w.WriteByte(metacacheStreamVersion); err != nil {
+		return err
+	}
+	for {
+		entry, err := r.next()
+		if err == io.EOF {
+			if err = w.WriteBool(false); err != nil {
+				return err
+			}
+			if err = w.Flush(); err != nil {
+				return err
+			}
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		if entry.name == d.tail {
+			if err = w.Flush(); err != nil {
+				return err
+			}
+			if err = compressed.Flush(); err != nil {
+				return err
+			}
+			d.blocked.Add(1)
+			select {
+			case <-d.resume:
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		}
+		if err = w.WriteBool(true); err != nil {
+			return err
+		}
+		if err = w.WriteString(entry.name); err != nil {
+			return err
+		}
+		if err = w.WriteBytes(entry.metadata); err != nil {
+			return err
+		}
+	}
+}
+
+func TestListingMergeDoesNotSkipFailedPool(t *testing.T) {
+	ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second)
+	defer cancel()
+	z, bucket, router := listingPoolsBackend(t)
+	broken, healthy := z.serverPools[0].sets[0], z.serverPools[1].sets[0]
+	brokenDisks, switchBrokenDisks := installListingDiskSwitch(broken)
+	healthyDisks, switchHealthyDisks := installListingDiskSwitch(healthy)
+	body := bytes.Repeat([]byte{'a'}, 8192)
+	mtime := time.Now().UTC().Add(-time.Hour)
+	put := func(set *erasureObjects, name string, tm time.Time) {
+		_, err := set.PutObject(ctx, bucket, name, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{MTime: tm})
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	put(broken, "a", mtime)
+	put(broken, "e", mtime)
+	for _, name := range []string{"f", "g", "h", "i", "j", "k", "zz"} {
+		put(healthy, name, mtime)
+	}
+	first := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1", "", "", ""))
+	var page ListObjectsV2Response
+	if first.Code != http.StatusOK || xml.Unmarshal(first.Body.Bytes(), &page) != nil || !page.IsTruncated || page.NextContinuationToken == "" || len(page.Contents) != 1 || page.Contents[0].Key != "a" {
+		t.Fatalf("first page HTTP=%d body=%s", first.Code, first.Body.String())
+	}
+	firstToken := page.NextContinuationToken
+	old := make([][]byte, 3)
+	for i := range old {
+		old[i] = mustReadNullQuorumMeta(t, brokenDisks[i], bucket, "e")
+	}
+	put(broken, "e", mtime.Add(time.Minute))
+	for i := range old {
+		if err := brokenDisks[i].WriteAll(ctx, bucket, "e/"+xlStorageFormatFile, old[i]); err != nil {
+			t.Fatal(err)
+		}
+	}
+	var reads, blocked, active atomic.Int64
+	brokenWalkers := slices.Clone(brokenDisks)
+	for i := range brokenWalkers {
+		brokenWalkers[i] = &listingReadFailDisk{StorageAPI: brokenDisks[i], object: "e", reads: &reads}
+		if i >= 8 {
+			brokenWalkers[i] = &failedMetadataWalkDisk{StorageAPI: brokenWalkers[i]}
+		}
+	}
+	switchBrokenDisks(brokenWalkers)
+	resume := make(chan struct{})
+	defer close(resume)
+	healthyWalkers := slices.Clone(healthyDisks)
+	for i, disk := range healthyWalkers {
+		healthyWalkers[i] = &listingPausedTailDisk{StorageAPI: disk, bucket: bucket, tail: "zz", resume: resume, blocked: &blocked, active: &active}
+	}
+	switchHealthyDisks(healthyWalkers)
+
+	target := getListObjectsV2URL("", bucket, "", "2", "", "", "") + "&continuation-token=" + url.QueryEscape(firstToken)
+	req, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	response := make(chan *httptest.ResponseRecorder, 1)
+	go func() {
+		rec := httptest.NewRecorder()
+		router.ServeHTTP(rec, req.WithContext(ctx))
+		response <- rec
+	}()
+	select {
+	case rec := <-response:
+		t.Logf("continuation LIST=%d failed fallback ReadXL=%d paused healthy walkers=%d body=%s", rec.Code, reads.Load(), blocked.Load(), rec.Body.String())
+		if reads.Load() == 0 || blocked.Load() == 0 {
+			t.Fatal("schedule did not exercise fallback failure and unfinished healthy pool")
+		}
+		var apiErr APIErrorResponse
+		if rec.Code != http.StatusServiceUnavailable || xml.Unmarshal(rec.Body.Bytes(), &apiErr) != nil || apiErr.Code != "SlowDownRead" {
+			t.Fatalf("failed pool advanced the page instead of reporting read quorum: %d %s", rec.Code, rec.Body.String())
+		}
+		waitForListingCondition(t, "canceled disk walkers did not exit", func() bool { return active.Load() == 0 })
+	case <-ctx.Done():
+		t.Fatalf("no early response: %v reads=%d blocked=%d", ctx.Err(), reads.Load(), blocked.Load())
+	}
+}
+
+func TestListingMergeCancellation(t *testing.T) {
+	z, bucket, _ := nullQuorumBackend(t)
+	for _, key := range []string{"a", "b", "zz"} {
+		nullQuorumPut(t, z, bucket, key, []byte(key), ObjectOptions{})
+	}
+	set := z.serverPools[0].sets[0]
+	var blocked, active atomic.Int64
+	resume := make(chan struct{})
+	defer close(resume)
+	walkers := slices.Clone(set.getDisks())
+	for i, disk := range walkers {
+		walkers[i] = &listingPausedTailDisk{StorageAPI: disk, bucket: bucket, tail: "zz", resume: resume, blocked: &blocked, active: &active}
+	}
+	set.getDisks = func() []StorageAPI { return walkers }
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	entries := make(chan metaCacheEntry, 10)
+	done := make(chan error, 1)
+	go func() {
+		done <- z.listMerged(ctx, listPathOptions{Bucket: bucket, Recursive: true, Limit: 10}, entries)
+	}()
+	waitForListingCondition(t, "disk walks did not reach the paused tail", func() bool { return blocked.Load() > 0 })
+	cancel()
+	select {
+	case err := <-done:
+		if !errors.Is(err, context.Canceled) {
+			t.Fatalf("canceled merge returned %v", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("canceled merge did not finish")
+	}
+	// listMerged must close its output and release every paused disk walk.
+	for range entries {
+	}
+	waitForListingCondition(t, "canceled disk walkers did not exit", func() bool { return active.Load() == 0 })
+}
+
+func waitForListingCondition(t *testing.T, failure string, ready func() bool) {
+	t.Helper()
+	timer := time.NewTimer(5 * time.Second)
+	defer timer.Stop()
+	tick := time.NewTicker(time.Millisecond)
+	defer tick.Stop()
+	for !ready() {
+		select {
+		case <-timer.C:
+			t.Fatal(failure)
+		case <-tick.C:
+		}
+	}
+}

cmd/metacache-listing-recovery_test.go

@@ -0,0 +1,391 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// 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"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"slices"
+	"sync/atomic"
+	"testing"
+	"time"
+	"unsafe"
+)
+
+type listingRecoveryDisk struct {
+	StorageAPI
+	reads        *atomic.Int64
+	err          error
+	emptySuccess bool
+}
+
+func (d *listingRecoveryDisk) ReadXL(ctx context.Context, bucket, object string, data bool) (RawFileInfo, error) {
+	if d.reads != nil {
+		d.reads.Add(1)
+	}
+	if d.emptySuccess {
+		return RawFileInfo{}, nil
+	}
+	if d.err != nil {
+		return RawFileInfo{}, d.err
+	}
+	return d.StorageAPI.ReadXL(ctx, bucket, object, data)
+}
+
+func TestListingRecoveryStrictAbsence(t *testing.T) {
+	z, bucket, _ := nullQuorumBackend(t)
+	set := z.serverPools[0].sets[0]
+	disks, switchDisks := installListingDiskSwitch(set)
+	for _, tc := range []struct {
+		name                   string
+		err                    error
+		offline, empty, absent bool
+	}{
+		{name: "all-native-not-found", absent: true},
+		{name: "offline", offline: true},
+		{name: "volume-not-found", err: errVolumeNotFound},
+		{name: "version-not-found", err: errFileVersionNotFound},
+		{name: "corrupt", err: errFileCorrupt},
+		{name: "decode-generated-not-found", empty: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			next := slices.Clone(disks)
+			next[0] = &listingRecoveryDisk{StorageAPI: disks[0], err: tc.err, emptySuccess: tc.empty}
+			if tc.offline {
+				next[0] = &offlineNullQuorumDisk{StorageAPI: disks[0]}
+			}
+			switchDisks(next)
+			_, absent, err := set.readListingFileInfo(t.Context(), bucket, "missing")
+			if absent != tc.absent || (tc.absent && err != nil) || (!tc.absent && !isErrReadQuorum(err)) {
+				t.Fatalf("absent=%v err=%v, want absent=%v", absent, err, tc.absent)
+			}
+		})
+	}
+	switchDisks(disks)
+}
+
+// The target walk can predate the copy. Only current cross-pool reads can
+// distinguish this copy-before-delete schedule from a genuine object deletion.
+func TestListingRecoveryCurrentPools(t *testing.T) {
+	for _, tc := range []struct {
+		name                           string
+		target, marker, unknown, stale bool
+	}{
+		{name: "copied-before-source-delete", target: true},
+		{name: "deleted-with-stale-other-walk", stale: true},
+		{name: "current-marker-over-stale-live", target: true, marker: true, stale: true},
+		{name: "missing-source-unknown-target", unknown: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			z, bucket, _ := listingPoolsBackend(t)
+			target, source := z.serverPools[0].sets[0], z.serverPools[1].sets[0]
+			targetDisks, switchTarget := installListingDiskSwitch(target)
+			sourceDisks, switchSource := installListingDiskSwitch(source)
+			const object = "moved"
+			body := bytes.Repeat([]byte{'a'}, 8192)
+			mtime := time.Now().UTC().Add(-time.Hour)
+			put := func(set *erasureObjects) {
+				_, err := set.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{MTime: mtime, VersionSuspended: true})
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+			put(source)
+			var stale *metaCacheEntry
+			if tc.stale || tc.target {
+				put(target)
+				stale = &metaCacheEntry{name: object, metadata: mustReadNullQuorumMeta(t, targetDisks[0], bucket, object)}
+			}
+			if tc.stale && !tc.target {
+				if _, err := target.DeleteObject(t.Context(), bucket, object, ObjectOptions{}); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if tc.marker {
+				if _, err := target.DeleteObject(t.Context(), bucket, object, ObjectOptions{VersionSuspended: true, DeleteMarker: true, MTime: mtime.Add(time.Minute)}); err != nil {
+					t.Fatal(err)
+				}
+			}
+			// Actual copy is committed before actual source cleanup.
+			if _, err := source.DeleteObject(t.Context(), bucket, object, ObjectOptions{}); err != nil {
+				t.Fatal(err)
+			}
+			var sourceReads, targetReads atomic.Int64
+			nextSource, nextTarget := slices.Clone(sourceDisks), slices.Clone(targetDisks)
+			for i := range nextSource {
+				nextSource[i] = &listingRecoveryDisk{StorageAPI: sourceDisks[i], reads: &sourceReads}
+			}
+			for i := range nextTarget {
+				nextTarget[i] = &listingRecoveryDisk{StorageAPI: targetDisks[i], reads: &targetReads}
+			}
+			if tc.unknown {
+				nextTarget[0] = &listingRecoveryDisk{StorageAPI: targetDisks[0], reads: &targetReads, err: errDiskNotFound}
+			}
+			switchSource(nextSource)
+			switchTarget(nextTarget)
+			group := []*metaCacheEntry{{name: object, listingSet: source}}
+			if stale != nil && tc.stale {
+				group = append(group, stale)
+			}
+			entry, err := z.resolveListingGroup(t.Context(), bucket, group)
+			if sourceReads.Load() != int64(len(sourceDisks)) || targetReads.Load() != int64(len(targetDisks)) {
+				t.Fatalf("reads source=%d target=%d", sourceReads.Load(), targetReads.Load())
+			}
+			if tc.unknown {
+				if entry != nil || !isErrReadQuorum(err) {
+					t.Fatalf("unknown became success: entry=%v err=%v", entry, err)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatal(err)
+			}
+			if !tc.target {
+				if entry != nil {
+					t.Fatalf("stale walk survived current absence: %+v", entry)
+				}
+				return
+			}
+			if entry == nil || entry.listingSet != nil {
+				t.Fatalf("missing current result: %+v", entry)
+			}
+			fi, err := entry.fileInfo(bucket)
+			if err != nil || fi.Deleted != tc.marker {
+				t.Fatalf("current result deleted=%v err=%v, want marker=%v", fi.Deleted, err, tc.marker)
+			}
+		})
+	}
+}
+
+func TestListingRecoveryNormalPathDoesNotReadOrLock(t *testing.T) {
+	z, bucket, _ := listingPoolsBackend(t)
+	const object = "normal"
+	set := z.serverPools[1].sets[0]
+	body := []byte("normal data")
+	if _, err := set.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	var reads atomic.Int64
+	for _, pool := range z.serverPools {
+		for _, set := range pool.sets {
+			disks, switchDisks := installListingDiskSwitch(set)
+			wrapped := slices.Clone(disks)
+			for i := range wrapped {
+				wrapped[i] = &listingRecoveryDisk{StorageAPI: disks[i], reads: &reads}
+			}
+			switchDisks(wrapped)
+		}
+	}
+	writer := z.NewNSLock(bucket, object)
+	lkctx, err := writer.GetLock(t.Context(), globalOperationTimeout)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer writer.Unlock(lkctx)
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+	defer cancel()
+	out := make(chan metaCacheEntry, 10)
+	err = z.listMerged(ctx, listPathOptions{Bucket: bucket, Recursive: true, Limit: 10}, out)
+	if err != nil && err != io.EOF {
+		t.Fatalf("normal listing waited on writer or failed: %v", err)
+	}
+	var names []string
+	for entry := range out {
+		names = append(names, entry.name)
+		if entry.listingSet != nil {
+			t.Fatal("pending escaped")
+		}
+	}
+	if !slices.Equal(names, []string{object}) || reads.Load() != 0 {
+		t.Fatalf("normal names=%v additional ReadXL=%d", names, reads.Load())
+	}
+}
+
+func TestListingPendingDirectoryCollision(t *testing.T) {
+	for _, objectFirst := range []bool{false, true} {
+		t.Run(fmt.Sprint(objectFirst), func(t *testing.T) {
+			in := []chan metaCacheEntry{make(chan metaCacheEntry, 1), make(chan metaCacheEntry, 1)}
+			objectIndex := 1
+			if objectFirst {
+				objectIndex = 0
+			}
+			in[objectIndex] <- metaCacheEntry{name: "dir/", listingSet: &erasureObjects{}}
+			in[1-objectIndex] <- metaCacheEntry{name: "dir/"}
+			for _, c := range in {
+				close(c)
+			}
+			out := make(chan metaCacheEntry, 2)
+			calls := 0
+			err := mergeEntryChannelsWithResolver(t.Context(), in, out, 1, nil, func(group []*metaCacheEntry) (*metaCacheEntry, error) {
+				calls++
+				if len(group) != 1 || group[0].listingSet == nil {
+					t.Fatalf("wrong pending group: %+v", group)
+				}
+				return &metaCacheEntry{name: "dir/", metadata: []byte{1}}, nil
+			})
+			if err != nil || calls != 1 || len(out) != 1 {
+				t.Fatalf("err=%v recoveries=%d outputs=%d", err, calls, len(out))
+			}
+			if entry := <-out; entry.isDir() || !entry.isObjectDir() {
+				t.Fatalf("object became directory: %+v", entry)
+			}
+		})
+	}
+}
+
+func TestListingRecoveryRejectsUnexpectedSinglePoolSet(t *testing.T) {
+	sets := []*erasureObjects{{}, {}}
+	pool := &erasureSets{sets: sets, distributionAlgo: formatErasureVersionV3DistributionAlgoV3}
+	z := &erasureServerPools{serverPools: []*erasureSets{pool}}
+	for _, object := range []string{"key", "directory/"} {
+		target := pool.getHashedSet(encodeDirObject(object))
+		wrong := sets[0]
+		if wrong == target {
+			wrong = sets[1]
+		}
+		entry, err := z.resolveListingGroup(t.Context(), "bucket", []*metaCacheEntry{{name: object, listingSet: wrong}})
+		if entry != nil || !isErrReadQuorum(err) {
+			t.Fatalf("unexpected set became success: %v %v", entry, err)
+		}
+	}
+}
+
+func TestListingRecoveryReleasesBeforeOutput(t *testing.T) {
+	z, bucket, _ := nullQuorumBackend(t)
+	const object = "object/"
+	nullQuorumPut(t, z, bucket, object, []byte("data"), ObjectOptions{})
+	set := z.serverPools[0].sets[0]
+	input := make(chan metaCacheEntry, 1)
+	input <- metaCacheEntry{name: object, listingSet: set}
+	close(input)
+	out := make(chan metaCacheEntry)
+	resolved := make(chan struct{})
+	done := make(chan error, 1)
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+	defer cancel()
+	go func() {
+		done <- mergeEntryChannelsWithResolver(ctx, []chan metaCacheEntry{input}, out, 1, nil, func(group []*metaCacheEntry) (*metaCacheEntry, error) {
+			entry, err := z.resolveListingGroup(ctx, bucket, group)
+			close(resolved)
+			return entry, err
+		})
+	}()
+	select {
+	case <-resolved:
+	case <-ctx.Done():
+		t.Fatal(ctx.Err())
+	}
+	lock := set.NewNSLock(bucket, encodeDirObject(object))
+	lkctx, err := lock.GetLock(ctx, globalOperationTimeout)
+	if err != nil {
+		t.Fatalf("writer blocked by output backpressure: %v", err)
+	}
+	lock.Unlock(lkctx)
+	entry := <-out
+	if entry.name != object || !entry.isObjectDir() || entry.isDir() || entry.listingSet != nil {
+		t.Fatalf("directory object identity lost: %+v", entry)
+	}
+	if err := <-done; err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestListingRecoveryTerminalErrorPrecedesLock(t *testing.T) {
+	for _, n := range []int{1, 2} {
+		t.Run(fmt.Sprint(n), func(t *testing.T) {
+			inputs := make([]chan metaCacheEntry, n)
+			for i := range inputs {
+				inputs[i] = make(chan metaCacheEntry, 1)
+				inputs[i] <- metaCacheEntry{name: "key", listingSet: &erasureObjects{}}
+				close(inputs[i])
+			}
+			out := make(chan metaCacheEntry, 1)
+			want := errors.New("walk failed")
+			called := false
+			err := mergeEntryChannelsWithResolver(t.Context(), inputs, out, 1, func(int) error { return want }, func([]*metaCacheEntry) (*metaCacheEntry, error) { called = true; return nil, nil })
+			if !errors.Is(err, want) || called || len(out) != 0 {
+				t.Fatalf("err=%v recovery=%v output=%d", err, called, len(out))
+			}
+		})
+	}
+}
+
+func TestListingPendingCannotBeSerialized(t *testing.T) {
+	entry := metaCacheEntry{name: "directory/", listingSet: &erasureObjects{}}
+	if entry.isDir() || !entry.isObjectDir() {
+		t.Fatal("pending object became a directory")
+	}
+	var buf bytes.Buffer
+	w := newMetacacheWriter(&buf, 0)
+	if err := w.write(entry); err == nil {
+		t.Fatal("pending entry serialized")
+	}
+	_ = w.Close()
+	buf.Reset()
+	w = newMetacacheWriter(&buf, 0)
+	stream, err := w.stream()
+	if err != nil {
+		t.Fatal(err)
+	}
+	stream <- entry
+	close(stream)
+	if err = w.Close(); err == nil {
+		t.Fatal("stream serialized pending entry")
+	}
+}
+
+func TestListingEntryRepresentationCost(t *testing.T) {
+	type before struct {
+		_ string    // name
+		_ []byte    // metadata
+		_ *xlMetaV2 // cached
+		_ bool      // reusable
+	}
+	t.Logf("metaCacheEntry before=%d after=%d bytes", unsafe.Sizeof(before{}), unsafe.Sizeof(metaCacheEntry{}))
+}
+
+func BenchmarkListingMergeNormal(b *testing.B) {
+	for _, n := range []int{1, 2} {
+		for _, resolver := range []bool{false, true} {
+			b.Run(fmt.Sprintf("inputs=%d/resolver=%v", n, resolver), func(b *testing.B) {
+				b.ReportAllocs()
+				for b.Loop() {
+					inputs := make([]chan metaCacheEntry, n)
+					for i := range inputs {
+						inputs[i] = make(chan metaCacheEntry, 100)
+						for j := range 100 {
+							inputs[i] <- metaCacheEntry{name: fmt.Sprintf("key-%03d-%d/", j, i)}
+						}
+						close(inputs[i])
+					}
+					out := make(chan metaCacheEntry, n*100)
+					var resolveGroup func([]*metaCacheEntry) (*metaCacheEntry, error)
+					if resolver {
+						resolveGroup = func([]*metaCacheEntry) (*metaCacheEntry, error) { return nil, io.ErrUnexpectedEOF }
+					}
+					if err := mergeEntryChannelsWithResolver(context.Background(), inputs, out, 1, nil, resolveGroup); err != nil {
+						b.Fatal(err)
+					}
+				}
+			})
+		}
+	}
+}

cmd/metacache-null-quorum_test.go

@@ -29,6 +29,7 @@ import (
 	"path/filepath"
 	"slices"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -52,6 +53,37 @@ type nullQuorumWalkDisk struct {
 	index    int
 }
 
+type failedMetadataWalkDisk struct {
+	StorageAPI
+}
+
+func (d *failedMetadataWalkDisk) WalkDir(context.Context, WalkDirOptions, io.Writer) error {
+	return errDiskNotFound
+}
+
+// offlineNullQuorumDisk models a drive of a node that is away.
+type offlineNullQuorumDisk struct {
+	StorageAPI
+}
+
+func (d *offlineNullQuorumDisk) IsOnline() bool { return false }
+
+func (d *offlineNullQuorumDisk) DiskInfo(context.Context, DiskInfoOptions) (DiskInfo, error) {
+	return DiskInfo{}, errDiskNotFound
+}
+
+func (d *offlineNullQuorumDisk) WalkDir(context.Context, WalkDirOptions, io.Writer) error {
+	return errDiskNotFound
+}
+
+func (d *offlineNullQuorumDisk) ReadXL(context.Context, string, string, bool) (RawFileInfo, error) {
+	return RawFileInfo{}, errDiskNotFound
+}
+
+func (d *offlineNullQuorumDisk) ReadVersion(context.Context, string, string, string, string, ReadOptions) (FileInfo, error) {
+	return FileInfo{}, errDiskNotFound
+}
+
 func (d *nullQuorumWalkDisk) WalkDir(ctx context.Context, opts WalkDirOptions, out io.Writer) error {
 	if opts.Bucket != d.schedule.bucket {
 		return d.StorageAPI.WalkDir(ctx, opts, out)
@@ -119,6 +151,30 @@ func nullQuorumRequest(t *testing.T, router http.Handler, method, target string)
 	return rec
 }
 
+func nullQuorumPut(t *testing.T, z *erasureServerPools, bucket, object string, body []byte, opts ObjectOptions) {
+	t.Helper()
+	if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), opts); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func nullQuorumListV2Keys(t *testing.T, router http.Handler, bucket string) (int, []string) {
+	t.Helper()
+	rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
+	if rec.Code != http.StatusOK {
+		return rec.Code, nil
+	}
+	var list ListObjectsV2Response
+	if err := xml.Unmarshal(rec.Body.Bytes(), &list); err != nil {
+		t.Fatal(err)
+	}
+	keys := make([]string, 0, len(list.Contents))
+	for _, object := range list.Contents {
+		keys = append(keys, object.Key)
+	}
+	return rec.Code, keys
+}
+
 func TestListObjectsSingleNullQuorumHTTP(t *testing.T) {
 	z, bucket, router := nullQuorumBackend(t)
 	oldBody := bytes.Repeat([]byte{'a'}, 8192)
@@ -227,6 +283,400 @@ func TestListObjectsSingleNullQuorumHTTP(t *testing.T) {
 	}
 }
 
+func TestListObjectsFallsBackToReadableMetadata(t *testing.T) {
+	z, bucket, router := nullQuorumBackend(t)
+	set := z.serverPools[0].sets[0]
+	disks, switchDisks := installListingDiskSwitch(set)
+	const object = "object"
+	oldBody := bytes.Repeat([]byte{'a'}, 8192)
+	newBody := bytes.Repeat([]byte{'b'}, 8192)
+	oldTime := time.Now().UTC().Add(-time.Hour)
+
+	if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(oldBody), int64(len(oldBody)), "", ""), ObjectOptions{MTime: oldTime}); err != nil {
+		t.Fatal(err)
+	}
+	oldMeta := make([][]byte, len(disks))
+	for i, disk := range disks {
+		oldMeta[i] = mustReadNullQuorumMeta(t, disk, bucket, object)
+	}
+	if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(newBody), int64(len(newBody)), "", ""), ObjectOptions{MTime: oldTime.Add(time.Minute)}); err != nil {
+		t.Fatal(err)
+	}
+
+	// The eight readable walk streams are split 3/5 across two generations;
+	// the other eight streams fail. Neither generation is proven readable by
+	// the listing inputs, so returning HTTP 200 would silently omit the key.
+	for i := range 3 {
+		if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, oldMeta[i]); err != nil {
+			t.Fatal(err)
+		}
+	}
+	wrapped := make([]StorageAPI, len(disks))
+	for i, disk := range disks {
+		wrapped[i] = disk
+		if i >= 8 {
+			wrapped[i] = &failedMetadataWalkDisk{StorageAPI: disk}
+		}
+	}
+	switchDisks(wrapped)
+
+	rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
+	var list ListObjectsV2Response
+	if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil {
+		t.Fatalf("LIST failed to resolve readable metadata: %d %s", rec.Code, rec.Body.String())
+	}
+	if list.KeyCount != 1 || len(list.Contents) != 1 || list.Contents[0].Key != object {
+		t.Fatalf("LIST omitted the readable key: %+v", list)
+	}
+
+	// The object itself still has a read quorum outside the failed walk
+	// streams, demonstrating why the LIST must report uncertainty, not absence.
+	get := nullQuorumRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object))
+	if get.Code != http.StatusOK || !bytes.Equal(get.Body.Bytes(), newBody) {
+		t.Fatalf("GET lost the readable generation: %d %q", get.Code, get.Body.String())
+	}
+	// Walk uses a separate producer but must share the same latest-only pool
+	// recovery. Versioned walks retain their existing no-fallback behavior.
+	for _, latestOnly := range []bool{false, true} {
+		out := make(chan itemOrErr[ObjectInfo], 10)
+		if err := z.Walk(t.Context(), bucket, "", out, WalkOptions{LatestOnly: latestOnly, AskDisks: "strict"}); err != nil {
+			t.Fatal(err)
+		}
+		var objects []ObjectInfo
+		for result := range out {
+			if result.Err != nil {
+				t.Fatal(result.Err)
+			}
+			objects = append(objects, result.Item)
+		}
+		if latestOnly {
+			if len(objects) != 1 || objects[0].Name != object || !objects[0].ModTime.Equal(oldTime.Add(time.Minute)) {
+				t.Fatalf("Walk latest recovery: %+v", objects)
+			}
+		} else if len(objects) != 0 {
+			t.Fatalf("versioned Walk unexpectedly enabled recovery: %+v", objects)
+		}
+	}
+	failedDisks := slices.Clone(wrapped)
+	for i, disk := range failedDisks {
+		failedDisks[i] = &listingReadFailDisk{StorageAPI: disk, object: object, reads: &atomic.Int64{}}
+	}
+	switchDisks(failedDisks)
+	out := make(chan itemOrErr[ObjectInfo], 10)
+	if err := z.Walk(t.Context(), bucket, "", out, WalkOptions{LatestOnly: true, AskDisks: "strict"}); err != nil {
+		t.Fatal(err)
+	}
+	failed := false
+	for result := range out {
+		if result.Err == nil {
+			t.Fatalf("failed Walk yielded object: %+v", result.Item)
+		}
+		if isErrReadQuorum(result.Err) {
+			failed = true
+		}
+	}
+	if !failed {
+		t.Fatal("failed Walk recovery did not report quorum error")
+	}
+}
+
+// The fallback must read under the object read lock. A lock-free read of an
+// object being overwritten stably reports a live object as absent, which is
+// the omission this repair exists to remove. Hold the write lock and assert
+// the listing waits for it rather than resolving straight through.
+func TestListFallbackTakesObjectReadLock(t *testing.T) {
+	z, bucket, router := nullQuorumBackend(t)
+	set := z.serverPools[0].sets[0]
+	disks := set.getDisks()
+	const object = "object"
+	oldBody, newBody := bytes.Repeat([]byte{'a'}, 8192), bytes.Repeat([]byte{'b'}, 8192)
+	oldTime := time.Now().UTC().Add(-time.Hour)
+
+	if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(oldBody), int64(len(oldBody)), "", ""), ObjectOptions{MTime: oldTime}); err != nil {
+		t.Fatal(err)
+	}
+	oldMeta := make([][]byte, len(disks))
+	for i, disk := range disks {
+		oldMeta[i] = mustReadNullQuorumMeta(t, disk, bucket, object)
+	}
+	if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(newBody), int64(len(newBody)), "", ""), ObjectOptions{MTime: oldTime.Add(time.Minute)}); err != nil {
+		t.Fatal(err)
+	}
+	// Split the readable walk streams so the entry cannot be resolved from
+	// the snapshots and the fallback has to run.
+	for i := range 3 {
+		if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, oldMeta[i]); err != nil {
+			t.Fatal(err)
+		}
+	}
+	wrapped := make([]StorageAPI, len(disks))
+	for i, disk := range disks {
+		wrapped[i] = disk
+		if i >= 8 {
+			wrapped[i] = &failedMetadataWalkDisk{StorageAPI: disk}
+		}
+	}
+	set.getDisks = func() []StorageAPI { return wrapped }
+
+	lock := set.NewNSLock(bucket, object)
+	lkctx, err := lock.GetLock(t.Context(), globalOperationTimeout)
+	if err != nil {
+		t.Fatal(err)
+	}
+	locked := true
+	defer func() {
+		if locked {
+			lock.Unlock(lkctx)
+		}
+	}()
+	target := getListObjectsV2URL("", bucket, "", "1000", "", "", "")
+	req, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	rec := httptest.NewRecorder()
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		router.ServeHTTP(rec, req)
+	}()
+	waitForListingReadLock(t, set.nsMutex, bucket, object, done)
+	lock.Unlock(lkctx)
+	locked = false
+	<-done
+
+	var list ListObjectsV2Response
+	if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil {
+		t.Fatalf("LIST failed after the lock was released: %d %s", rec.Code, rec.Body.String())
+	}
+	if list.KeyCount != 1 || len(list.Contents) != 1 || list.Contents[0].Key != object {
+		t.Fatalf("LIST omitted the readable key: %+v", list)
+	}
+}
+
+// A truncated xl.meta is what a concurrent overwrite looks like on one drive,
+// and the walker skips that entry. Skipping is safe and deliberate: the key is
+// only omitted when no generation reaches quorum, which listPath now resolves
+// against the read path. Pin that, so the walker is not "hardened" into
+// failing a whole listing over a minority of unreadable drives.
+func TestListObjectsSurvivesTruncatedMetadataMinority(t *testing.T) {
+	z, bucket, router := nullQuorumBackend(t)
+	set := z.serverPools[0].sets[0]
+	disks := set.getDisks()
+	const object = "object"
+	body := bytes.Repeat([]byte{'a'}, 8192)
+
+	if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+
+	// Truncate xl.meta on a minority of drives, as an overwrite in flight does.
+	for i := range 4 {
+		meta := mustReadNullQuorumMeta(t, disks[i], bucket, object)
+		if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, meta[:len(meta)/2]); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
+	var list ListObjectsV2Response
+	if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil {
+		t.Fatalf("a truncated minority failed the listing: %d %s", rec.Code, rec.Body.String())
+	}
+	if list.KeyCount != 1 || len(list.Contents) != 1 || list.Contents[0].Key != object {
+		t.Fatalf("a truncated minority omitted the key: %+v", list)
+	}
+}
+
+func TestListStaleMinorityIgnored(t *testing.T) {
+	z, bucket, router := nullQuorumBackend(t)
+	disks := z.serverPools[0].sets[0].getDisks()
+	body := bytes.Repeat([]byte{'a'}, 8192)
+	nullQuorumPut(t, z, bucket, "gone", body, ObjectOptions{})
+	nullQuorumPut(t, z, bucket, "keep", body, ObjectOptions{})
+	for _, disk := range disks[4:] {
+		if err := disk.Delete(t.Context(), bucket, "gone", DeleteOptions{Recursive: true, Immediate: true}); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	if code, keys := nullQuorumListV2Keys(t, router, bucket); code != http.StatusOK || !slices.Equal(keys, []string{"keep"}) {
+		t.Errorf("ListObjectsV2: got %d %v; want 200 [keep]", code, keys)
+	}
+	results := make(chan itemOrErr[ObjectInfo], 16)
+	if err := z.Walk(t.Context(), bucket, "", results, WalkOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	var names []string
+	for result := range results {
+		if result.Err != nil {
+			t.Fatal(result.Err)
+		}
+		names = append(names, result.Item.Name)
+	}
+	if !slices.Equal(names, []string{"keep"}) {
+		t.Errorf("Walk: got %v; want [keep]", names)
+	}
+
+	rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectVersionsURL("", bucket, "", "1000", ""))
+	var list struct {
+		Versions []struct {
+			Key string `xml:"Key"`
+		} `xml:"Version"`
+	}
+	if rec.Code != http.StatusOK {
+		t.Errorf("ListObjectVersions: got %d; want 200", rec.Code)
+	} else if err := xml.Unmarshal(rec.Body.Bytes(), &list); err != nil {
+		t.Fatal(err)
+	} else if len(list.Versions) != 1 || list.Versions[0].Key != "keep" {
+		t.Errorf("ListObjectVersions: got %+v; want [keep]", list.Versions)
+	}
+
+	if _, err := z.DeleteObject(t.Context(), bucket, "keep", ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	if err := z.DeleteBucket(t.Context(), bucket, DeleteBucketOptions{}); err != nil {
+		t.Errorf("DeleteBucket with only a stale minority: %v", err)
+	}
+}
+
+func TestIAMListingIgnoresStaleMinority(t *testing.T) {
+	z, _, _ := nullQuorumBackend(t)
+	body := []byte(`{"version":1}`)
+	alive := iamConfigUsersPrefix + "alive/identity.json"
+	ghost := iamConfigUsersPrefix + "ghost/identity.json"
+	nullQuorumPut(t, z, minioMetaBucket, alive, body, ObjectOptions{})
+	nullQuorumPut(t, z, minioMetaBucket, ghost, body, ObjectOptions{})
+	for _, disk := range z.serverPools[0].getHashedSet(ghost).getDisks()[4:] {
+		if err := disk.Delete(t.Context(), minioMetaBucket, ghost, DeleteOptions{Recursive: true, Immediate: true}); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	items, err := newIAMObjectStore(z, MinIOUsersSysType).listAllIAMConfigItems(t.Context())
+	if err != nil {
+		t.Fatalf("IAM listing failed on a stale minority (retriable=%v): %v", configRetriableErrors(err), err)
+	}
+	if got := items["users/"]; !slices.Equal(got, []string{"alive/identity.json"}) {
+		t.Errorf("IAM users: got %v; want [alive/identity.json]", got)
+	}
+}
+
+func TestListSplitDirectoryObject(t *testing.T) {
+	z, bucket, router := nullQuorumBackend(t)
+	set := z.serverPools[0].sets[0]
+	disks := set.getDisks()
+	const object = "dir/"
+	encoded := encodeDirObject(object)
+	mtime := time.Now().UTC().Add(-time.Hour)
+
+	nullQuorumPut(t, z, bucket, object, nil, ObjectOptions{MTime: mtime})
+	old := make([][]byte, 3)
+	for i := range old {
+		old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, encoded)
+	}
+	nullQuorumPut(t, z, bucket, object, nil, ObjectOptions{MTime: mtime.Add(time.Minute)})
+	for i := range old {
+		if err := disks[i].WriteAll(t.Context(), bucket, encoded+"/"+xlStorageFormatFile, old[i]); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	walkers := slices.Clone(disks)
+	for i := 8; i < len(walkers); i++ {
+		walkers[i] = &failedMetadataWalkDisk{StorageAPI: disks[i]}
+	}
+	set.getDisks = func() []StorageAPI { return walkers }
+	if code, keys := nullQuorumListV2Keys(t, router, bucket); code != http.StatusOK || !slices.Equal(keys, []string{object}) {
+		t.Errorf("ListObjectsV2: got %d %v; want 200 [%s]", code, keys, object)
+	}
+}
+
+func TestListFallbackConfirmsCurrentAbsence(t *testing.T) {
+	z, bucket, _ := nullQuorumBackend(t)
+	set := z.serverPools[0].sets[0]
+	disks := set.getDisks()
+	const object = "object"
+	body := bytes.Repeat([]byte{'a'}, 8192)
+	mtime := time.Now().UTC().Add(-time.Hour)
+
+	nullQuorumPut(t, z, bucket, object, body, ObjectOptions{MTime: mtime})
+	old := make([][]byte, len(disks))
+	for i := range disks {
+		old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, object)
+	}
+	nullQuorumPut(t, z, bucket, object, body, ObjectOptions{MTime: mtime.Add(time.Minute)})
+	newer := make([][]byte, len(disks))
+	for i := range disks {
+		newer[i] = mustReadNullQuorumMeta(t, disks[i], bucket, object)
+		if err := disks[i].Delete(t.Context(), bucket, object, DeleteOptions{Recursive: true, Immediate: true}); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	entries := make(metaCacheEntries, len(disks))
+	errs := make([]error, len(disks))
+	for i := range 3 {
+		entries[i] = metaCacheEntry{name: object, metadata: old[i]}
+	}
+	for i := 3; i < 8; i++ {
+		entries[i] = metaCacheEntry{name: object, metadata: newer[i]}
+	}
+	for i := 8; i < len(errs); i++ {
+		errs[i] = errDiskNotFound
+	}
+	resolver := metadataResolutionParams{dirQuorum: 8, objQuorum: 8, bucket: bucket, requestedVersions: 1}
+	entry, err := z.resolveListingGroup(t.Context(), bucket, []*metaCacheEntry{set.resolveListEntry(entries, errs, &resolver)})
+	if entry != nil || err != nil {
+		t.Fatalf("strict current absence: got entry=%v err=%v; want no entry or error", entry, err)
+	}
+}
+
+// G0 was written 8+8 while a node was away. G1 was later written 12+4 but
+// missed drive 0, which still holds G0. Then a node holding four G1 drives
+// goes away: eleven G1 and one G0 remain online. G1 exists but lacks one of
+// its twelve data shards until the node returns, so HEAD must not report 404
+// and LIST must still show the key.
+func TestListCrossParityMinorityStaysVisible(t *testing.T) {
+	z, bucket, router := nullQuorumBackend(t)
+	set := z.serverPools[0].sets[0]
+	disks, switchDisks := installListingDiskSwitch(set)
+	const object = "object"
+	mtime := time.Now().UTC().Add(-time.Hour)
+
+	nullQuorumPut(t, z, bucket, object, bytes.Repeat([]byte{'a'}, 256<<10), ObjectOptions{MTime: mtime, MaxParity: true})
+	g0 := mustReadNullQuorumMeta(t, disks[0], bucket, object)
+	var x xlMetaV2
+	if err := x.LoadOrConvert(g0); err != nil {
+		t.Fatal(err)
+	}
+	if h := x.versions[0].header; h.EcM != 8 || h.EcN != 8 {
+		t.Fatalf("G0 is %d+%d, want 8+8", h.EcM, h.EcN)
+	}
+	g1Body := bytes.Repeat([]byte{'b'}, 256<<10)
+	nullQuorumPut(t, z, bucket, object, g1Body, ObjectOptions{MTime: mtime.Add(time.Minute)})
+	if err := disks[0].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, g0); err != nil {
+		t.Fatal(err)
+	}
+
+	online := slices.Clone(disks)
+	for i := 12; i < len(online); i++ {
+		online[i] = &offlineNullQuorumDisk{StorageAPI: disks[i]}
+	}
+	switchDisks(online)
+	if code := nullQuorumRequest(t, router, http.MethodHead, getHeadObjectURL("", bucket, object)).Code; code == http.StatusNotFound {
+		t.Errorf("HEAD with a node away: 404 for an existing object")
+	}
+	if code, keys := nullQuorumListV2Keys(t, router, bucket); code != http.StatusOK || !slices.Equal(keys, []string{object}) {
+		t.Errorf("ListObjectsV2 with a node away: got %d %v; want 200 [%s]", code, keys, object)
+	}
+
+	switchDisks(disks)
+	if get := nullQuorumRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object)); get.Code != http.StatusOK || !bytes.Equal(get.Body.Bytes(), g1Body) {
+		t.Fatalf("GET after the node returns: %d", get.Code)
+	}
+}
+
 func TestSingleNullQuorumReadAndConsumers(t *testing.T) {
 	z, bucket, router := nullQuorumBackend(t)
 	set := z.serverPools[0].sets[0]

cmd/metacache-parts-retry_test.go

@@ -0,0 +1,444 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// 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"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"slices"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/minio/minio/internal/grid"
+)
+
+// Read real cache objects and use an actual peer RPC, counting only the block
+// under test. Gates control publication, never replace quorum or EOF decisions.
+type listingCachePartsProbe struct {
+	state    atomic.Pointer[metacache]
+	rpcFails atomic.Bool
+	reads    atomic.Int64
+	mu       sync.Mutex
+	probes   []int
+	checks   []time.Time
+	onProbe  func(int) bool
+	onCheck  func(int)
+	target   string
+	readErr  error
+}
+
+func (p *listingCachePartsProbe) setState(o listPathOptions, failed bool) {
+	m := o.newMetacache()
+	m.lastUpdate = time.Now()
+	if failed {
+		m.status, m.error = scanStateError, "cache producer failed"
+	} else {
+		m.status = scanStateStarted
+	}
+	p.state.Store(&m)
+}
+
+func (p *listingCachePartsProbe) snapshot() ([]int, []time.Time) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return slices.Clone(p.probes), slices.Clone(p.checks)
+}
+
+type listingCachePartsDisk struct {
+	StorageAPI
+	index int
+	probe *listingCachePartsProbe
+}
+
+func (d *listingCachePartsDisk) ReadXL(ctx context.Context, bucket, object string, readData bool) (RawFileInfo, error) {
+	if bucket == minioMetaBucket && object == d.probe.target {
+		d.probe.reads.Add(1)
+		if d.probe.readErr != nil {
+			return RawFileInfo{}, d.probe.readErr
+		}
+	}
+	return d.StorageAPI.ReadXL(ctx, bucket, object, readData)
+}
+
+func (d *listingCachePartsDisk) ReadVersion(ctx context.Context, origVolume, volume, path, versionID string, opts ReadOptions) (FileInfo, error) {
+	if volume == minioMetaBucket && path == d.probe.target {
+		d.probe.mu.Lock()
+		d.probe.probes = append(d.probe.probes, d.index)
+		d.probe.mu.Unlock()
+		if d.probe.onProbe != nil && d.probe.onProbe(d.index) {
+			return FileInfo{}, errFileNotFound
+		}
+	}
+	return d.StorageAPI.ReadVersion(ctx, origVolume, volume, path, versionID, opts)
+}
+
+func listingCachePartsBackend(t *testing.T, part int) (*erasureObjects, listPathOptions, *listingCachePartsProbe, func([]StorageAPI)) {
+	t.Helper()
+	z, bucket, _ := nullQuorumBackend(t)
+	set := z.serverPools[0].sets[0]
+	o := listPathOptions{ID: mustGetUUID(), Bucket: bucket, Limit: 100, Recursive: true, IncludeDirectories: true}
+	p := &listingCachePartsProbe{target: o.objectPath(part)}
+	p.setState(o, false)
+	disks, switchDisks := installListingDiskSwitch(set)
+	wrapped := slices.Clone(disks)
+	for i, disk := range disks {
+		wrapped[i] = &listingCachePartsDisk{StorageAPI: disk, index: i, probe: p}
+	}
+	switchDisks(wrapped)
+
+	tg, err := grid.SetupTestGrid(2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	previous := globalNotificationSys
+	t.Cleanup(func() {
+		globalNotificationSys = previous
+		tg.Cleanup()
+	})
+	if err := getMetacacheListingRPC.Register(tg.Managers[1], func(got *listPathOptions) (*metacache, *grid.RemoteErr) {
+		if got.ID != o.ID || got.Create {
+			return nil, grid.NewRemoteErr(errors.New("invalid cache state lookup"))
+		}
+		p.mu.Lock()
+		p.checks = append(p.checks, time.Now())
+		n := len(p.checks)
+		p.mu.Unlock()
+		if p.onCheck != nil {
+			p.onCheck(n)
+		}
+		if p.rpcFails.Load() {
+			return nil, grid.NewRemoteErr(errors.New("cache state RPC failed"))
+		}
+		m := *p.state.Load()
+		return &m, nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	client := &peerRESTClient{gridConn: func() *grid.Connection { return tg.Managers[0].Connection(tg.Hosts[1]) }}
+	globalNotificationSys = &NotificationSys{peerClients: []*peerRESTClient{client}, allPeerClients: []*peerRESTClient{client}}
+	return set, o, p, switchDisks
+}
+
+func putListingCachePart(t *testing.T, set *erasureObjects, o listPathOptions, part int, name string, eos bool) {
+	t.Helper()
+	var buf bytes.Buffer
+	w := newMetacacheWriter(&buf, 0)
+	if err := w.write(metaCacheEntry{name: name}); err != nil {
+		t.Fatal(err)
+	}
+	if err := w.Close(); err != nil {
+		t.Fatal(err)
+	}
+	block := metacacheBlock{n: part, First: name, Last: name, EOS: eos}
+	_, err := set.putMetacacheObject(t.Context(), o.objectPath(part), mustGetPutObjReader(t, bytes.NewReader(buf.Bytes()), int64(buf.Len()), "", ""), ObjectOptions{UserDefined: block.headerKV()})
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestStreamMetadataPartsHealthyNoStateRPC(t *testing.T) {
+	set, o, p, _ := listingCachePartsBackend(t, 1)
+	putListingCachePart(t, set, o, 0, "a/", false)
+	putListingCachePart(t, set, o, 1, "b/", true)
+	entries, err := set.streamMetadataParts(t.Context(), o)
+	if !errors.Is(err, io.EOF) || entries.len() != 2 || entries.o[0].name != "a/" || entries.o[1].name != "b/" {
+		t.Fatalf("healthy two-block read: entries=%v err=%v", entries.o, err)
+	}
+	probes, checks := p.snapshot()
+	if len(probes) != 0 || len(checks) != 0 {
+		t.Fatalf("healthy read added probes=%v state RPCs=%d", probes, len(checks))
+	}
+	t.Logf("healthy persisted blocks: state RPCs=0 retry probes=0 target ReadXL=%d", p.reads.Load())
+}
+
+func TestStreamMetadataPartsTerminalError(t *testing.T) {
+	for _, part := range []int{0, 1} {
+		t.Run(fmt.Sprintf("part=%d", part), func(t *testing.T) {
+			set, o, p, _ := listingCachePartsBackend(t, part)
+			if part > 0 {
+				putListingCachePart(t, set, o, 0, "a/", false)
+			}
+			p.setState(o, true)
+			start := time.Now()
+			entries, err := set.streamMetadataParts(t.Context(), o)
+			elapsed := time.Since(start)
+			if err == nil || errors.Is(err, io.EOF) || !strings.Contains(err.Error(), "cache producer failed") || entries.len() != part {
+				t.Fatalf("failed cache became successful: entries=%v err=%v", entries.o, err)
+			}
+			probes, checks := p.snapshot()
+			if len(checks) != 1 || len(probes) != 0 || elapsed > time.Second {
+				t.Fatalf("terminal error waited for disk retries: elapsed=%v state RPCs=%d probes=%v", elapsed, len(checks), probes)
+			}
+			t.Logf("terminal part=%d elapsed=%v state RPCs=%d probes=%d target ReadXL=%d", part, elapsed, len(checks), len(probes), p.reads.Load())
+		})
+	}
+}
+
+func TestStreamMetadataPartsLateTerminalError(t *testing.T) {
+	set, o, p, _ := listingCachePartsBackend(t, 1)
+	putListingCachePart(t, set, o, 0, "a/", false)
+	p.onProbe = func(int) bool {
+		p.setState(o, true)
+		return false
+	}
+	start := time.Now()
+	entries, err := set.streamMetadataParts(t.Context(), o)
+	elapsed := time.Since(start)
+	probes, checks := p.snapshot()
+	if err == nil || errors.Is(err, io.EOF) || !strings.Contains(err.Error(), "cache producer failed") || entries.len() != 1 {
+		t.Fatalf("late producer error lost: entries=%v err=%v", entries.o, err)
+	}
+	if elapsed > time.Second || len(probes) != 2 || !slices.Equal(probes, []int{0, 1}) || len(checks) != 2 {
+		t.Fatalf("late error retry work: elapsed=%v probes=%v state RPCs=%d", elapsed, probes, len(checks))
+	}
+	if checks[1].Sub(checks[0]) < 250*time.Millisecond || p.reads.Load() > 16 {
+		t.Fatalf("unbounded retry: state interval=%v target ReadXL=%d", checks[1].Sub(checks[0]), p.reads.Load())
+	}
+	t.Logf("late terminal elapsed=%v state RPCs=%d probes=%v target ReadXL=%d", elapsed, len(checks), probes, p.reads.Load())
+}
+
+type listingCachePartsResult struct {
+	entries metaCacheEntriesSorted
+	err     error
+}
+
+func startListingCacheParts(t *testing.T, set *erasureObjects, o listPathOptions) (<-chan listingCachePartsResult, context.CancelFunc) {
+	t.Helper()
+	ctx, cancel := context.WithCancel(t.Context())
+	done := make(chan listingCachePartsResult, 1)
+	joined := make(chan struct{})
+	go func() {
+		defer close(joined)
+		entries, err := set.streamMetadataParts(ctx, o)
+		done <- listingCachePartsResult{entries: entries, err: err}
+	}()
+	t.Cleanup(func() {
+		cancel()
+		select {
+		case <-joined:
+		case <-time.After(8 * time.Second):
+			t.Error("cache continuation worker did not finish")
+		}
+	})
+	return done, cancel
+}
+
+func receiveListingCacheParts(t *testing.T, done <-chan listingCachePartsResult) listingCachePartsResult {
+	t.Helper()
+	select {
+	case result := <-done:
+		return result
+	case <-time.After(8 * time.Second):
+		t.Fatal("cache continuation did not finish")
+		return listingCachePartsResult{}
+	}
+}
+
+func TestStreamMetadataPartsGeneratingThenPublished(t *testing.T) {
+	set, o, p, _ := listingCachePartsBackend(t, 1)
+	putListingCachePart(t, set, o, 0, "a/", false)
+	checked := make(chan struct{})
+	p.onCheck = func(n int) {
+		if n == 1 {
+			close(checked)
+		}
+	}
+	// The first selected replica is absent even after the producer publishes
+	// the block elsewhere. Rotation must reach a readable replica.
+	p.onProbe = func(index int) bool { return index == 0 }
+	done, _ := startListingCacheParts(t, set, o)
+	select {
+	case <-checked:
+	case <-time.After(5 * time.Second):
+		t.Fatal("missing block did not check producer state")
+	}
+	select {
+	case result := <-done:
+		t.Fatalf("still-generating cache finished early: %+v", result)
+	case <-time.After(100 * time.Millisecond):
+	}
+	putListingCachePart(t, set, o, 1, "b/", true)
+	result := receiveListingCacheParts(t, done)
+	if !errors.Is(result.err, io.EOF) || result.entries.len() != 2 || result.entries.o[0].name != "a/" || result.entries.o[1].name != "b/" {
+		t.Fatalf("published second block lost: entries=%v err=%v", result.entries.o, result.err)
+	}
+	probes, checks := p.snapshot()
+	if len(probes) < 2 || probes[0] != 0 || probes[1] != 1 {
+		t.Fatalf("retry did not rotate: probes=%v checks=%d", probes, len(checks))
+	}
+	t.Logf("later block recovered: state RPCs=%d probes=%v target ReadXL=%d", len(checks), probes, p.reads.Load())
+}
+
+func TestStreamMetadataPartsRPCFailure(t *testing.T) {
+	set, o, p, _ := listingCachePartsBackend(t, 1)
+	putListingCachePart(t, set, o, 0, "a/", false)
+	p.rpcFails.Store(true)
+	entries, err := set.streamMetadataParts(t.Context(), o)
+	if err == nil || errors.Is(err, io.EOF) || !strings.Contains(err.Error(), "cache state RPC failed") || entries.len() != 1 {
+		t.Fatalf("RPC failure became successful EOF: entries=%v err=%v", entries.o, err)
+	}
+	probes, checks := p.snapshot()
+	if len(checks) != 1 || len(probes) != 0 {
+		t.Fatalf("state RPC failure continued retrying: probes=%v checks=%d", probes, len(checks))
+	}
+}
+
+func TestStreamMetadataPartsMissingIndex(t *testing.T) {
+	set, o, p, _ := listingCachePartsBackend(t, 0)
+	putListingCachePart(t, set, o, 0, "a/", false)
+	o.Marker = "b/"
+	p.setState(o, true)
+	entries, err := set.streamMetadataParts(t.Context(), o)
+	if err == nil || errors.Is(err, io.EOF) || !strings.Contains(err.Error(), "cache producer failed") || entries.len() != 0 {
+		t.Fatalf("unpublished index became successful EOF: entries=%v err=%v", entries.o, err)
+	}
+	probes, checks := p.snapshot()
+	if len(checks) != 1 || len(probes) != 0 {
+		t.Fatalf("missing index delayed terminal state: probes=%v checks=%d", probes, len(checks))
+	}
+}
+
+func TestStreamMetadataPartsOtherErrorRetriesOnce(t *testing.T) {
+	set, o, p, _ := listingCachePartsBackend(t, 0)
+	p.readErr = errVolumeNotFound
+	entries, err := set.streamMetadataParts(t.Context(), o)
+	if err == nil || errors.Is(err, io.EOF) || !strings.Contains(err.Error(), "reading first part metadata:") || entries.len() != 0 {
+		t.Fatalf("unexpected metadata error result: entries=%v err=%v", entries.o, err)
+	}
+	probes, checks := p.snapshot()
+	if p.reads.Load() != 32 || len(probes) != 0 || len(checks) != 1 {
+		t.Fatalf("other metadata error did not get exactly one full retry: ReadXL=%d probes=%v checks=%d", p.reads.Load(), probes, len(checks))
+	}
+}
+
+func TestStreamMetadataPartsMissingBlockWorkBound(t *testing.T) {
+	set, o, p, _ := listingCachePartsBackend(t, 0)
+	checked := make(chan struct{})
+	p.onCheck = func(n int) {
+		if n == 3 {
+			close(checked)
+		}
+	}
+	start := time.Now()
+	done, cancel := startListingCacheParts(t, set, o)
+	select {
+	case <-checked:
+	case <-time.After(5 * time.Second):
+		t.Fatal("missing cache stopped checking producer state")
+	}
+	cancel()
+	result := receiveListingCacheParts(t, done)
+	probes, checks := p.snapshot()
+	if !errors.Is(result.err, context.Canceled) || result.entries.len() != 0 {
+		t.Fatalf("generating cache became successful EOF: %+v", result)
+	}
+	if len(probes) != 3 || !slices.Equal(probes, []int{0, 1, 2}) || p.reads.Load() != 16 || len(checks) != 3 {
+		t.Fatalf("missing block exceeded work bound: probes=%v target ReadXL=%d checks=%d", probes, p.reads.Load(), len(checks))
+	}
+	for i := 1; i < len(checks); i++ {
+		// Receive timestamps allow a small amount of peer scheduling jitter.
+		if checks[i].Sub(checks[i-1]) < 240*time.Millisecond {
+			t.Fatalf("cache state queried too often: %v", checks)
+		}
+	}
+	t.Logf("still missing for %v: state RPCs=%d probes=%v target ReadXL=%d", time.Since(start), len(checks), probes, p.reads.Load())
+}
+
+func TestStreamMetadataPartsCancelRetryWait(t *testing.T) {
+	set, o, p, _ := listingCachePartsBackend(t, 1)
+	putListingCachePart(t, set, o, 0, "a/", false)
+	probed := make(chan struct{})
+	p.onProbe = func(index int) bool {
+		if index == 1 {
+			close(probed)
+		}
+		return false
+	}
+	done, cancel := startListingCacheParts(t, set, o)
+	select {
+	case <-probed:
+	case <-time.After(5 * time.Second):
+		t.Fatal("retry probe did not run")
+	}
+	// Let the probe return and enter the regular 250 ms wait before canceling.
+	time.Sleep(20 * time.Millisecond)
+	start := time.Now()
+	cancel()
+	result := receiveListingCacheParts(t, done)
+	elapsed := time.Since(start)
+	if !errors.Is(result.err, context.Canceled) || elapsed > 150*time.Millisecond {
+		t.Fatalf("retry wait ignored cancellation: elapsed=%v err=%v", elapsed, result.err)
+	}
+	t.Logf("cancel during retry wait: elapsed=%v", elapsed)
+}
+
+func TestStreamMetadataPartsChangedDisks(t *testing.T) {
+	for _, empty := range []bool{false, true} {
+		t.Run(fmt.Sprintf("empty=%v", empty), func(t *testing.T) {
+			testStreamMetadataPartsChangedDisks(t, empty)
+		})
+	}
+}
+
+func testStreamMetadataPartsChangedDisks(t *testing.T, empty bool) {
+	set, o, p, switchDisks := listingCachePartsBackend(t, 1)
+	putListingCachePart(t, set, o, 0, "a/", false)
+	disks := set.getDisks()
+	probeDone, restore := make(chan struct{}), make(chan struct{})
+	p.onProbe = func(index int) bool {
+		if index == 0 {
+			// A later probe must safely skip an empty/offline disk snapshot.
+			if empty {
+				switchDisks([]StorageAPI{})
+			} else {
+				switchDisks([]StorageAPI{nil})
+			}
+			close(probeDone)
+			return true
+		}
+		return false
+	}
+	p.onCheck = func(n int) {
+		if n == 2 {
+			close(restore)
+		}
+	}
+	done, _ := startListingCacheParts(t, set, o)
+	select {
+	case <-probeDone:
+	case <-time.After(5 * time.Second):
+		t.Fatal("initial retry probe did not run")
+	}
+	select {
+	case <-restore:
+	case <-time.After(5 * time.Second):
+		t.Fatal("empty disk snapshot prevented cache state checks")
+	}
+	switchDisks(disks)
+	putListingCachePart(t, set, o, 1, "b/", true)
+	result := receiveListingCacheParts(t, done)
+	if !errors.Is(result.err, io.EOF) || result.entries.len() != 2 {
+		t.Fatalf("changed disks did not recover: entries=%v err=%v", result.entries.o, result.err)
+	}
+}

cmd/metacache-server-pool.go

@@ -191,19 +191,19 @@ func (z *erasureServerPools) listPath(ctx context.Context, o *listPathOptions) (
 			return entries, err
 		}
 		entries.truncate(0)
-		go func() {
+		rpc := globalNotificationSys.restClientFromHash(pathJoin(o.Bucket, o.Prefix))
-			rpc := globalNotificationSys.restClientFromHash(pathJoin(o.Bucket, o.Prefix))
+		go func(o listPathOptions) {
 			if rpc != nil {
 				ctx, cancel := context.WithTimeout(GlobalContext, 5*time.Second)
 				defer cancel()
-				c, err := rpc.GetMetacacheListing(ctx, *o)
+				c, err := rpc.GetMetacacheListing(ctx, o)
 				if err == nil {
 					c.error = "no longer used"
 					c.status = scanStateError
 					rpc.UpdateMetacacheListing(ctx, *c)
 				}
 			}
-		}()
+		}(*o)
 		o.ID = ""
 	}
 
@@ -264,6 +264,176 @@ func (z *erasureServerPools) listPathShortcutEOF(ctx context.Context, bucket str
 	return io.EOF
 }
 
+// listingRLock covers the lock domains used by normal same-key mutations.
+// Multi-pool writes use the pools lock. In a single pool, DELETE/COPY use the
+// first set while PUT and metadata updates use the destination set, so recovery
+// takes both in that order. No writer routing or wire protocol changes here.
+func (z *erasureServerPools) listingRLock(ctx context.Context, bucket, object string, source *erasureObjects) (LockContext, func(), error) {
+	firstSet := z.serverPools[0].sets[0]
+	first := z.NewNSLock(bucket, object)
+	firstCtx, err := first.GetRLock(ctx, globalListingOperationTimeout)
+	if err != nil {
+		return LockContext{}, nil, err
+	}
+	if !z.SinglePool() || source == firstSet || (!source.nsMutex.isDistErasure && source.nsMutex == firstSet.nsMutex) {
+		return firstCtx, func() { first.RUnlock(firstCtx) }, nil
+	}
+	// Shared nsMutex pointers do not prove shared distributed locker sets.
+	second := source.NewNSLock(bucket, object)
+	secondCtx, err := second.GetRLock(firstCtx.Context(), globalListingOperationTimeout)
+	if err != nil {
+		first.RUnlock(firstCtx)
+		return LockContext{}, nil, err
+	}
+	return secondCtx, func() {
+		second.RUnlock(secondCtx)
+		first.RUnlock(firstCtx)
+	}, nil
+}
+
+// resolveListingGroup recovers a complete same-name group, never a private
+// per-set contribution. A strictly absent source requires current reads from
+// every pool, since another pool's earlier walk may predate a migration copy.
+func (z *erasureServerPools) resolveListingGroup(ctx context.Context, bucket string, group []*metaCacheEntry) (*metaCacheEntry, error) {
+	var source *erasureObjects
+	for _, entry := range group {
+		if entry.listingSet != nil {
+			source = entry.listingSet
+			break
+		}
+	}
+	if source == nil {
+		return nil, errInvalidArgument
+	}
+	// Recovery consumes this group. Return reusable walk buffers only after
+	// their contents have been copied into the newly encoded result.
+	defer func() {
+		for _, entry := range group {
+			if entry.reusable {
+				metaDataPoolPut(entry.metadata)
+			}
+		}
+	}()
+	name := group[0].name
+	object := encodeDirObject(name)
+	if z.SinglePool() {
+		target := z.serverPools[0].getHashedSet(object)
+		for _, entry := range group {
+			if entry.listingSet != nil && entry.listingSet != target {
+				// Unexpected placement is not evidence of absence at the
+				// current write target. Do not extend recovery to arbitrary sets.
+				return nil, listingReadQuorumError(bucket, object)
+			}
+		}
+	}
+	type readResult struct {
+		fi     FileInfo
+		absent bool
+	}
+	reads := make(map[*erasureObjects]readResult)
+	currentOnly := false
+	var current *FileInfo
+	err := func() error {
+		lockCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+		defer cancel()
+		lkctx, unlock, err := z.listingRLock(lockCtx, bucket, object, source)
+		if err != nil {
+			return err
+		}
+		defer unlock()
+		read := func(set *erasureObjects) (readResult, error) {
+			if result, ok := reads[set]; ok {
+				return result, nil
+			}
+			fi, absent, err := set.readListingFileInfo(lkctx.Context(), bucket, object)
+			result := readResult{fi: fi, absent: absent}
+			if err == nil {
+				reads[set] = result
+			}
+			return result, err
+		}
+		for _, entry := range group {
+			if entry.listingSet == nil {
+				continue
+			}
+			result, err := read(entry.listingSet)
+			if err != nil {
+				return err
+			}
+			currentOnly = currentOnly || result.absent
+		}
+		if !currentOnly {
+			return lkctx.Context().Err()
+		}
+		// Migration publishes the replacement through the pooled write lock
+		// before source cleanup. While this read lock is held, new publication
+		// cannot pass these reads; any concurrent cleanup already has a copy.
+		// Unknown pools still fail rather than being counted as absent.
+		for _, pool := range z.serverPools {
+			result, err := read(pool.getHashedSet(object))
+			if err != nil {
+				return err
+			}
+			if result.absent {
+				continue
+			}
+			// Match latest GET: newest ModTime, then lowest pool index.
+			if current == nil || result.fi.ModTime.After(current.ModTime) {
+				fi := result.fi
+				current = &fi
+			}
+		}
+		return lkctx.Context().Err()
+	}()
+	if err != nil {
+		return nil, err
+	}
+	// All FileInfo values above are owned decoded results. Encoding and merge
+	// allocation happen after RUnlock, as do all stream sends and cache RPCs.
+	encode := func(fi FileInfo) (*metaCacheEntry, error) {
+		xl := &xlMetaV2{}
+		if err := xl.AddVersion(fi); err != nil {
+			return nil, listingReadQuorumError(bucket, object)
+		}
+		entry := &metaCacheEntry{name: name, cached: xl, reusable: true}
+		var err error
+		entry.metadata, err = xl.AppendTo(metaDataPoolGet())
+		return entry, err
+	}
+	if currentOnly {
+		if current == nil {
+			return nil, nil
+		}
+		// Replace the whole group: no stale live copy or marker can override
+		// this current global decision.
+		return encode(*current)
+	}
+	versions := make([][]xlMetaV2ShallowVersion, 0, len(group))
+	for _, entry := range group {
+		if entry.listingSet != nil {
+			fi := reads[entry.listingSet].fi
+			xl := &xlMetaV2{}
+			if err := xl.AddVersion(fi); err != nil {
+				return nil, listingReadQuorumError(bucket, object)
+			}
+			versions = append(versions, xl.versions)
+			continue
+		}
+		xl, err := entry.xlmeta()
+		if err != nil {
+			return nil, listingReadQuorumError(bucket, object)
+		}
+		versions = append(versions, xl.versions)
+	}
+	// Preserve the existing cross-pool merge rule for recovered live metadata
+	// and delete markers when no source was strictly absent.
+	xl := &xlMetaV2{versions: mergeXLV2Versions(1, true, 0, versions...)}
+	// Do not retain shallow versions referencing reusable input buffers.
+	resolved := &metaCacheEntry{name: name, reusable: true}
+	resolved.metadata, err = xl.AppendTo(metaDataPoolGet())
+	return resolved, err
+}
+
 // listMerged will list across all sets and return a merged results stream.
 // The result channel is closed when no more results are expected.
 func (z *erasureServerPools) listMerged(ctx context.Context, o listPathOptions, results chan<- metaCacheEntry) error {
@@ -285,11 +455,14 @@ func (z *erasureServerPools) listMerged(ctx context.Context, o listPathOptions,
 				defer wg.Done()
 				err := set.listPath(listCtx, o, innerResults)
 				mu.Lock()
-				defer mu.Unlock()
 				if err == nil {
 					allAtEOF = false
 				}
 				errs[i] = err
+				mu.Unlock()
+				// The merger must observe the terminal error before treating
+				// this stream as exhausted and advancing to another pool.
+				xioutil.SafeClose(innerResults)
 			}(len(errs), set)
 			errs = append(errs, nil)
 		}
@@ -298,7 +471,18 @@ func (z *erasureServerPools) listMerged(ctx context.Context, o listPathOptions,
 
 	// Gather results to a single channel.
 	// Quorum is one since we are merging across sets.
-	err := mergeEntryChannels(ctx, inputs, results, 1)
+	err := mergeEntryChannelsWithResolver(ctx, inputs, results, 1, func(i int) error {
+		mu.Lock()
+		err := errs[i]
+		mu.Unlock()
+		// Keep the existing aggregate handling of absent paths and buckets.
+		if errors.Is(err, io.EOF) || isAllNotFound([]error{err}) {
+			return nil
+		}
+		return err
+	}, func(group []*metaCacheEntry) (*metaCacheEntry, error) {
+		return z.resolveListingGroup(ctx, o.Bucket, group)
+	})
 
 	cancelList()
 	wg.Wait()
@@ -415,13 +599,24 @@ func (z *erasureServerPools) listAndSave(ctx context.Context, o *listPathOptions
 		cancel()
 	}()
 
-	// Do listing...
+	// Publish the typed terminal result before either consumer observes EOF.
+	// mergeEntryChannels closes inCh before listMerged has inspected set errors.
+	listingDone := make(chan struct{})
+	var cacheErrorDone chan struct{}
+	var listingErr error
 	go func(o listPathOptions) {
-		err := z.listMerged(listCtx, o, inCh)
+		listingErr = z.listMerged(listCtx, o, inCh)
-		if err != nil {
+		failed := listingErr != nil && !errors.Is(listingErr, io.EOF)
-			meta.setErr(err.Error())
+		if failed {
+			cacheErrorDone = make(chan struct{})
 		}
-		o.debugln("listAndSave: listing", o.ID, "finished with ", err)
+		close(listingDone)
+		if failed {
+			meta.setErr(listingErr.Error())
+			cancel()
+			close(cacheErrorDone)
+		}
+		o.debugln("listAndSave: listing", o.ID, "finished with ", listingErr)
 	}(*o)
 
 	// Keep track of when we return since we no longer have to send entries to output.
@@ -451,16 +646,33 @@ func (z *erasureServerPools) listAndSave(ctx context.Context, o *listPathOptions
 			entry.reusable = returned
 			saveCh <- entry
 		}
+		// The typed scan result is enough to finish the HTTP response.
+		<-listingDone
 		if !returned {
 			xioutil.SafeClose(outCh)
 		}
+		// Closing saveCh writes an end-of-stream cache block. On failure,
+		// wait until the error is recorded and the writer is canceled.
+		if cacheErrorDone != nil {
+			<-cacheErrorDone
+		}
 		xioutil.SafeClose(saveCh)
 	}()
 
 	entries, err = filteredResults()
-	if err == nil {
+	if err == nil || errors.Is(err, io.EOF) {
-		// Check if listing recorded an error.
+		// Full pages may return before the scan ends. A final page has
-		err = meta.getErr()
+		// crossed the completion barrier above and must preserve scan errors.
+		select {
+		case <-listingDone:
+			if listingErr != nil && !errors.Is(listingErr, io.EOF) {
+				return entries, listingErr
+			}
+		default:
+		}
+		if cacheErr := meta.getErr(); cacheErr != nil {
+			return entries, cacheErr
+		}
 	}
 	return entries, err
 }

cmd/metacache-set.go

@@ -39,7 +39,6 @@ import (
 	"github.com/minio/minio/internal/bucket/versioning"
 	"github.com/minio/minio/internal/color"
 	"github.com/minio/minio/internal/hash"
-	xioutil "github.com/minio/minio/internal/ioutil"
 	"github.com/pgsty/silo-pkg/v3/console"
 )
 
@@ -423,63 +422,89 @@ func (er *erasureObjects) streamMetadataParts(ctx context.Context, o listPathOpt
 		retryDelay    = 50 * time.Millisecond
 		retryDelay250 = 250 * time.Millisecond
 	)
+	var lastStateCheck time.Time
+	retryWait := func() error {
+		if contextCanceled(ctx) {
+			return ctx.Err()
+		}
+		// A failed producer cannot publish another block. Check on the first
+		// failure, then at most once per retry interval while it is still running.
+		if time.Since(lastStateCheck) >= retryDelay250 {
+			lastStateCheck = time.Now()
+			if err := o.checkMetacacheState(ctx, rpc); err != nil {
+				return fmt.Errorf("remote listing canceled: %w", err)
+			}
+		}
+		retries++
+		delay := retryDelay250
+		if retries == 1 {
+			delay = retryDelay
+		}
+		timer := time.NewTimer(delay)
+		defer timer.Stop()
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-timer.C:
+			return nil
+		}
+	}
+	// Probe one online disk per retry, rotating so a missing replica cannot
+	// pin the reader to one disk. A miss only delays the next probe: it never
+	// establishes absence or replaces the quorum read after a probe succeeds.
+	// This bounds missing-block I/O without sleeping once for every disk.
+	nextProbeDisk := 0
+	probePart := func(part int) bool {
+		disks := er.getDisks()
+		for range disks {
+			if contextCanceled(ctx) {
+				return false
+			}
+			i := nextProbeDisk % len(disks)
+			nextProbeDisk = i + 1
+			disk := disks[i]
+			if disk == nil || !disk.IsOnline() {
+				continue
+			}
+			_, err := disk.ReadVersion(ctx, "", minioMetaBucket, o.objectPath(part), "", ReadOptions{})
+			return err == nil
+		}
+		return false
+	}
 
+	retryOtherError := false
 	for {
 		if contextCanceled(ctx) {
 			return entries, ctx.Err()
 		}
 
-		// If many failures, check the cache state.
-		if retries > 10 {
-			err := o.checkMetacacheState(ctx, rpc)
-			if err != nil {
-				return entries, fmt.Errorf("remote listing canceled: %w", err)
-			}
-			retries = 1
-		}
-
 		// All operations are performed without locks, so we must be careful and allow for failures.
-		// Read metadata associated with the object from a disk.
+		if retries > 0 && !retryOtherError && !probePart(0) {
-		if retries > 0 {
+			if err := retryWait(); err != nil {
-			for _, disk := range er.getDisks() {
+				return entries, err
-				if disk == nil {
-					continue
-				}
-				if !disk.IsOnline() {
-					continue
-				}
-				_, err := disk.ReadVersion(ctx, "", minioMetaBucket,
-					o.objectPath(0), "", ReadOptions{})
-				if err != nil {
-					time.Sleep(retryDelay250)
-					retries++
-					continue
-				}
-				break
-			}
-		}
-		retryWait := func() {
-			retries++
-			if retries == 1 {
-				time.Sleep(retryDelay)
-			} else {
-				time.Sleep(retryDelay250)
 			}
+			continue
 		}
 		// Load first part metadata...
 		// Read metadata associated with the object from all disks.
+		retryOtherError = false
 		fi, metaArr, onlineDisks, err := er.getObjectFileInfo(ctx, minioMetaBucket, o.objectPath(0), ObjectOptions{}, true)
 		if err != nil {
 			switch toObjectErr(err, minioMetaBucket, o.objectPath(0)).(type) {
 			case ObjectNotFound, InsufficientReadQuorum:
-				retryWait()
+				if err := retryWait(); err != nil {
+					return entries, err
+				}
 				continue
 			}
 			// Allow one fast retry for other errors.
 			if retries > 0 {
 				return entries, fmt.Errorf("reading first part metadata: %v", err)
 			}
-			retryWait()
+			retryOtherError = true
+			if err := retryWait(); err != nil {
+				return entries, err
+			}
 			continue
 		}
 
@@ -487,14 +512,9 @@ func (er *erasureObjects) streamMetadataParts(ctx context.Context, o listPathOpt
 		switch {
 		case err == nil:
 		case errors.Is(err, io.ErrUnexpectedEOF):
-			if retries == 10 {
+			if err := retryWait(); err != nil {
-				err := o.checkMetacacheState(ctx, rpc)
+				return entries, err
-				if err != nil {
-					return entries, fmt.Errorf("remote listing canceled: %w", err)
-				}
-				retries = -1
 			}
-			retryWait()
 			continue
 		case errors.Is(err, io.EOF):
 			return entries, io.EOF
@@ -508,39 +528,19 @@ func (er *erasureObjects) streamMetadataParts(ctx context.Context, o listPathOpt
 			}
 
 			if partN != loadedPart {
-				if retries > 10 {
+				if retries > 0 && !probePart(partN) {
-					err := o.checkMetacacheState(ctx, rpc)
+					if err := retryWait(); err != nil {
-					if err != nil {
+						return entries, err
-						return entries, fmt.Errorf("waiting for next part %d: %w", partN, err)
-					}
-					retries = 1
-				}
-
-				if retries > 0 {
-					// Load from one disk only
-					for _, disk := range er.getDisks() {
-						if disk == nil {
-							continue
-						}
-						if !disk.IsOnline() {
-							continue
-						}
-						_, err := disk.ReadVersion(ctx, "", minioMetaBucket,
-							o.objectPath(partN), "", ReadOptions{})
-						if err != nil {
-							time.Sleep(retryDelay250)
-							retries++
-							continue
-						}
-						break
 					}
+					continue
 				}
 
 				// Load partN metadata...
 				fi, metaArr, onlineDisks, err = er.getObjectFileInfo(ctx, minioMetaBucket, o.objectPath(partN), ObjectOptions{}, true)
 				if err != nil {
-					time.Sleep(retryDelay250)
+					if err := retryWait(); err != nil {
-					retries++
+						return entries, err
+					}
 					continue
 				}
 				loadedPart = partN
@@ -576,13 +576,15 @@ func (er *erasureObjects) streamMetadataParts(ctx context.Context, o listPathOpt
 			if err != nil && !errors.Is(err, io.EOF) {
 				switch toObjectErr(err, minioMetaBucket, o.objectPath(partN)).(type) {
 				case ObjectNotFound:
-					retries++
+					if err := retryWait(); err != nil {
-					time.Sleep(retryDelay250)
+						return entries, err
+					}
 					continue
 				case InsufficientReadQuorum:
-					retries++
 					loadedPart = -1
-					time.Sleep(retryDelay250)
+					if err := retryWait(); err != nil {
+						return entries, err
+					}
 					continue
 				default:
 					internalLogIf(ctx, err)
@@ -700,9 +702,46 @@ func getQuorumDisks(disks []StorageAPI, infos []DiskInfo, readQuorum int) (newDi
 	return newDisks
 }
 
+// resolveListEntry keeps the existing walk resolution and fallback eligibility.
+// Eligible latest-only disagreements are resolved at the pool merger, where a
+// current absence can replace the entire same-name group, including stale walks.
+func (er *erasureObjects) resolveListEntry(entries metaCacheEntries, errs []error, resolver *metadataResolutionParams) *metaCacheEntry {
+	entry, ok := entries.resolve(resolver)
+	if ok {
+		return entry
+	}
+	if resolver.requestedVersions != 1 {
+		return nil
+	}
+	valid := 0
+	for i := range entries {
+		if !entries[i].isObject() {
+			continue
+		}
+		if _, err := entries[i].xlmeta(); err != nil {
+			continue
+		}
+		valid++
+		if entry == nil {
+			entry = &entries[i]
+		}
+	}
+	for _, err := range errs {
+		if err != nil {
+			valid++
+		}
+	}
+	if entry == nil || valid < resolver.objQuorum {
+		return nil
+	}
+	// This entry exists only in the internal set-to-pool stream. Do not retain
+	// an arbitrary walk snapshot or encode the pending state as metadata.
+	return &metaCacheEntry{name: entry.name, listingSet: er}
+}
+
 // Will return io.EOF if continuing would not yield more results.
 func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions, results chan<- metaCacheEntry) (err error) {
-	defer xioutil.SafeClose(results)
+	// The caller records the terminal error before closing results.
 	o.debugf(color.Green("listPath:")+" with options: %#v", o)
 
 	// get prioritized non-healing disks for listing
@@ -778,8 +817,8 @@ func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions, resul
 		},
 		partial: func(entries metaCacheEntries, errs []error) {
 			// Results Disagree :-(
-			entry, ok := entries.resolve(&resolver)
+			entry := er.resolveListEntry(entries, errs, &resolver)
-			if ok {
+			if entry != nil {
 				select {
 				case <-ctxDone:
 				case results <- *entry:
@@ -809,7 +848,9 @@ func (m *metaCacheRPC) setErr(err string) {
 		// An error is already set.
 		return
 	}
-	meta, _ = m.o.updateMetacacheListing(meta, m.rpc)
+	if updated, err := m.o.updateMetacacheListing(meta, m.rpc); err == nil {
+		meta = updated
+	}
 	*m.meta = meta
 }
 

cmd/metacache-stream.go

@@ -126,6 +126,9 @@ func (w *metacacheWriter) write(objs ...metaCacheEntry) error {
 		}
 	}
 	for _, o := range objs {
+		if o.listingSet != nil {
+			return errors.New("metacacheWriter: unresolved listing entry")
+		}
 		if len(o.name) == 0 {
 			return errors.New("metacacheWriter: no name provided")
 		}
@@ -170,6 +173,9 @@ func (w *metacacheWriter) stream() (chan<- metaCacheEntry, error) {
 	go func() {
 		defer w.streamWg.Done()
 		for o := range objs {
+			if o.listingSet != nil && w.streamErr == nil {
+				w.streamErr = errors.New("metacacheWriter: unresolved listing entry")
+			}
 			if len(o.name) == 0 || w.streamErr != nil {
 				continue
 			}

cmd/xl-storage-format-v2-cross-parity_test.go

@@ -0,0 +1,222 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// 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"
+	"fmt"
+	"net/http"
+	"slices"
+	"testing"
+	"time"
+)
+
+func crossParityNullVersion(t *testing.T, generation int64, data, parity int) xlMetaV2ShallowVersion {
+	t.Helper()
+	fi := newFileInfo("bucket/object", data, parity)
+	fi.Erasure.Index = 1
+	fi.DataDir = "11111111-1111-1111-1111-111111111111"
+	fi.ModTime = time.Unix(generation, 0).UTC()
+	fi.Size = 1024
+	fi.Parts = []ObjectPartInfo{{Number: 1, Size: fi.Size, ActualSize: fi.Size}}
+	fi.Metadata = map[string]string{"etag": fmt.Sprintf("%032d", generation)}
+	var xl xlMetaV2
+	if err := xl.AddVersion(fi); err != nil {
+		t.Fatal(err)
+	}
+	return xl.versions[0]
+}
+
+// A restarted node can retain four 12+4 shards while a successful degraded
+// write leaves twelve newer 8+8 shards. Neither disk order nor the minority's
+// EC must hide the independently sufficient newer group.
+func TestMergeXLV2LatestNullCrossParity(t *testing.T) {
+	versions := []xlMetaV2ShallowVersion{
+		crossParityNullVersion(t, 100, 12, 4),
+		crossParityNullVersion(t, 200, 8, 8),
+		crossParityNullVersion(t, 300, 10, 6),
+	}
+	for _, tt := range []struct {
+		name   string
+		counts [3]int
+		empty  int
+		quorum int
+		want   int
+	}{
+		{"new-twelve-old-four", [3]int{4, 12}, 0, 8, 1},
+		{"old-twelve-new-four", [3]int{12, 4}, 0, 8, 0},
+		{"new-nine-old-seven", [3]int{7, 9}, 0, 8, 1},
+		{"eight-meets-eight", [3]int{4, 8}, 4, 8, 1},
+		{"eight-does-not-meet-nine", [3]int{8, 8}, 0, 9, -1},
+		{"three-subquorums", [3]int{7, 5, 4}, 0, 8, -1},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			var input [][]xlMetaV2ShallowVersion
+			for group, count := range tt.counts {
+				for range count {
+					input = append(input, []xlMetaV2ShallowVersion{versions[group]})
+				}
+			}
+			input = append(input, make([][]xlMetaV2ShallowVersion, tt.empty)...)
+			for order, streams := range quorumVersionOrders(input) {
+				got := mergeXLV2Versions(tt.quorum, false, 1, streams...)
+				if tt.want < 0 {
+					if len(got) != 0 {
+						t.Fatalf("order %d: subquorum groups were combined: %v", order, got)
+					}
+					continue
+				}
+				if len(got) != 1 || got[0].header != versions[tt.want].header {
+					t.Fatalf("order %d: got %v, want group %d", order, got, tt.want)
+				}
+				xl := xlMetaV2{versions: got}
+				fi, err := xl.ToFileInfo("bucket", "object", "", false, true)
+				if err != nil || !fi.IsValid() || fi.Erasure.DataBlocks != int(versions[tt.want].header.EcM) {
+					t.Fatalf("order %d: selected metadata is invalid: %+v, %v", order, fi, err)
+				}
+			}
+		})
+	}
+}
+
+func TestMergeXLV2LatestNullCrossParityScope(t *testing.T) {
+	old := crossParityNullVersion(t, 100, 12, 4)
+	newer := crossParityNullVersion(t, 200, 8, 8)
+	for _, kind := range []string{"strict", "all-versions", "uuid", "delete", "free", "history", "legacy-modern-ec", "partial-ec"} {
+		t.Run(kind, func(t *testing.T) {
+			a, b := old, newer
+			strict, requested := false, 1
+			switch kind {
+			case "strict":
+				strict = true
+			case "all-versions":
+				requested = 0
+			case "uuid":
+				a.header.VersionID, b.header.VersionID = [16]byte{1}, [16]byte{1}
+			case "delete":
+				b.header.Type = DeleteType
+			case "free":
+				b.header.Flags |= xlFlagFreeVersion
+			case "legacy-modern-ec":
+				b.header.EcN, b.header.EcM = 0, 0
+			case "partial-ec":
+				b.header.EcN = 0
+			}
+			// These previously unresolved inputs remain outside the repair.
+			// Strict and legacy-header cases use an older majority followed by
+			// a newer minority; the other cases use the opposite grouping.
+			input := make([][]xlMetaV2ShallowVersion, 16)
+			for i := range input {
+				input[i] = []xlMetaV2ShallowVersion{b}
+				if i >= 12 {
+					input[i] = []xlMetaV2ShallowVersion{a}
+				}
+			}
+			if kind == "strict" || kind == "legacy-modern-ec" || kind == "partial-ec" {
+				for i := range input {
+					input[i] = []xlMetaV2ShallowVersion{a}
+					if i >= 12 {
+						input[i] = []xlMetaV2ShallowVersion{b}
+					}
+				}
+			}
+			if kind == "history" {
+				history := old
+				history.header.VersionID = [16]byte{1}
+				input[0] = append(input[0], history)
+			}
+			if got := mergeXLV2Versions(8, strict, requested, input...); len(got) != 0 {
+				t.Fatalf("excluded input changed: %v", got)
+			}
+		})
+	}
+}
+
+func TestCountXLV2MixedECGroups(t *testing.T) {
+	a := crossParityNullVersion(t, 100, 12, 4)
+	b := crossParityNullVersion(t, 100, 8, 8)
+	c := crossParityNullVersion(t, 200, 10, 6)
+	// Same-time groups remain separate, and equal counts have a deterministic
+	// final EC tiebreak even though sortsBefore does not compare EC itself.
+	input := [][]xlMetaV2ShallowVersion{{a}, {a}, {a}, {b}, {b}, {b}, {c}}
+	for order, streams := range quorumVersionOrders(input) {
+		tops := make([]xlMetaV2ShallowVersion, len(streams))
+		for i := range streams {
+			tops[i] = streams[i][0]
+		}
+		for range 20 {
+			got, count := countXLV2Versions(tops, c.header, c, false, true)
+			if count != 3 || got.header != a.header {
+				t.Fatalf("order %d: got %v, count %d; want distinct group of three", order, got.header, count)
+			}
+		}
+	}
+	differentSig := a
+	differentSig.header.Signature[0]++
+	if _, count := countXLV2Versions([]xlMetaV2ShallowVersion{a, a, differentSig}, a.header, a, false, true); count != 3 {
+		t.Fatalf("non-strict signature groups did not combine: count %d", count)
+	}
+	// Same EC with different time or flags must not accumulate votes either.
+	flags := a
+	flags.header.Flags ^= xlFlagInlineData
+	for _, other := range []xlMetaV2ShallowVersion{crossParityNullVersion(t, 200, 12, 4), flags} {
+		got, count := countXLV2Versions([]xlMetaV2ShallowVersion{a, a, other, other}, a.header, a, false, true)
+		if count != 2 {
+			t.Fatalf("incompatible header groups combined: %v, count %d", got.header, count)
+		}
+	}
+}
+
+// Exercise the regular full-metadata read and HTTP LIST with actual on-disk
+// object metadata, not merely the metadata selection helper.
+func TestListCrossParityReadableMajority(t *testing.T) {
+	for _, size := range []int{1024, 256 << 10} {
+		t.Run(fmt.Sprintf("bytes-%d", size), func(t *testing.T) {
+			z, bucket, router := nullQuorumBackend(t)
+			set := z.serverPools[0].sets[0]
+			disks := set.getDisks()
+			const object = "object"
+			mtime := time.Now().UTC().Add(-time.Hour)
+			oldBody := bytes.Repeat([]byte{'a'}, size)
+			newBody := bytes.Repeat([]byte{'b'}, size)
+			nullQuorumPut(t, z, bucket, object, oldBody, ObjectOptions{MTime: mtime})
+			old := make([][]byte, len(disks))
+			for i, disk := range disks {
+				old[i] = mustReadNullQuorumMeta(t, disk, bucket, object)
+			}
+			nullQuorumPut(t, z, bucket, object, newBody, ObjectOptions{MTime: mtime.Add(time.Minute), MaxParity: true})
+			for i := 12; i < len(disks); i++ {
+				if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, old[i]); err != nil {
+					t.Fatal(err)
+				}
+			}
+			for _, readData := range []bool{false, true} {
+				fi, _, _, err := set.getObjectFileInfo(t.Context(), bucket, object, ObjectOptions{}, readData)
+				if err != nil || !fi.ModTime.Equal(mtime.Add(time.Minute)) || fi.Erasure.DataBlocks != 8 {
+					t.Fatalf("readData=%v: full metadata read lost readable majority: %+v, %v", readData, fi, err)
+				}
+			}
+			if get := nullQuorumRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object)); get.Code != http.StatusOK || !bytes.Equal(get.Body.Bytes(), newBody) {
+				t.Fatalf("GET lost readable majority: status %d", get.Code)
+			}
+			if code, keys := nullQuorumListV2Keys(t, router, bucket); code != http.StatusOK || !slices.Equal(keys, []string{object}) {
+				t.Fatalf("LIST lost readable majority: %d %v", code, keys)
+			}
+		})
+	}
+}

cmd/xl-storage-format-v2.go

@@ -1944,7 +1944,7 @@ func mergeXLV2Versions(quorum int, strict bool, requestedVersions int, versions
 	// Keep the original stream shapes: pruning must not make a versioned object
 	// eligible for the single null-version recount below.
 	originalVersions := versions
-	var checkedSingleNull, singleNull bool
+	var checkedSingleNull, singleNull, mixedECNull bool
 	// Shallow copy input
 	versions = append(make([][]xlMetaV2ShallowVersion, 0, len(versions)), versions...)
 
@@ -2019,19 +2019,24 @@ func mergeXLV2Versions(quorum int, strict bool, requestedVersions int, versions
 					// Version IDs match, but otherwise unable to resolve.
 					// We are either strict, or don't have enough information to match.
 					// Switch to a pure counting algo.
-					latest, latestCount = countXLV2Versions(tops, ver.header, latest, strict)
+					latest, latestCount = countXLV2Versions(tops, ver.header, latest, strict, false)
 					break
 				}
 			}
 			if latestCount < quorum {
 				if !checkedSingleNull {
-					singleNull = singleNullVersionStreams(originalVersions)
+					singleNull = singleNullVersionStreams(originalVersions, false)
+					// Availability-optimized writes can change EC while a drive
+					// is offline. Only latest-only non-strict reads may recount
+					// these ordinary null versions across separate modern EC groups.
+					mixedECNull = !singleNull && !strict && requestedVersions == 1 &&
+						singleNullVersionStreams(originalVersions, true)
 					checkedSingleNull = true
 				}
-				if singleNull {
+				if singleNull || mixedECNull {
 					// A newer minority at the end can hide an older quorum from
 					// the selection loop. Recount before discarding the null ID.
-					if candidate, count := countXLV2Versions(tops, latest.header, latest, strict); count >= quorum {
+					if candidate, count := countXLV2Versions(tops, latest.header, latest, strict, mixedECNull); count >= quorum {
 						latest, latestCount = candidate, count
 					}
 				}
@@ -2098,7 +2103,8 @@ func mergeXLV2Versions(quorum int, strict bool, requestedVersions int, versions
 
 // singleNullVersionStreams excludes histories and other version types from the
 // additional recount. Check the original inputs, before any stream is pruned.
-func singleNullVersionStreams(versions [][]xlMetaV2ShallowVersion) bool {
+// Mixed EC groups require known modern EC headers on every non-empty stream.
+func singleNullVersionStreams(versions [][]xlMetaV2ShallowVersion, allowMixedEC bool) bool {
 	var ec xlMetaV2VersionHeader
 	var haveEC bool
 	for _, stream := range versions {
@@ -2112,7 +2118,10 @@ func singleNullVersionStreams(versions [][]xlMetaV2ShallowVersion) bool {
 		if h.VersionID != [16]byte{} || h.Type != ObjectType || h.FreeVersion() {
 			return false
 		}
-		if haveEC && (h.EcN != ec.EcN || h.EcM != ec.EcM) {
+		if allowMixedEC && !h.hasEC() {
+			return false
+		}
+		if !allowMixedEC && haveEC && (h.EcN != ec.EcN || h.EcM != ec.EcM) {
 			return false
 		}
 		ec, haveEC = h, true
@@ -2121,16 +2130,18 @@ func singleNullVersionStreams(versions [][]xlMetaV2ShallowVersion) bool {
 }
 
 // countXLV2Versions selects the most frequent compatible header for reference's
-// VersionID, retaining the existing sort tiebreak and last matching entry.
+// VersionID and the last matching entry. The mixed-EC null-version recount adds
-func countXLV2Versions(tops []xlMetaV2ShallowVersion, reference xlMetaV2VersionHeader, latest xlMetaV2ShallowVersion, strict bool) (xlMetaV2ShallowVersion, int) {
+// a stable EC tiebreak without changing the ordinary counting path.
+func countXLV2Versions(tops []xlMetaV2ShallowVersion, reference xlMetaV2VersionHeader, latest xlMetaV2ShallowVersion, strict, allowMixedEC bool) (xlMetaV2ShallowVersion, int) {
 	x := make(map[xlMetaV2VersionHeader]int, len(tops))
 	for _, a := range tops {
 		if a.header.VersionID != reference.VersionID {
 			continue
 		}
 		if !strict {
-			// we must match EC, when we are not strict.
+			// The narrow null-version recount can consider other EC groups,
-			if !a.header.matchesEC(reference) {
+			// but EcN/EcM stay in the map key: their votes never combine.
+			if !allowMixedEC && !a.header.matchesEC(reference) {
 				continue
 			}
 			a.header.Signature = [4]byte{}
@@ -2142,9 +2153,19 @@ func countXLV2Versions(tops []xlMetaV2ShallowVersion, reference xlMetaV2VersionH
 		if v < latestCount {
 			continue
 		}
-		if v == latestCount && latest.header.sortsBefore(k) {
+		if v == latestCount {
-			// Tiebreak, use sort.
+			if allowMixedEC {
-			continue
+				current := latest.header
+				current.Signature = [4]byte{}
+				// Signatures are not part of these groups. Use EC counts as
+				// the final tiebreak when the existing sort fields are equal.
+				if current.sortsBefore(k) || !k.sortsBefore(current) &&
+					(current.EcM > k.EcM || current.EcM == k.EcM && current.EcN > k.EcN) {
+					continue
+				}
+			} else if latest.header.sortsBefore(k) {
+				continue
+			}
 		}
 		for _, a := range tops {
 			hdr := a.header

internal/config/api/api.go

@@ -68,7 +68,7 @@ const (
 
 	EnvAPIStaleUploadsCleanupInterval = "MINIO_API_STALE_UPLOADS_CLEANUP_INTERVAL"
 	EnvAPIStaleUploadsExpiry          = "MINIO_API_STALE_UPLOADS_EXPIRY"
-	EnvAPIMultipartListing            = "MINIO_API_MULTIPART_LISTING"
+	EnvAPIMultipartListing            = "MINIO_API_MULTIPART_LISTING" // Retired; used only for a startup warning.
 	EnvAPIDeleteCleanupInterval       = "MINIO_API_DELETE_CLEANUP_INTERVAL"
 	EnvDeleteCleanupInterval          = "MINIO_DELETE_CLEANUP_INTERVAL"
 	EnvAPIODirect                     = "MINIO_API_ODIRECT"
@@ -180,7 +180,6 @@ type Config struct {
 	TransitionWorkers           int           `json:"transition_workers"`
 	StaleUploadsCleanupInterval time.Duration `json:"stale_uploads_cleanup_interval"`
 	StaleUploadsExpiry          time.Duration `json:"stale_uploads_expiry"`
-	MultipartListing            string        `json:"multipart_listing"`
 	DeleteCleanupInterval       time.Duration `json:"delete_cleanup_interval"`
 	EnableODirect               bool          `json:"enable_odirect"`
 	GzipObjects                 bool          `json:"gzip_objects"`
@@ -209,7 +208,7 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) {
 		apiReplicationWorkers,
 		apiReplicationFailedWorkers,
 		"expiry_workers",
-		apiMultipartListing, // Ignore the retired shared-config key during migration.
+		apiMultipartListing, // Tolerate the retired key without changing other API settings.
 	}
 
 	disableODirect := env.Get(EnvAPIDisableODirect, kvs.Get(apiDisableODirect)) == config.EnableOn
@@ -344,11 +343,5 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) {
 		cfg.ObjectMaxVersions = math.MaxInt64
 	}
 
-	cfg.MultipartListing = env.Get(EnvAPIMultipartListing, "legacy")
-	if cfg.MultipartListing != "strict" && cfg.MultipartListing != "legacy" {
-		cfg.MultipartListing = "legacy"
-		return cfg, fmt.Errorf("%s must be strict or legacy; using legacy", EnvAPIMultipartListing)
-	}
-
 	return cfg, nil
 }

internal/config/api/api_test.go

@@ -11,22 +11,21 @@ import (
 	"github.com/minio/minio/internal/config"
 )
 
-func TestMultipartListingMigrationMode(t *testing.T) {
+func TestMultipartListingRetiredConfiguration(t *testing.T) {
 	t.Setenv(EnvAPIRequestsMax, "")
 	t.Setenv(EnvAPISyncEvents, "")
 	t.Setenv(EnvAPIObjectMaxVersions, "")
 	for _, tc := range []struct {
-		name, stored, override, want string
+		name, stored, override string
-		invalid                      bool
 	}{
-		{name: "default", want: "legacy"},
+		{name: "default"},
-		{name: "explicit-legacy", override: "legacy", want: "legacy"},
+		{name: "explicit-legacy", override: "legacy"},
-		{name: "explicit-strict", override: "strict", want: "strict"},
+		{name: "explicit-strict", override: "strict"},
-		{name: "retired-stored-strict", stored: "strict", want: "legacy"},
+		{name: "retired-stored-strict", stored: "strict"},
-		{name: "retired-stored-legacy", stored: "legacy", want: "legacy"},
+		{name: "retired-stored-legacy", stored: "legacy"},
-		{name: "retired-stored-invalid", stored: "automatic", want: "legacy"},
+		{name: "retired-stored-invalid", stored: "automatic"},
-		{name: "environment-override", stored: "legacy", override: "strict", want: "strict"},
+		{name: "environment-override", stored: "legacy", override: "strict"},
-		{name: "invalid-environment", stored: "strict", override: "automatic", want: "legacy", invalid: true},
+		{name: "invalid-environment", stored: "strict", override: "automatic"},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Setenv(EnvAPIMultipartListing, tc.override)
@@ -38,11 +37,11 @@ func TestMultipartListingMigrationMode(t *testing.T) {
 				kvs.Set(apiMultipartListing, tc.stored)
 			}
 			cfg, err := LookupConfig(kvs)
-			if (err != nil) != tc.invalid || cfg.MultipartListing != tc.want {
+			if err != nil {
-				t.Fatalf("mode=%q err=%v, want %q invalid=%v", cfg.MultipartListing, err, tc.want, tc.invalid)
+				t.Fatalf("retired setting must be ignored: %v", err)
 			}
 			if cfg.RequestsMax != 17 || !cfg.SyncEvents || cfg.ObjectMaxVersions != 71 {
-				t.Fatalf("migration setting discarded unrelated config: %+v", cfg)
+				t.Fatalf("retired setting discarded unrelated config: %+v", cfg)
 			}
 		})
 	}
@@ -58,7 +57,7 @@ func TestMultipartListingConfigPersistence(t *testing.T) {
 	t.Cleanup(func() { config.DefaultKVS = previous })
 	for _, help := range Help {
 		if help.Key == apiMultipartListing {
-			t.Fatal("process-only setting advertised as shared config")
+			t.Fatal("retired setting advertised as shared config")
 		}
 	}
 	for _, input := range []string{
@@ -82,11 +81,11 @@ func TestMultipartListingConfigPersistence(t *testing.T) {
 			}
 			kvs := loaded.Merge()[config.APISubSys][config.Default]
 			if _, present := kvs.Lookup(apiMultipartListing); present {
-				t.Fatal("process-only setting persisted in shared config")
+				t.Fatal("retired setting persisted in shared config")
 			}
 			cfg, err := LookupConfig(kvs)
-			if err != nil || cfg.MultipartListing != "strict" {
+			if err != nil {
-				t.Fatalf("environment mode lost: %+v %v", cfg, err)
+				t.Fatalf("retired environment setting caused failure: %+v %v", cfg, err)
 			}
 			if input != "" && cfg.RequestsMax != 17 {
 				t.Fatalf("unrelated setting lost: %+v", cfg)

internal/dsync/drwmutex-retry_test.go

@@ -0,0 +1,218 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// 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 dsync
+
+import (
+	"context"
+	"math/rand"
+	"sync/atomic"
+	"testing"
+	"testing/synctest"
+	"time"
+)
+
+// retryTestLocker deliberately leaves unexpected methods unimplemented.
+// These tests exercise the real DRWMutex retry and cleanup code without HTTP.
+type retryTestLocker struct {
+	NetLocker
+	acquire func(context.Context, LockArgs) (bool, error)
+	release func(context.Context, LockArgs) (bool, error)
+}
+
+func (l retryTestLocker) Lock(ctx context.Context, args LockArgs) (bool, error) {
+	return l.acquire(ctx, args)
+}
+
+func (l retryTestLocker) RLock(ctx context.Context, args LockArgs) (bool, error) {
+	return l.acquire(ctx, args)
+}
+
+func (l retryTestLocker) Unlock(ctx context.Context, args LockArgs) (bool, error) {
+	return l.release(ctx, args)
+}
+
+func (l retryTestLocker) RUnlock(ctx context.Context, args LockArgs) (bool, error) {
+	return l.release(ctx, args)
+}
+
+func (l retryTestLocker) String() string { return "retry-test-locker" }
+
+func newRetryTestMutex(lockers []NetLocker) *DRWMutex {
+	dm := NewDRWMutex(&Dsync{
+		GetLockers: func() ([]NetLocker, string) { return lockers, "retry-test-owner" },
+		Timeouts:   DefaultTimeouts,
+	}, "retry-test-resource")
+	dm.rng = rand.New(&lockedRandSource{src: rand.NewSource(1)})
+	return dm
+}
+
+func retryTestAcquire(ctx context.Context, dm *DRWMutex, read bool, opts Options) bool {
+	if read {
+		return dm.GetRLock(ctx, nil, "retry-test-uid", "retry-test", opts)
+	}
+	return dm.GetLock(ctx, nil, "retry-test-uid", "retry-test", opts)
+}
+
+func TestDRWMutexRetryCancellation(t *testing.T) {
+	for _, read := range []bool{false, true} {
+		for _, opts := range []Options{
+			{Timeout: time.Hour},
+			{Timeout: time.Hour, RetryInterval: time.Hour},
+		} {
+			synctest.Test(t, func(t *testing.T) {
+				var calls atomic.Int64
+				lockers := make([]NetLocker, 4)
+				for i := range lockers {
+					lockers[i] = retryTestLocker{acquire: func(context.Context, LockArgs) (bool, error) {
+						calls.Add(1)
+						return false, nil
+					}}
+				}
+				dm := newRetryTestMutex(lockers)
+				ctx, cancel := context.WithCancel(t.Context())
+				defer cancel()
+				done := make(chan bool, 1)
+				go func() {
+					done <- retryTestAcquire(ctx, dm, read, opts)
+				}()
+				synctest.Wait()
+				if got := calls.Load(); got != 4 {
+					t.Fatalf("first attempt made %d RPCs, want 4", got)
+				}
+				cancel()
+				synctest.Wait()
+				select {
+				case locked := <-done:
+					if locked {
+						t.Fatal("canceled waiter acquired lock")
+					}
+				default:
+					t.Errorf("read=%v opts=%+v: cancellation did not interrupt retry wait", read, opts)
+					<-done
+				}
+				time.Sleep(time.Second)
+				synctest.Wait()
+				if got := calls.Load(); got != 4 {
+					t.Fatalf("canceled waiter issued a new RPC attempt: %d calls", got)
+				}
+			})
+		}
+	}
+}
+
+func TestDRWMutexRetryDeadline(t *testing.T) {
+	for _, interval := range []time.Duration{0, time.Hour} {
+		synctest.Test(t, func(t *testing.T) {
+			locker := retryTestLocker{acquire: func(context.Context, LockArgs) (bool, error) {
+				return false, nil
+			}}
+			dm := newRetryTestMutex([]NetLocker{locker})
+			start := time.Now()
+			if retryTestAcquire(t.Context(), dm, false, Options{Timeout: 10 * time.Millisecond, RetryInterval: interval}) {
+				t.Fatal("unavailable lock was acquired")
+			}
+			if elapsed := time.Since(start); elapsed != 10*time.Millisecond {
+				t.Fatalf("retry wait overran deadline: %v", elapsed)
+			}
+		})
+	}
+}
+
+func TestDRWMutexCanceledPartialAndLateGrants(t *testing.T) {
+	for _, read := range []bool{false, true} {
+		for _, slowRelease := range []bool{false, true} {
+			synctest.Test(t, func(t *testing.T) {
+				lateGrant := make(chan struct{})
+				releaseGate := make(chan struct{})
+				if !slowRelease {
+					close(releaseGate)
+				}
+				var calls, releases atomic.Int64
+				var held [4]atomic.Bool
+				lockers := make([]NetLocker, len(held))
+				for i := range lockers {
+					lockers[i] = retryTestLocker{
+						acquire: func(ctx context.Context, _ LockArgs) (bool, error) {
+							calls.Add(1)
+							if i > 0 {
+								<-lateGrant
+							}
+							if ctx.Err() != nil {
+								t.Error("NetLocker acquisition unexpectedly inherits caller cancellation")
+							}
+							held[i].Store(true)
+							return true, nil
+						},
+						release: func(context.Context, LockArgs) (bool, error) {
+							if i == 0 {
+								<-releaseGate
+							}
+							held[i].Store(false)
+							releases.Add(1)
+							return true, nil
+						},
+					}
+				}
+				dm := newRetryTestMutex(lockers)
+				ctx, cancel := context.WithCancel(t.Context())
+				defer cancel()
+				done := make(chan bool, 1)
+				go func() { done <- retryTestAcquire(ctx, dm, read, Options{Timeout: time.Hour}) }()
+				synctest.Wait()
+				if !held[0].Load() || calls.Load() != 4 {
+					t.Fatal("partial acquisition was not established")
+				}
+				cancel()
+				synctest.Wait()
+				if slowRelease {
+					// Cancellation of retry sleep is not a whole-call bound:
+					// synchronous cleanup of a partial grant can still block.
+					select {
+					case <-done:
+						t.Fatal("acquisition returned before partial-grant cleanup completed")
+					default:
+					}
+					close(releaseGate)
+					synctest.Wait()
+				}
+				select {
+				case locked := <-done:
+					if locked {
+						t.Fatal("canceled partial acquisition returned success")
+					}
+				default:
+					t.Error("acquisition did not return after partial-grant cleanup")
+					<-done
+				}
+				if held[0].Load() || releases.Load() != 1 {
+					t.Fatal("partial grant was not released")
+				}
+				close(lateGrant)
+				synctest.Wait()
+				for i := range held {
+					if held[i].Load() {
+						t.Errorf("late grant on node %d was not released", i)
+					}
+				}
+				if calls.Load() != 4 || releases.Load() != 4 {
+					t.Fatalf("calls=%d releases=%d, want 4/4", calls.Load(), releases.Load())
+				}
+			})
+		}
+	}
+}

internal/dsync/drwmutex.go

@@ -260,14 +260,22 @@ func (dm *DRWMutex) lockBlocking(ctx context.Context, lockLossCallback func(), i
 				return locked
 			}
 
+			var retryInterval time.Duration
 			switch {
 			case opts.RetryInterval < 0:
 				return false
 			case opts.RetryInterval > 0:
-				time.Sleep(opts.RetryInterval)
+				retryInterval = opts.RetryInterval
 			default:
 				attempt++
-				time.Sleep(lockRetryBackOff(dm.rng, attempt))
+				retryInterval = lockRetryBackOff(dm.rng, attempt)
+			}
+			timer := time.NewTimer(retryInterval)
+			select {
+			case <-ctx.Done():
+				timer.Stop()
+				return false
+			case <-timer.C:
 			}
 		}
 	}

internal/lsync/lrwmutex-retry_test.go

@@ -0,0 +1,72 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// 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 lsync
+
+import (
+	"context"
+	"testing"
+	"testing/synctest"
+	"time"
+)
+
+func TestLRWMutexRetryCancellation(t *testing.T) {
+	for _, read := range []bool{false, true} {
+		synctest.Test(t, func(t *testing.T) {
+			lm := NewLRWMutex()
+			lm.Lock()
+			defer lm.Unlock()
+			ctx, cancel := context.WithCancel(t.Context())
+			defer cancel()
+			done := make(chan bool, 1)
+			go func() {
+				if read {
+					done <- lm.GetRLock(ctx, "waiter", "retry-test", time.Hour)
+				} else {
+					done <- lm.GetLock(ctx, "waiter", "retry-test", time.Hour)
+				}
+			}()
+			synctest.Wait()
+			cancel()
+			synctest.Wait()
+			select {
+			case locked := <-done:
+				if locked {
+					t.Fatal("canceled waiter acquired lock")
+				}
+			default:
+				t.Error("cancellation did not interrupt local retry wait")
+				<-done
+			}
+		})
+	}
+}
+
+func TestLRWMutexRetryDeadline(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		lm := NewLRWMutex()
+		lm.Lock()
+		defer lm.Unlock()
+		start := time.Now()
+		if lm.GetLock(t.Context(), "waiter", "retry-test", time.Millisecond) {
+			t.Fatal("unavailable lock was acquired")
+		}
+		if elapsed := time.Since(start); elapsed != time.Millisecond {
+			t.Fatalf("local retry wait overran deadline: %v", elapsed)
+		}
+	})
+}

internal/lsync/lrwmutex.go

@@ -115,7 +115,13 @@ func (lm *LRWMutex) lockLoop(ctx context.Context, id, source string, timeout tim
 			if lm.lock(id, source, isWriteLock) {
 				return true
 			}
-			time.Sleep(time.Duration(r.Float64() * float64(lockRetryInterval)))
+			timer := time.NewTimer(time.Duration(r.Float64() * float64(lockRetryInterval)))
+			select {
+			case <-retryCtx.Done():
+				timer.Stop()
+				return false
+			case <-timer.C:
+			}
 		}
 	}
 }