fix(listing): snapshot asynchronous cache cleanup inputs

Capture the metacache peer before launching cleanup and pass listing
options by value. The parent can clear its cache ID without changing the
background RPC lookup or racing with its reads of shared request state.

Publish complete immutable disk snapshots through atomic.Value in the
listing fixtures instead of replacing getDisks after requests start.
Add a real peer RPC regression for cleanup using the original cache ID;
reverting the options copy reproduces both the race and an empty RPC ID.

Verified the final 18-test listing race suite, 30 repetitions of the real
peer regression, formatting checks and make lint. Prior runtime evidence
retains its original e90cb206 source identity.

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

.github/workflows/go.yml

@@ -96,12 +96,20 @@ 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|MergeXLV2LatestNullCrossParity|CountXLV2MixedECGroups|ListCrossParity)' -count=1 -timeout=5m
+
       - name: Run CPU metrics tests under race detector
         run: go test -race ./cmd -run '^TestLoadCPUMetrics' -count=1 -timeout=5m
 
       - name: Run tag replication tests under race detector
         run: go test -race ./cmd -run '^TestAPITagging' -count=1 -timeout=5m
 
+      - name: Run Object Lock authorization and overwrite tests under race detector
+        env:
+          MINIO_API_REQUESTS_MAX: "10000"
+        run: go test -race -tags kqueue,dev ./cmd -run '^Test(APIPutObjectRetention|APIVersionAddressedWrite|PutObjectRetention|DirectoryObjectLock|ObjectLockOverwrite|APICopyObjectPreservesVersionAfterSuspension|MultipartNullObjectLock|PoolsObjectLockOverwrite|MultipartVersioningTransitions|PlanReview|RecoveredMultipartUpload)' -count=1 -timeout=5m
+
   crosscompile:
     name: Cross Compile
     runs-on: ubuntu-latest

.github/workflows/release.yml

@@ -70,6 +70,12 @@ jobs:
           buildscripts/verify-rebrand.sh
           dockerscripts/docker-entrypoint_test.sh
 
+      # Keep this gate aligned with the Object Lock race regression in go.yml.
+      - name: Gate release on Object Lock authorization and overwrite tests under race detector
+        env:
+          MINIO_API_REQUESTS_MAX: "10000"
+        run: go test -race -tags kqueue,dev ./cmd -run '^Test(APIPutObjectRetention|APIVersionAddressedWrite|PutObjectRetention|DirectoryObjectLock|ObjectLockOverwrite|APICopyObjectPreservesVersionAfterSuspension|MultipartNullObjectLock|PoolsObjectLockOverwrite|MultipartVersioningTransitions|PlanReview|RecoveredMultipartUpload)' -count=1 -timeout=5m
+
       - name: Compute release variables
         env:
           # Passed through the environment, never interpolated into the script

cmd/api-errors-checksum_test.go

@@ -0,0 +1,45 @@
+// Copyright (c) 2026 PGSTY
+//
+// This file is part of SILO Object Storage.
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//
+// 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 (
+	"fmt"
+	"net/http"
+	"testing"
+
+	"github.com/minio/minio/internal/hash"
+)
+
+func TestChecksumMismatchAPIMessage(t *testing.T) {
+	for _, algorithm := range []string{"CRC32", "CRC32C", "SHA1", "SHA256", "CRC64NVME", ""} {
+		t.Run(algorithm, func(t *testing.T) {
+			mismatch := hash.ChecksumMismatch{Algorithm: algorithm, Want: "expected", Got: "actual"}
+			want := errorCodes.ToAPIErr(ErrContentChecksumMismatch).Description
+			if algorithm != "" {
+				want = fmt.Sprintf("The %s you specified did not match the calculated checksum.", algorithm)
+			}
+			for _, err := range []error{mismatch, fmt.Errorf("upload: %w", mismatch)} {
+				got := toAPIError(t.Context(), err)
+				if got.HTTPStatusCode != http.StatusBadRequest || got.Code != "XAmzContentChecksumMismatch" || got.Description != want {
+					t.Fatalf("toAPIError(%v) = %+v, want 400 XAmzContentChecksumMismatch: %s", err, got, want)
+				}
+			}
+		})
+	}
+}

cmd/api-errors.go

@@ -2505,6 +2505,14 @@ func toAPIError(ctx context.Context, err error) APIError {
 	case errors.Is(err, errMissingPartChecksum):
 		apiErr.Description = strings.TrimPrefix(err.Error(), errMissingPartChecksum.Error()+": ")
 	}
+	// PutObject/UploadPart checksum mismatches name the algorithm in the
+	// message, matching S3 (e.g. "The CRC32 you specified did not match the
+	// calculated checksum."). Falls through to the generic table description
+	// when the algorithm is unknown.
+	var checksumMismatch hash.ChecksumMismatch
+	if errors.As(err, &checksumMismatch) && checksumMismatch.Algorithm != "" {
+		apiErr.Description = fmt.Sprintf("The %s you specified did not match the calculated checksum.", checksumMismatch.Algorithm)
+	}
 	switch apiErr.Code {
 	case "NotImplemented":
 		apiErr = APIError{

cmd/bucket-object-lock.go

@@ -332,7 +332,12 @@ func checkPutObjectLockAllowed(ctx context.Context, rq *http.Request, bucket, ob
 	}
 
 	if opts.VersionID != "" && !replicaTrusted {
-		if objInfo, err := getObjectInfoFn(ctx, bucket, object, opts); err == nil {
+		objInfo, err := getObjectInfoFn(ctx, bucket, object, opts)
+		if err != nil {
+			if !isErrObjectNotFound(err) && !isErrVersionNotFound(err) && !errors.As(err, &MethodNotAllowed{}) {
+				return mode, retainDate, legalHold, toAPIErrorCode(ctx, err)
+			}
+		} else {
 			r := objectlock.GetObjectRetentionMeta(objInfo.UserDefined)
 			t, err := objectlock.UTCNowNTP()
 			if err != nil {

cmd/data-scanner.go

@@ -744,10 +744,10 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
 					}
 				},
 				// Some disks have data for this.
-				partial: func(entries metaCacheEntries, errs []error) {
+				partial: func(entries metaCacheEntries, errs []error) error {
 					if !f.shouldHeal() {
 						cancel()
-						return
+						return nil
 					}
 					entry, ok := entries.resolve(&resolver)
 					if !ok {
@@ -772,7 +772,7 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
 					}
 
 					if entry.isDir() {
-						return
+						return nil
 					}
 
 					// We got an entry which we should be able to heal.
@@ -787,7 +787,7 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
 							scannerLogIf(ctx, err)
 						}
 						foundObjs = foundObjs || err == nil
-						return
+						return nil
 					}
 
 					custom["versions"] = fmt.Sprint(len(fiv.Versions))
@@ -814,6 +814,7 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
 					}
 					custom["success_versions"] = fmt.Sprint(successVersions)
 					custom["failed_versions"] = fmt.Sprint(failVersions)
+					return nil
 				},
 				// Too many disks failed.
 				finished: func(errs []error) {

cmd/erasure-healing.go

@@ -86,7 +86,7 @@ func (er erasureObjects) listAndHeal(ctx context.Context, bucket, prefix string,
 				cancel()
 			}
 		},
-		partial: func(entries metaCacheEntries, _ []error) {
+		partial: func(entries metaCacheEntries, _ []error) error {
 			entry, ok := entries.resolve(&resolver)
 			if !ok {
 				// check if we can get one entry at least
@@ -94,12 +94,12 @@ func (er erasureObjects) listAndHeal(ctx context.Context, bucket, prefix string,
 				entry, _ = entries.firstFound()
 			}
 			if !recursive && prefix != entry.name {
-				return
+				return nil
 			}
 			if err := healEntry(bucket, *entry, scanMode); err != nil {
 				cancel()
-				return
 			}
+			return nil
 		},
 		finished: nil,
 	}

cmd/erasure-metadata-utils_test.go

@@ -20,6 +20,7 @@ package cmd
 import (
 	"context"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"math/rand"
 	"reflect"
@@ -110,6 +111,96 @@ func TestReduceErrs(t *testing.T) {
 	}
 }
 
+// TestReduceWriteQuorumErrsCommitMetaMissing verifies that an in-quorum
+// errCommitMetaMissing reduces to the sentinel, which renameData() converts
+// to errErasureWriteQuorum, while below-quorum reduces directly to
+// errErasureWriteQuorum.
+func TestReduceWriteQuorumErrsCommitMetaMissing(t *testing.T) {
+	errs := make([]error, 16)
+	for i := range 9 {
+		errs[i] = errCommitMetaMissing
+	}
+	if err := reduceWriteQuorumErrs(t.Context(), errs, objectOpIgnoredErrs, 9); !errors.Is(err, errCommitMetaMissing) {
+		t.Errorf("expected %s, got %s", errCommitMetaMissing, err)
+	}
+	// The sentinel must be distinguishable from errFileNotFound so that
+	// commit failures are never mapped to ObjectNotFound (404).
+	if errors.Is(errCommitMetaMissing, errFileNotFound) {
+		t.Error("errCommitMetaMissing must not match errFileNotFound")
+	}
+
+	// Below quorum the reduction must yield a write-quorum failure (503);
+	// eight sentinels against eight committed drives is not a quorum.
+	belowQuorum := make([]error, 16)
+	for i := range 8 {
+		belowQuorum[i] = errCommitMetaMissing
+	}
+	if err := reduceWriteQuorumErrs(t.Context(), belowQuorum, objectOpIgnoredErrs, 9); !errors.Is(err, errErasureWriteQuorum) {
+		t.Errorf("expected %s, got %s", errErasureWriteQuorum, err)
+	}
+}
+
+// renameDataFailingDisk is a StorageAPI stub returning a fixed error
+// from RenameData, enough to drive renameData() in unit tests.
+type renameDataFailingDisk struct {
+	StorageAPI
+	err       error
+	undoWrite bool
+}
+
+func (d *renameDataFailingDisk) RenameData(context.Context, string, string, FileInfo, string, string, RenameOptions) (RenameDataResp, error) {
+	return RenameDataResp{}, d.err
+}
+
+func (d *renameDataFailingDisk) DeleteVersion(_ context.Context, _, _ string, _ FileInfo, _ bool, opts DeleteOptions) error {
+	d.undoWrite = opts.UndoWrite
+	return nil
+}
+
+// TestRenameDataCommitMetaMissing verifies that renameData() converts an
+// in-quorum errCommitMetaMissing into errErasureWriteQuorum (HTTP 503),
+// while an in-quorum errFileNotFound keeps its meaning so that
+// PutObject() still reports IncompleteBody for client disconnects.
+func TestRenameDataCommitMetaMissing(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		failures int
+		diskErr  error
+		wantErr  error
+	}{
+		{"all_commit_metadata_missing", 16, errCommitMetaMissing, errErasureWriteQuorum},
+		{"quorum_commit_metadata_missing", 9, errCommitMetaMissing, errErasureWriteQuorum},
+		{"neither_side_has_quorum", 8, errCommitMetaMissing, errErasureWriteQuorum},
+		{"successful_write_quorum", 7, errCommitMetaMissing, nil},
+		{"staging_failure", 16, errFileNotFound, errFileNotFound},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			metadata := make([]FileInfo, 16)
+			disks := make([]StorageAPI, 16)
+			for i := range disks {
+				metadata[i] = FileInfo{Deleted: true} // valid without erasure info
+				disk := &renameDataFailingDisk{}
+				if i < tc.failures {
+					disk.err = tc.diskErr
+				}
+				disks[i] = disk
+			}
+			_, _, _, err := renameData(t.Context(), disks, minioMetaTmpBucket, "tmp", metadata, "bucket", "object", 9)
+			if !errors.Is(err, tc.wantErr) {
+				t.Fatalf("expected %v, got %v", tc.wantErr, err)
+			}
+			for i, disk := range disks {
+				// Failed writes undo only the drives that committed. A
+				// successful quorum must never be rolled back.
+				wantUndo := tc.wantErr != nil && i >= tc.failures
+				if got := disk.(*renameDataFailingDisk).undoWrite; got != wantUndo {
+					t.Errorf("disk %d: UndoWrite = %v, want %v", i, got, wantUndo)
+				}
+			}
+		})
+	}
+}
+
 // TestHashOrder - test order of ints in array
 func TestHashOrder(t *testing.T) {
 	testCases := []struct {

cmd/erasure-multipart-cleanup_test.go

@@ -0,0 +1,177 @@
+// Copyright (c) 2026 clown15
+//
+// 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"
+	"encoding/base64"
+	"fmt"
+	"net/url"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func TestExtractModTimeFromUploadIDDir(t *testing.T) {
+	now := time.Unix(1792056747, 123456789).UTC()
+	nano := now.UnixNano()
+	uuid := "d16ad42a-6fe5-4a3b-8092-69983142238f"
+
+	testCases := []struct {
+		name        string
+		dirName     string
+		expectedMod time.Time
+		wantZero    bool
+	}{
+		{
+			name:        "Modern Plaintext Format (<UUID>x<UnixNano>)",
+			dirName:     fmt.Sprintf("%sx%d", uuid, nano),
+			expectedMod: time.Unix(0, nano),
+			wantZero:    false,
+		},
+		{
+			// This valid UUID decodes to bytes containing 'x'. Checking
+			// for that byte before parsing plaintext loses the timestamp.
+			name:        "Plaintext Also Decodes To Base64 Containing x",
+			dirName:     fmt.Sprintf("4eacd477-e733-4851-b821-6fd42785ff87x%d", nano),
+			expectedMod: now,
+		},
+		{
+			name:        "Legacy Format Without Timestamp (<UUID>)",
+			dirName:     uuid,
+			expectedMod: time.Time{},
+			wantZero:    true,
+		},
+		{
+			name:        "Base64 Encoded Format",
+			dirName:     base64.RawURLEncoding.EncodeToString([]byte(fmt.Sprintf("%sx%d", uuid, nano))),
+			expectedMod: time.Unix(0, nano),
+			wantZero:    false,
+		},
+		{
+			name:        "Encoded Format With Deployment Prefix",
+			dirName:     base64.RawURLEncoding.EncodeToString([]byte(fmt.Sprintf("%s.%sx%d", uuid, uuid, nano))),
+			expectedMod: now,
+		},
+		{
+			name:        "Malformed Suffix (<UUID>xinvalid)",
+			dirName:     fmt.Sprintf("%sxnotanumber", uuid),
+			expectedMod: time.Time{},
+			wantZero:    true,
+		},
+		{
+			name:     "Timestamp Overflow",
+			dirName:  uuid + "x9223372036854775808",
+			wantZero: true,
+		},
+		{
+			name:     "Malformed Plaintext UUID",
+			dirName:  fmt.Sprintf("not-a-uuidx%d", nano),
+			wantZero: true,
+		},
+		{
+			name:        "Empty Directory Name",
+			dirName:     "",
+			expectedMod: time.Time{},
+			wantZero:    true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := extractModTimeFromUploadIDDir(tc.dirName)
+			if tc.wantZero {
+				if !got.IsZero() {
+					t.Fatalf("expected zero time for %q, got %v", tc.dirName, got)
+				}
+				return
+			}
+			if !got.Equal(tc.expectedMod) {
+				t.Fatalf("extractModTimeFromUploadIDDir(%q) = %v; want %v", tc.dirName, got, tc.expectedMod)
+			}
+		})
+	}
+}
+
+type multipartCleanupDisk struct {
+	StorageAPI
+	root    string
+	modTime time.Time
+	readErr error
+	reads   int
+}
+
+func (d *multipartCleanupDisk) Endpoint() Endpoint {
+	return Endpoint{URL: &url.URL{Path: d.root}}
+}
+
+func (d *multipartCleanupDisk) ReadVersion(context.Context, string, string, string, string, ReadOptions) (FileInfo, error) {
+	d.reads++
+	return FileInfo{ModTime: d.modTime}, d.readErr
+}
+
+func (d *multipartCleanupDisk) StatVol(context.Context, string) (VolInfo, error) {
+	return VolInfo{Created: time.Now()}, nil
+}
+
+func TestCleanupStaleUploadsOnDisk(t *testing.T) {
+	const uuid = "d16ad42a-6fe5-4a3b-8092-69983142238f"
+	expiry := globalAPIConfig.getStaleUploadsExpiry()
+	for _, tc := range []struct {
+		name, prefix string
+		age          time.Duration
+		legacy       bool
+		readErr      error
+		wantReads    int
+		wantDeleted  bool
+	}{
+		{name: "recent", prefix: uuid, age: expiry / 2, readErr: errFileNotFound},
+		{name: "plaintext_base64_collision", prefix: "4eacd477-e733-4851-b821-6fd42785ff87", age: expiry / 2, readErr: errFileNotFound},
+		{name: "expired", prefix: uuid, age: 2 * expiry, readErr: errFileNotFound, wantDeleted: true},
+		{name: "legacy_recent", prefix: uuid, age: expiry / 2, legacy: true, wantReads: 1},
+		{name: "legacy_expired", prefix: uuid, age: 2 * expiry, legacy: true, wantReads: 1, wantDeleted: true},
+		{name: "legacy_metadata_missing", prefix: uuid, age: 2 * expiry, legacy: true, readErr: errFileNotFound, wantReads: 1},
+		{name: "malformed_metadata_missing", prefix: "not-a-uuid", age: 2 * expiry, readErr: errFileNotFound, wantReads: 1},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			modTime := time.Now().Add(-tc.age)
+			id := tc.prefix
+			if !tc.legacy {
+				id = fmt.Sprintf("%sx%d", tc.prefix, modTime.UnixNano())
+			}
+			disk := &multipartCleanupDisk{root: t.TempDir(), modTime: modTime, readErr: tc.readErr}
+			uploadDir := filepath.Join(disk.root, minioMetaMultipartBucket, "sha", id)
+			if err := os.MkdirAll(uploadDir, 0o755); err != nil {
+				t.Fatal(err)
+			}
+			erasureObjects{}.cleanupStaleUploadsOnDisk(t.Context(), disk)
+			if disk.reads != tc.wantReads {
+				t.Errorf("ReadVersion calls = %d, want %d", disk.reads, tc.wantReads)
+			}
+			_, err := os.Stat(uploadDir)
+			if tc.wantDeleted {
+				if !os.IsNotExist(err) {
+					t.Errorf("expired upload must be removed: %v", err)
+				}
+			} else if err != nil {
+				t.Errorf("upload must be preserved: %v", err)
+			}
+		})
+	}
+}

cmd/erasure-multipart.go

@@ -174,6 +174,24 @@ func (er erasureObjects) deleteAll(ctx context.Context, bucket, prefix string) {
 	wg.Wait()
 }
 
+// extractModTimeFromUploadIDDir extracts creation time from the upload ID directory on disk.
+// On-disk directory names are of the form <UUID>x<UnixNano>.
+// Parse the native format first: plaintext UUIDs can also be valid Base64.
+// Preserve the previous encoded-format parser for compatibility.
+func extractModTimeFromUploadIDDir(uploadIDDir string) time.Time {
+	if modTime, ok := multipartUploadTime(uploadIDDir); ok {
+		return modTime
+	}
+	if b64, err := base64.RawURLEncoding.DecodeString(uploadIDDir); err == nil {
+		if split := strings.Split(string(b64), "x"); len(split) == 2 {
+			if t, err := strconv.ParseInt(split[1], 10, 64); err == nil {
+				return time.Unix(0, t)
+			}
+		}
+	}
+	return time.Time{}
+}
+
 // Remove the old multipart uploads on the given disk.
 func (er erasureObjects) cleanupStaleUploadsOnDisk(ctx context.Context, disk StorageAPI) {
 	drivePath := disk.Endpoint().Path
@@ -181,16 +199,7 @@ func (er erasureObjects) cleanupStaleUploadsOnDisk(ctx context.Context, disk Sto
 	readDirFn(pathJoin(drivePath, minioMetaMultipartBucket), func(shaDir string, typ os.FileMode) error {
 		readDirFn(pathJoin(drivePath, minioMetaMultipartBucket, shaDir), func(uploadIDDir string, typ os.FileMode) error {
 			uploadIDPath := pathJoin(shaDir, uploadIDDir)
-			var modTime time.Time
-			// Upload IDs are of the form base64_url(<UUID>x<UnixNano>), we can extract the time from the UUID.
-			if b64, err := base64.RawURLEncoding.DecodeString(uploadIDDir); err == nil {
-				if split := strings.Split(string(b64), "x"); len(split) == 2 {
-					t, err := strconv.ParseInt(split[1], 10, 64)
-					if err == nil {
-						modTime = time.Unix(0, t)
-					}
-				}
-			}
+			modTime := extractModTimeFromUploadIDDir(uploadIDDir)
 			// Fallback for older uploads without time in the ID.
 			if modTime.IsZero() {
 				wait := deleteMultipartCleanupSleeper.Timer(ctx)

cmd/erasure-object.go

@@ -1083,6 +1083,12 @@ func renameData(ctx context.Context, disks []StorageAPI, srcBucket, srcEntry str
 	errs := g.Wait()
 
 	err := reduceWriteQuorumErrs(ctx, errs, objectOpIgnoredErrs, writeQuorum)
+	if errors.Is(err, errCommitMetaMissing) {
+		// Commit metadata went missing on a quorum of drives; the write
+		// did not commit. Return a 503 class error so clients retry
+		// instead of treating the object as absent (404).
+		err = errErasureWriteQuorum
+	}
 	if err != nil {
 		dg := errgroup.WithNErrs(len(disks))
 		for index, nerr := range errs {

cmd/erasure-server-pool-decom.go

@@ -730,11 +730,12 @@ func (set *erasureObjects) listObjectsToDecommission(ctx context.Context, bi dec
 		minDisks:       listingQuorum,
 		reportNotFound: false,
 		agreed:         fn,
-		partial: func(entries metaCacheEntries, _ []error) {
+		partial: func(entries metaCacheEntries, _ []error) error {
 			entry, ok := entries.resolve(&resolver)
 			if ok {
 				fn(*entry)
 			}
+			return nil
 		},
 		finished: nil,
 	})

cmd/erasure-server-pool-rebalance.go

@@ -525,11 +525,12 @@ func (set *erasureObjects) listObjectsToRebalance(ctx context.Context, bucketNam
 		minDisks:       listingQuorum,
 		reportNotFound: false,
 		agreed:         fn,
-		partial: func(entries metaCacheEntries, _ []error) {
+		partial: func(entries metaCacheEntries, _ []error) error {
 			entry, ok := entries.resolve(&resolver)
 			if ok {
 				fn(*entry)
 			}
+			return nil
 		},
 		finished: nil,
 	})

cmd/erasure-server-pool.go

@@ -2651,11 +2651,15 @@ func (z *erasureServerPools) Walk(ctx context.Context, bucket, prefix string, re
 					minDisks:       listingQuorum,
 					reportNotFound: false,
 					agreed:         send,
-					partial: func(entries metaCacheEntries, _ []error) {
-						entry, ok := entries.resolve(&resolver)
-						if ok {
+					partial: func(entries metaCacheEntries, errs []error) error {
+						entry, err := set.resolveListEntry(ctx, bucket, entries, errs, &resolver, z.listingNSLock(set))
+						if err != nil {
+							return err
+						}
+						if entry != nil {
 							send(*entry)
 						}
+						return nil
 					},
 					finished: nil,
 				}

cmd/global-heal.go

@@ -529,7 +529,7 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
 				jt.Take()
 				go healEntry(bucket, entry)
 			},
-			partial: func(entries metaCacheEntries, _ []error) {
+			partial: func(entries metaCacheEntries, _ []error) error {
 				entry, ok := entries.resolve(&resolver)
 				if !ok {
 					// check if we can get one entry at least
@@ -538,6 +538,7 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
 				}
 				jt.Take()
 				go healEntry(bucket, *entry)
+				return nil
 			},
 			finished: func(errs []error) {
 				success := countErrs(errs, nil)

cmd/metacache-entries.go

@@ -670,6 +670,12 @@ 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 {
 	defer xioutil.SafeClose(out)
 	top := make([]*metaCacheEntry, len(in))
 	nDone := 0
@@ -683,6 +689,9 @@ func mergeEntryChannels(ctx context.Context, in []chan metaCacheEntry, out chan<
 				return ctx.Err()
 			case v, ok := <-in[0]:
 				if !ok {
+					if terminalErr != nil {
+						return terminalErr(0)
+					}
 					return nil
 				}
 				select {
@@ -700,6 +709,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 {

cmd/metacache-listing-cache_test.go

@@ -0,0 +1,326 @@
+// 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"
+	"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)
+	}
+}

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 := set.resolveListEntry(t.Context(), bucket, entries, nil, &resolver, z.listingNSLock(set))
+	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,84 @@
+// 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},
+	} {
+		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" }}
+			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 {
+				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)
+				}
+			}()
+			reader := z.listingNSLock(target)("bucket", "object")
+			ctx, cancel := context.WithTimeout(t.Context(), 80*time.Millisecond)
+			defer cancel()
+			rctx, err := reader.GetRLock(ctx, globalOperationTimeout)
+			if err == nil {
+				reader.RUnlock(rctx)
+				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()
+			rctx, err = reader.GetRLock(ctx, globalOperationTimeout)
+			if err != nil {
+				t.Fatalf("fallback read lock failed after writer unlock: %v", err)
+			}
+			reader.RUnlock(rctx)
+		})
+	}
+}

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 = set.resolveListEntry(ctx, bucket, entries, errs, &resolver, z.listingNSLock(set))
+	}()
+	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-null-quorum_test.go

@@ -52,6 +52,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 +150,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 +282,357 @@ func TestListObjectsSingleNullQuorumHTTP(t *testing.T) {
 	}
 }
 
+func TestListObjectsFallsBackToReadableMetadata(t *testing.T) {
+	z, bucket, router := nullQuorumBackend(t)
+	set := z.serverPools[0].sets[0]
+	disks := set.getDisks()
+	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}
+		}
+	}
+	set.getDisks = func() []StorageAPI { return 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())
+	}
+}
+
+// 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 TestListFallbackDoesNotTrustNotFound(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 := set.resolveListEntry(t.Context(), bucket, entries, errs, &resolver, z.listingNSLock(set))
+	if entry != nil || !isErrReadQuorum(err) {
+		t.Fatalf("ambiguous not-found: got entry=%v err=%v; want read quorum 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-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,16 @@ func (z *erasureServerPools) listPathShortcutEOF(ctx context.Context, bucket str
 	return io.EOF
 }
 
+// listingNSLock chooses the namespace used by regular object writes.
+// Multiple pools serialize at the pools layer; single-pool PUT and GET use
+// the destination set. This keeps the existing single-pool lock routing.
+func (z *erasureServerPools) listingNSLock(er *erasureObjects) func(string, ...string) RWLocker {
+	if z.SinglePool() {
+		return er.NewNSLock
+	}
+	return z.NewNSLock
+}
+
 // 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 {
@@ -283,13 +293,16 @@ func (z *erasureServerPools) listMerged(ctx context.Context, o listPathOptions,
 			inputs = append(inputs, innerResults)
 			go func(i int, set *erasureObjects) {
 				defer wg.Done()
-				err := set.listPath(listCtx, o, innerResults)
+				err := set.listPath(listCtx, o, innerResults, z.listingNSLock(set))
 				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 +311,16 @@ 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 := mergeEntryChannelsWithErrors(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
+	})
 
 	cancelList()
 	wg.Wait()
@@ -415,13 +437,18 @@ 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 listingErr error
 	go func(o listPathOptions) {
-		err := z.listMerged(listCtx, o, inCh)
-		if err != nil {
-			meta.setErr(err.Error())
+		listingErr = z.listMerged(listCtx, o, inCh)
+		if listingErr != nil && !errors.Is(listingErr, io.EOF) {
+			meta.setErr(listingErr.Error())
+			cancel()
 		}
-		o.debugln("listAndSave: listing", o.ID, "finished with ", err)
+		close(listingDone)
+		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,6 +478,9 @@ func (z *erasureServerPools) listAndSave(ctx context.Context, o *listPathOptions
 			entry.reusable = returned
 			saveCh <- entry
 		}
+		// Do not publish an end-of-stream cache block or a final response
+		// before the scan has recorded its terminal error.
+		<-listingDone
 		if !returned {
 			xioutil.SafeClose(outCh)
 		}
@@ -458,9 +488,19 @@ func (z *erasureServerPools) listAndSave(ctx context.Context, o *listPathOptions
 	}()
 
 	entries, err = filteredResults()
-	if err == nil {
-		// Check if listing recorded an error.
-		err = meta.getErr()
+	if err == nil || errors.Is(err, io.EOF) {
+		// Full pages may return before the scan ends. A final page has
+		// 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"
 )
 
@@ -700,9 +699,88 @@ func getQuorumDisks(disks []StorageAPI, infos []DiskInfo, readQuorum int) (newDi
 	return newDisks
 }
 
+// resolveListEntry resolves walk snapshots and, for latest-only listings,
+// falls back to the regular object read path when those snapshots disagree.
+//
+// Only unresolved entries pay for the additional namespace read lock and
+// metadata read. The caller supplies the lock domain used by object writes.
+func (er *erasureObjects) resolveListEntry(ctx context.Context, bucket string, entries metaCacheEntries, errs []error, resolver *metadataResolutionParams, newNSLock func(string, ...string) RWLocker) (*metaCacheEntry, error) {
+	entry, ok := entries.resolve(resolver)
+	if ok {
+		return entry, nil
+	}
+	if resolver.requestedVersions != 1 {
+		return nil, 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, nil
+	}
+	readQuorumErr := func() error {
+		return InsufficientReadQuorum{
+			Bucket: bucket,
+			Object: entry.name,
+			Err:    errErasureReadQuorum,
+			Type:   RQInconsistentMeta,
+		}
+	}
+	// Read under the object read lock, exactly as the GET path does. A
+	// lock-free read of an object being overwritten stably reports a live
+	// object as absent, so an unlocked fallback only trades one silent
+	// omission for another.
+	object := encodeDirObject(entry.name)
+	lock := newNSLock(bucket, object)
+	lockCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	lkctx, err := lock.GetRLock(lockCtx, globalOperationTimeout)
+	if err != nil {
+		return nil, err
+	}
+	defer lock.RUnlock(lkctx)
+
+	fi, _, _, err := er.getObjectFileInfo(lkctx.Context(), bucket, object, ObjectOptions{}, false)
+	if err != nil {
+		// getObjectFileInfo maps inconsistent metadata to not-found, so the
+		// walk evidence means not-found is still undecidable here.
+		if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
+			return nil, readQuorumErr()
+		}
+		return nil, err
+	}
+	// Preserve delete markers until the cross-pool merge has discarded any
+	// older live copy. The normal listing filter removes the merged marker.
+	xl := &xlMetaV2{}
+	if err = xl.AddVersion(fi); err != nil {
+		return nil, readQuorumErr()
+	}
+	resolved := &metaCacheEntry{name: entry.name, cached: xl, reusable: true}
+	resolved.metadata, err = xl.AppendTo(metaDataPoolGet())
+	if err != nil {
+		return nil, err
+	}
+	return resolved, nil
+}
+
 // 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)
+func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions, results chan<- metaCacheEntry, newNSLock func(string, ...string) RWLocker) (err error) {
+	// 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
@@ -776,15 +854,19 @@ func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions, resul
 			case results <- entry:
 			}
 		},
-		partial: func(entries metaCacheEntries, errs []error) {
+		partial: func(entries metaCacheEntries, errs []error) error {
 			// Results Disagree :-(
-			entry, ok := entries.resolve(&resolver)
-			if ok {
+			entry, err := er.resolveListEntry(ctx, o.Bucket, entries, errs, &resolver, newNSLock)
+			if err != nil {
+				return err
+			}
+			if entry != nil {
 				select {
 				case <-ctxDone:
 				case results <- *entry:
 				}
 			}
+			return nil
 		},
 	})
 }
@@ -980,7 +1062,7 @@ type listPathRawOptions struct {
 	// partial will be called when there is disagreement between disks.
 	// if disk did not return any result, but also haven't errored
 	// the entry will be empty and errs will
-	partial func(entries metaCacheEntries, errs []error)
+	partial func(entries metaCacheEntries, errs []error) error
 
 	// finished will be called when all streams have finished and
 	// more than one disk returned an error.
@@ -1191,7 +1273,9 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) {
 			continue
 		}
 		if opts.partial != nil {
-			opts.partial(topEntries, errs)
+			if err := opts.partial(topEntries, errs); err != nil {
+				return err
+			}
 		}
 		// Skip the inputs we used.
 		for i, r := range readers {

cmd/object-api-errors.go

@@ -162,7 +162,7 @@ func toObjectErr(oerr error, params ...string) error {
 			apiErr.Type = v.Type
 		}
 		return apiErr
-	case errErasureWriteQuorum.Error():
+	case errErasureWriteQuorum.Error(), errCommitMetaMissing.Error():
 		apiErr := InsufficientWriteQuorum{}
 		if len(params) >= 1 {
 			apiErr.Bucket = params[0]

cmd/object-handlers-chunked-checksum_test.go

@@ -62,12 +62,13 @@ func testAPIPutObjectChunkedChecksum(obj ObjectLayer, instanceType, bucketName s
 	// A well-formed 4-byte value that does not match the content.
 	wrongCRC := base64.StdEncoding.EncodeToString([]byte{0x01, 0x02, 0x03, 0x04})
 
-	apiCode := func(rec *httptest.ResponseRecorder) string {
+	apiErrOf := func(rec *httptest.ResponseRecorder) APIErrorResponse {
 		var apiErr APIErrorResponse
 		b, _ := io.ReadAll(rec.Body)
 		_ = xml.Unmarshal(b, &apiErr)
-		return apiErr.Code
+		return apiErr
 	}
+	apiCode := func(rec *httptest.ResponseRecorder) string { return apiErrOf(rec).Code }
 
 	// newChunkedJavaForm builds a non-trailer signed chunked PutObject request that
 	// mirrors the AWS Java SDK v2 wire form: the checksum value sits in the header
@@ -131,8 +132,12 @@ func testAPIPutObjectChunkedChecksum(obj ObjectLayer, instanceType, bucketName s
 		if rec.Code != http.StatusBadRequest {
 			t.Fatalf("%s: chunked+wrong CRC32: expected 400, got %d", instanceType, rec.Code)
 		}
-		if code := apiCode(rec); code != "XAmzContentChecksumMismatch" {
-			t.Fatalf("%s: chunked+wrong CRC32: want XAmzContentChecksumMismatch, got %q", instanceType, code)
+		gotErr := apiErrOf(rec)
+		if gotErr.Code != "XAmzContentChecksumMismatch" {
+			t.Fatalf("%s: chunked+wrong CRC32: want XAmzContentChecksumMismatch, got %q", instanceType, gotErr.Code)
+		}
+		if gotErr.Message != "The CRC32 you specified did not match the calculated checksum." {
+			t.Fatalf("%s: chunked+wrong CRC32: message = %q, want S3 algorithm-specific wording", instanceType, gotErr.Message)
 		}
 	}
 }

cmd/object-retention-overwrite_test.go

@@ -25,9 +25,12 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
+	"net/http/httptest"
 	"net/url"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -300,3 +303,244 @@ func TestPoolsObjectLockOverwrite(t *testing.T) {
 		}
 	}
 }
+
+type planReviewGatedBody struct {
+	io.ReadCloser
+	first   sync.Once
+	ready   chan struct{}
+	release chan struct{}
+}
+
+func (g *planReviewGatedBody) Read(p []byte) (int, error) {
+	g.first.Do(func() { close(g.ready); <-g.release })
+	return g.ReadCloser.Read(p)
+}
+
+func TestPlanReviewSlowDirectoryPut(t *testing.T) {
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
+			writer := newObjectAttributesAuthzUser(t, instance, bucket, `"s3:PutObject"`)
+			t.Run(instance, func(t *testing.T) {
+				key := "slow-body-directory/"
+				before, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewBufferString("original"), 8, "", ""), ObjectOptions{Versioned: true, VersionID: nullVersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				payload := []byte("replacement-body")
+				req, err := newTestSignedRequestV4(http.MethodPut, makeTestTargetURL("", bucket, key, nil), int64(len(payload)), bytes.NewReader(payload), writer.AccessKey, writer.SecretKey, nil)
+				if err != nil {
+					t.Fatal(err)
+				}
+				gate := &planReviewGatedBody{ReadCloser: req.Body, ready: make(chan struct{}), release: make(chan struct{})}
+				req.Body = gate
+				var once sync.Once
+				release := func() { once.Do(func() { close(gate.release) }) }
+				defer release()
+				rec := httptest.NewRecorder()
+				done := make(chan struct{})
+				go func() { router.ServeHTTP(rec, req); close(done) }()
+				select {
+				case <-gate.ready:
+				case <-done:
+					t.Fatalf("PUT finished before body: %d %s", rec.Code, rec.Body.String())
+				case <-time.After(5 * time.Second):
+					t.Fatal("PUT did not reach body read")
+				}
+				locked := make(chan int, 1)
+				go func() {
+					locked <- putRetention(t, router, root, bucket, key, nullVersionID, "COMPLIANCE", UTCNow().Add(365*24*time.Hour), false)
+				}()
+				select {
+				case status := <-locked:
+					if status != http.StatusOK {
+						t.Fatalf("set lock while body paused: %d", status)
+					}
+				case <-time.After(5 * time.Second):
+					t.Fatal("retention update blocked behind paused body")
+				}
+				release()
+				select {
+				case <-done:
+				case <-time.After(5 * time.Second):
+					t.Fatal("PUT did not finish")
+				}
+				after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: nullVersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if rec.Code != http.StatusBadRequest || after.ETag != before.ETag || objectlock.GetObjectRetentionMeta(after.UserDefined).Mode != objectlock.RetCompliance {
+					t.Fatal("paused ordinary PUT overwrote the COMPLIANCE null version after the lock was acknowledged")
+				}
+			})
+		},
+	})
+}
+
+type planReviewReadFault struct {
+	ObjectLayer
+	key       string
+	fired     bool
+	putCalled bool
+}
+
+func (p *planReviewReadFault) GetObjectInfo(ctx context.Context, bucket, key string, opts ObjectOptions) (ObjectInfo, error) {
+	if key == p.key && opts.VersionID == nullVersionID && !p.fired {
+		p.fired = true
+		return ObjectInfo{}, InsufficientReadQuorum{}
+	}
+	return p.ObjectLayer.GetObjectInfo(ctx, bucket, key, opts)
+}
+
+func (p *planReviewReadFault) PutObject(ctx context.Context, bucket, key string, data *PutObjReader, opts ObjectOptions) (ObjectInfo, error) {
+	if key == p.key {
+		p.putCalled = true
+	}
+	return p.ObjectLayer.PutObject(ctx, bucket, key, data, opts)
+}
+
+func TestPlanReviewPreflightReadFailure(t *testing.T) {
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, _ auth.Credentials, t *testing.T) {
+			writer := newObjectAttributesAuthzUser(t, instance, bucket, `"s3:PutObject"`)
+			t.Run(instance, func(t *testing.T) {
+				key := "locked-directory-read-fault/"
+				_, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewBufferString("original"), 8, "", ""), ObjectOptions{
+					Versioned: true, VersionID: nullVersionID, ReplicationRequest: true,
+					UserDefined: map[string]string{
+						strings.ToLower(xhttp.AmzObjectLockMode):            "COMPLIANCE",
+						strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): amztime.ISO8601Format(UTCNow().Add(365 * 24 * time.Hour)),
+					},
+				})
+				if err != nil {
+					t.Fatal(err)
+				}
+				initial, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: nullVersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				fault := &planReviewReadFault{ObjectLayer: obj, key: key}
+				setObjectLayer(fault)
+				defer setObjectLayer(obj)
+				rec := retentionAuthzRequest(t, router, writer, http.MethodPut, makeTestTargetURL("", bucket, key, nil), []byte("replacement"), nil)
+				if !fault.fired {
+					t.Fatal("preflight fault was not injected")
+				}
+				if fault.putCalled {
+					t.Fatal("preflight read error allowed PutObject to proceed to storage layer")
+				}
+				after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: nullVersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if rec.Code < 400 {
+					t.Errorf("metadata read quorum error was treated as absence: status=%d", rec.Code)
+				}
+				retentionOverwriteUnchanged(t, initial, after)
+			})
+		},
+	})
+}
+
+func TestPlanReviewLegacyMPUCompliance(t *testing.T) {
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
+			writer := newObjectAttributesAuthzUser(t, instance, bucket, `"s3:PutObject"`)
+			t.Run(instance, func(t *testing.T) {
+				key := "legacy-mpu-compliance"
+				initial, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewBufferString("original"), 8, "", ""), ObjectOptions{Versioned: true})
+				if err != nil {
+					t.Fatal(err)
+				}
+				up, err := obj.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{Versioned: true, VersionID: initial.VersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				payload := []byte("replacement-part")
+				part, err := obj.PutObjectPart(t.Context(), bucket, key, up.UploadID, 1, mustGetPutObjReader(t, bytes.NewReader(payload), int64(len(payload)), "", ""), ObjectOptions{})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if status := putRetention(t, router, root, bucket, key, initial.VersionID, "COMPLIANCE", UTCNow().Add(365*24*time.Hour), false); status != http.StatusOK {
+					t.Fatalf("set compliance: %d", status)
+				}
+				before, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: initial.VersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				body := fmt.Appendf(nil, `<CompleteMultipartUpload><Part><PartNumber>1</PartNumber><ETag>"%s"</ETag></Part></CompleteMultipartUpload>`, part.ETag)
+				rec := retentionAuthzRequest(t, router, writer, http.MethodPost, makeTestTargetURL("", bucket, key, url.Values{"uploadId": {up.UploadID}}), body, nil)
+				after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: initial.VersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if rec.Code != http.StatusForbidden {
+					t.Errorf("legacy upload replaced locked UUID: status=%d", rec.Code)
+				}
+				retentionOverwriteUnchanged(t, before, after)
+			})
+		},
+	})
+}
+
+// A recovered upload retains an authentic server-generated version ID while a
+// surviving replica reintroduces its already-completed version and lock.
+// Only the fixture uses replication privileges; completion uses PutObject only.
+func TestRecoveredMultipartUploadAfterReplica(t *testing.T) {
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
+			writer := newObjectAttributesAuthzUser(t, instance, bucket, `"s3:PutObject"`)
+			t.Run(instance, func(t *testing.T) {
+				z := obj.(*erasureServerPools)
+				key := "recovered-generated-upload"
+				up, err := obj.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{Versioned: true})
+				if err != nil {
+					t.Fatal(err)
+				}
+				er := z.serverPools[0].getHashedSet(key)
+				fi, _, err := er.checkUploadIDExists(t.Context(), bucket, key, up.UploadID, true)
+				if err != nil {
+					t.Fatal(err)
+				}
+				_, err = obj.PutObject(t.Context(), bucket, key,
+					mustGetPutObjReader(t, bytes.NewBufferString("protected-original"), 18, "", ""),
+					ObjectOptions{Versioned: true, VersionID: fi.VersionID, ReplicationRequest: true})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if status := putRetention(t, router, root, bucket, key, fi.VersionID, "COMPLIANCE", UTCNow().Add(365*24*time.Hour), false); status != http.StatusOK {
+					t.Fatalf("setting compliance lock: %d", status)
+				}
+				before, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: fi.VersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				partRec := retentionAuthzRequest(t, router, writer, http.MethodPut,
+					makeTestTargetURL("", bucket, key, url.Values{"uploadId": {up.UploadID}, "partNumber": {"1"}}),
+					[]byte("attacker-replacement"), nil)
+				if partRec.Code != http.StatusOK {
+					t.Fatalf("upload part: %d %s", partRec.Code, partRec.Body.String())
+				}
+				etag := multipartConditionResponseETag(partRec)
+				body := []byte(multipartConditionCompleteBody([]CompletePart{{PartNumber: 1, ETag: etag}}))
+				rec := retentionAuthzRequest(t, router, writer, http.MethodPost,
+					makeTestTargetURL("", bucket, key, url.Values{"uploadId": {up.UploadID}}), body, nil)
+				after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: fi.VersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if rec.Code != http.StatusForbidden {
+					t.Errorf("recovered upload bypassed existing UUID guard: status=%d", rec.Code)
+				}
+				retentionOverwriteUnchanged(t, before, after)
+			})
+		},
+	})
+}

cmd/storage-errors.go

@@ -70,6 +70,11 @@ var errDiskAccessDenied = StorageErr("drive access denied")
 // errFileNotFound - cannot find the file.
 var errFileNotFound = StorageErr("file not found")
 
+// errCommitMetaMissing - the staged xl.meta disappeared between the
+// data rename and the meta-file commit; a write-quorum failure, not
+// a missing object.
+var errCommitMetaMissing = StorageErr("commit metadata is missing")
+
 // errFileNotFound - cannot find requested file version.
 var errFileVersionNotFound = StorageErr("file version not found")
 

cmd/storage-rest-client.go

@@ -106,6 +106,8 @@ func toStorageErr(err error) error {
 		return errVolumeExists
 	case errFileNotFound.Error():
 		return errFileNotFound
+	case errCommitMetaMissing.Error():
+		return errCommitMetaMissing
 	case errFileVersionNotFound.Error():
 		return errFileVersionNotFound
 	case errFileNameTooLong.Error():

cmd/storage-rest-server.go

@@ -290,6 +290,9 @@ func (s *storageRESTServer) NSScannerHandler(ctx context.Context, params *nsScan
 		for update := range updates {
 			resp := storageNSScannerRPC.NewResponse()
 			resp.Update = &update
+			// Pooled responses may carry fields from a previous
+			// scan, always clear the fields not being set.
+			resp.Final = nil
 			out <- resp
 		}
 	}()
@@ -301,6 +304,7 @@ func (s *storageRESTServer) NSScannerHandler(ctx context.Context, params *nsScan
 	// Send final response.
 	resp := storageNSScannerRPC.NewResponse()
 	resp.Final = &ui
+	resp.Update = nil
 	out <- resp
 	return nil
 }

cmd/storage-rest-server_test.go

@@ -0,0 +1,94 @@
+// Copyright (c) 2026 Julien Laurenceau and Feng Ruohang
+//
+// This file is part of Silo Object Storage.
+//
+// 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"
+
+	"github.com/minio/madmin-go/v3"
+	"github.com/minio/minio/internal/grid"
+)
+
+type scannerResponseTestDisk struct {
+	StorageAPI
+}
+
+func (scannerResponseTestDisk) NSScanner(_ context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, _ madmin.HealScanMode, _ func() bool) (dataUsageCache, error) {
+	defer close(updates)
+	updates <- dataUsageEntry{Size: int64(cache.Info.NextCycle)}
+	cache.Info.NextCycle++
+	return cache, nil
+}
+
+func TestNSScannerHandlerClearsStaleResponses(t *testing.T) {
+	endpoint, err := NewEndpoint(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	globalLocalDrivesMu.Lock()
+	oldDrives, oldSets := globalLocalDrivesMap, globalLocalSetDrives
+	globalLocalDrivesMap = map[string]StorageAPI{endpoint.String(): scannerResponseTestDisk{}}
+	globalLocalSetDrives = nil
+	globalLocalDrivesMu.Unlock()
+	t.Cleanup(func() {
+		globalLocalDrivesMu.Lock()
+		globalLocalDrivesMap, globalLocalSetDrives = oldDrives, oldSets
+		globalLocalDrivesMu.Unlock()
+	})
+
+	oldRPC := storageNSScannerRPC
+	t.Cleanup(func() { storageNSScannerRPC = oldRPC })
+	// Every new allocation is dirty too: the check must not depend on
+	// sync.Pool returning a previously used object.
+	storageNSScannerRPC = grid.NewStream[*nsScannerOptions, grid.NoPayload, *nsScannerResp](grid.HandlerNSScanner,
+		func() *nsScannerOptions { return &nsScannerOptions{} }, nil,
+		func() *nsScannerResp {
+			return &nsScannerResp{Update: &dataUsageEntry{Size: -1}, Final: &dataUsageCache{}}
+		})
+	server := storageRESTServer{endpoint: endpoint}
+	for cycle := uint32(1); cycle <= 3; cycle++ {
+		out := make(chan *nsScannerResp, 2)
+		params := &nsScannerOptions{Cache: &dataUsageCache{Info: dataUsageCacheInfo{Name: "bucket", NextCycle: cycle}}}
+		if err := server.NSScannerHandler(t.Context(), params, out); err != nil {
+			t.Fatal(err)
+		}
+		if len(out) != 2 {
+			t.Fatalf("cycle %d: got %d responses, want an update and a final cache", cycle, len(out))
+		}
+		for i := 0; i < 2; i++ {
+			resp := <-out
+			wire, err := resp.MarshalMsg(nil)
+			storageNSScannerRPC.PutResponse(resp)
+			if err != nil {
+				t.Fatal(err)
+			}
+			var decoded nsScannerResp
+			if _, err := decoded.UnmarshalMsg(wire); err != nil {
+				t.Fatal(err)
+			}
+			if i == 0 {
+				if decoded.Update == nil || decoded.Final != nil || decoded.Update.Size != int64(cycle) {
+					t.Errorf("cycle %d: invalid update response: %+v", cycle, decoded)
+				}
+			} else if decoded.Update != nil || decoded.Final == nil || decoded.Final.Info.NextCycle != cycle+1 {
+				t.Errorf("cycle %d: invalid final response: %+v", cycle, decoded)
+			}
+		}
+	}
+}

cmd/xl-storage-errors_test.go

@@ -18,10 +18,16 @@
 package cmd
 
 import (
+	"context"
+	"errors"
+	"net/http"
 	"os"
+	"path/filepath"
 	"runtime"
+	"sync"
 	"syscall"
 	"testing"
+	"time"
 )
 
 func TestSysErrors(t *testing.T) {
@@ -56,3 +62,120 @@ func TestSysErrors(t *testing.T) {
 		}
 	}
 }
+
+// TestCommitMetaMissingMapping verifies the RPC and object-layer mapping
+// of errCommitMetaMissing, returned when the staged xl.meta disappears
+// between the data rename and the meta-file commit in RenameData().
+func TestCommitMetaMissingMapping(t *testing.T) {
+	// The sentinel must survive a storage REST/grid round-trip; errors
+	// cross the wire as strings and are remapped by toStorageErr().
+	if got := toStorageErr(errors.New(errCommitMetaMissing.Error())); !errors.Is(got, errCommitMetaMissing) {
+		t.Fatalf("expected %s, got %s", errCommitMetaMissing, got)
+	}
+	if got := toStorageErr(errCommitMetaMissing); got != errCommitMetaMissing {
+		t.Fatalf("expected %s, got %s", errCommitMetaMissing, got)
+	}
+
+	// toObjectErr() must map the sentinel to InsufficientWriteQuorum
+	// (503), never to ObjectNotFound (404).
+	err := toObjectErr(errCommitMetaMissing, "bucket", "object")
+	qerr, ok := err.(InsufficientWriteQuorum)
+	if !ok {
+		t.Fatalf("expected InsufficientWriteQuorum, got %s (%T)", err, err)
+	}
+	if qerr.Bucket != "bucket" || qerr.Object != "object" {
+		t.Fatalf("unexpected bucket/object %s/%s", qerr.Bucket, qerr.Object)
+	}
+	apiErr := toAPIError(t.Context(), err)
+	if apiErr.Code != "SlowDownWrite" || apiErr.HTTPStatusCode != http.StatusServiceUnavailable {
+		t.Fatalf("expected SlowDownWrite/503, got %+v", apiErr)
+	}
+}
+
+// missingCommitMetaContext removes staged metadata at an existing cancellation
+// check, after WriteAll and (for non-inline objects) the data rename succeeded.
+// This deterministically models concurrent cleanup without changing production
+// code or relying on a filesystem race. The context itself remains live.
+type missingCommitMetaContext struct {
+	context.Context
+	mu             sync.Mutex
+	meta, dataDest string
+	removed        bool
+	removeErr      error
+}
+
+func (c *missingCommitMetaContext) Done() <-chan struct{} {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if !c.removed {
+		if _, err := os.Stat(c.meta); err == nil {
+			dataMoved := c.dataDest == ""
+			if !dataMoved {
+				_, err = os.Stat(c.dataDest)
+				dataMoved = err == nil
+			}
+			if dataMoved {
+				c.removeErr = os.Remove(c.meta)
+				c.removed = c.removeErr == nil
+			}
+		}
+	}
+	return c.Context.Done()
+}
+
+func TestXLStorageRenameDataCommitMetaMissing(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		dataFile bool
+	}{
+		{"empty_object", false},
+		{"data_directory", true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			disk, root, err := newXLStorageTestSetup(t)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer disk.Close()
+			s := disk.storage
+			if err := s.MakeVol(t.Context(), "bucket"); err != nil {
+				t.Fatal(err)
+			}
+			fi := newFileInfo("object", 2, 2)
+			fi.DataDir, fi.VersionID = mustGetUUID(), mustGetUUID()
+			fi.Erasure.Index = 1
+			fi.ModTime = time.Now().UTC()
+			fi.Fresh = true
+			ctx := &missingCommitMetaContext{
+				Context: t.Context(),
+				meta:    filepath.Join(root, minioMetaTmpBucket, "staged", xlStorageFormatFile),
+			}
+			if tc.dataFile {
+				fi.Size = 1
+				fi.AddObjectPart(1, "", 1, 1, fi.ModTime, nil, nil)
+				part := filepath.Join(root, minioMetaTmpBucket, "staged", fi.DataDir, "part.1")
+				if err := os.MkdirAll(filepath.Dir(part), 0o755); err != nil {
+					t.Fatal(err)
+				}
+				if err := os.WriteFile(part, []byte("a"), 0o644); err != nil {
+					t.Fatal(err)
+				}
+				ctx.dataDest = filepath.Join(root, "bucket", "object", fi.DataDir)
+			}
+			_, err = s.RenameData(ctx, minioMetaTmpBucket, "staged", fi, "bucket", "object", RenameOptions{})
+			if !ctx.removed {
+				t.Fatalf("metadata removal did not execute: removal error = %v, RenameData error = %v", ctx.removeErr, err)
+			}
+			if !errors.Is(err, errCommitMetaMissing) {
+				t.Fatalf("expected commit metadata missing, got %v", err)
+			}
+			apiErr := toAPIError(t.Context(), toObjectErr(err, "bucket", "object"))
+			if apiErr.Code != "SlowDownWrite" || apiErr.HTTPStatusCode != http.StatusServiceUnavailable {
+				t.Fatalf("expected SlowDownWrite/503, got %+v", apiErr)
+			}
+			if _, err := os.Stat(filepath.Join(root, "bucket", "object", xlStorageFormatFile)); !os.IsNotExist(err) {
+				t.Fatalf("failed commit must not produce destination metadata: %v", err)
+			}
+		})
+	}
+}

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-legacy.go

@@ -40,6 +40,10 @@ func (x *xlMetaV2VersionHeader) unmarshalV(v uint8, bts []byte) (o []byte, err e
 
 // unmarshalV1 decodes version 1, never released.
 func (x *xlMetaV2VersionHeader) unmarshalV1(bts []byte) (o []byte, err error) {
+	// Fields absent from the V1 wire format must be reset since the
+	// header may be reused across decodes.
+	x.Signature = [4]byte{}
+	x.EcN, x.EcM = 0, 0
 	var zb0001 uint32
 	zb0001, bts, err = msgp.ReadArrayHeaderBytes(bts)
 	if err != nil {
@@ -134,7 +138,7 @@ type xlMetaV2VersionHeaderV2 struct {
 
 // UnmarshalMsg implements msgp.Unmarshaler
 func (z *xlMetaV2VersionHeaderV2) UnmarshalMsg(bts []byte) (o []byte, err error) {
-	z.EcN, z.EcN = 0, 0
+	z.EcN, z.EcM = 0, 0
 	var zb0001 uint32
 	zb0001, bts, err = msgp.ReadArrayHeaderBytes(bts)
 	if err != nil {
@@ -184,7 +188,7 @@ func (z *xlMetaV2VersionHeaderV2) UnmarshalMsg(bts []byte) (o []byte, err error)
 
 // DecodeMsg implements msgp.Decodable
 func (z *xlMetaV2VersionHeaderV2) DecodeMsg(dc *msgp.Reader) (err error) {
-	z.EcN, z.EcN = 0, 0
+	z.EcN, z.EcM = 0, 0
 	var zb0001 uint32
 	zb0001, err = dc.ReadArrayHeader()
 	if err != nil {

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.
-func countXLV2Versions(tops []xlMetaV2ShallowVersion, reference xlMetaV2VersionHeader, latest xlMetaV2ShallowVersion, strict bool) (xlMetaV2ShallowVersion, int) {
+// VersionID and the last matching entry. The mixed-EC null-version recount adds
+// 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.
-			if !a.header.matchesEC(reference) {
+			// The narrow null-version recount can consider other EC groups,
+			// 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) {
-			// Tiebreak, use sort.
-			continue
+		if v == latestCount {
+			if allowMixedEC {
+				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

cmd/xl-storage-format-v2_test.go

@@ -38,6 +38,7 @@ import (
 	"github.com/klauspost/compress/zstd"
 	"github.com/minio/minio/internal/bucket/lifecycle"
 	xhttp "github.com/minio/minio/internal/http"
+	"github.com/tinylib/msgp/msgp"
 )
 
 func TestReadXLMetaNoData(t *testing.T) {
@@ -1190,3 +1191,92 @@ func BenchmarkToFileInfoNoParts(b *testing.B) {
 func BenchmarkToFileInfoWithParts(b *testing.B) {
 	benchmarkManyPartsOptionally(b, true)
 }
+
+// TestXLMetaV2HeaderLegacyUnmarshalResetsEc checks that decoding legacy
+// version headers, which carry no erasure metadata fields, cannot leak
+// stale EcN/EcM values from a previously decoded object into the same
+// structure.
+func TestXLMetaV2HeaderLegacyUnmarshalResetsEc(t *testing.T) {
+	base := xlMetaV2VersionHeader{
+		VersionID: [16]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16},
+		ModTime:   time.Now().Unix(),
+		Signature: [4]byte{1, 2, 3, 4},
+		Type:      ObjectType,
+		Flags:     xlFlagInlineData,
+	}
+
+	// Legacy V2 header wire format: 5 fields, no EcN/EcM.
+	var legacy []byte
+	legacy = msgp.AppendArrayHeader(legacy, 5)
+	legacy = msgp.AppendBytes(legacy, base.VersionID[:])
+	legacy = msgp.AppendInt64(legacy, base.ModTime)
+	legacy = msgp.AppendBytes(legacy, base.Signature[:])
+	legacy = msgp.AppendUint8(legacy, uint8(base.Type))
+	legacy = msgp.AppendUint8(legacy, uint8(base.Flags))
+
+	for _, decoder := range []string{"UnmarshalMsg", "DecodeMsg", "loadIndexed"} {
+		t.Run(decoder, func(t *testing.T) {
+			header := base
+			header.EcN, header.EcM = 4, 8
+			x2 := xlMetaV2VersionHeaderV2{xlMetaV2VersionHeader: &header}
+			var err error
+			switch decoder {
+			case "UnmarshalMsg":
+				_, err = x2.UnmarshalMsg(legacy)
+			case "DecodeMsg":
+				err = x2.DecodeMsg(msgp.NewReader(bytes.NewReader(legacy)))
+			case "loadIndexed":
+				indexed := msgp.AppendUint(nil, 2)
+				indexed = msgp.AppendUint(indexed, xlMetaVersion)
+				indexed = msgp.AppendInt(indexed, 1)
+				indexed = msgp.AppendBytes(indexed, legacy)
+				indexed = msgp.AppendBytes(indexed, nil)
+				xl := xlMetaV2{versions: []xlMetaV2ShallowVersion{{header: header}}}
+				err = xl.loadIndexed(indexed, nil)
+				header = xl.versions[0].header
+			}
+			if err != nil {
+				t.Fatal(err)
+			}
+			if header != base {
+				t.Fatalf("legacy header: got %+v, want %+v", header, base)
+			}
+		})
+	}
+
+	// V1 headers (4 fields, never released) must zero the fields too.
+	var v1 []byte
+	v1 = msgp.AppendArrayHeader(v1, 4)
+	v1 = msgp.AppendBytes(v1, base.VersionID[:])
+	v1 = msgp.AppendInt64(v1, base.ModTime)
+	v1 = msgp.AppendUint8(v1, uint8(base.Type))
+	v1 = msgp.AppendUint8(v1, uint8(base.Flags))
+
+	header := base
+	header.EcN, header.EcM = 4, 8
+	header.Signature = [4]byte{9, 9, 9, 9}
+	if _, err := header.unmarshalV1(v1); err != nil {
+		t.Fatalf("legacy V1 header unmarshal failed: %v", err)
+	}
+	if header.EcN != 0 || header.EcM != 0 {
+		t.Fatalf("stale erasure metadata leaked on V1 decode: EcN %d, EcM %d", header.EcN, header.EcM)
+	}
+	if header.Signature != ([4]byte{}) {
+		t.Fatalf("stale signature leaked on V1 decode: %v", header.Signature)
+	}
+
+	// Current headers must keep their erasure metadata.
+	current := base
+	current.EcN, current.EcM = 4, 8
+	wire, err := current.MarshalMsg(nil)
+	if err != nil {
+		t.Fatalf("header marshal failed: %v", err)
+	}
+	decoded := base
+	if _, err := decoded.UnmarshalMsg(wire); err != nil {
+		t.Fatalf("header unmarshal failed: %v", err)
+	}
+	if decoded.EcN != 4 || decoded.EcM != 8 {
+		t.Fatalf("erasure metadata lost on current decode: EcN %d, EcM %d", decoded.EcN, decoded.EcM)
+	}
+}

cmd/xl-storage.go

@@ -2925,7 +2925,15 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 		// if its a partial rename() do not attempt to delete recursively.
 		// this can be healed since all parts are available.
 		s.deleteFile(dstVolumeDir, dstDataPath, false, false)
-		return res, osErrToFileErr(err)
+		err = osErrToFileErr(err)
+		if errors.Is(err, errFileNotFound) {
+			// The staged xl.meta disappeared between staging and commit.
+			// This is a write-quorum failure, not a missing object; use a
+			// dedicated error so the object layer returns 503 instead of
+			// mapping it to ObjectNotFound (404).
+			err = errCommitMetaMissing
+		}
+		return res, err
 	}
 
 	if srcVolume != minioMetaMultipartBucket {

go.mod

@@ -4,7 +4,7 @@ go 1.27.2
 
 // Console and MC retain their historical module paths for best-effort upstream
 // compatibility. Pin the maintained PGSTY implementations used by SILO.
-replace github.com/minio/console => github.com/pgsty/silo-console v0.0.0-20260916075814-1360e26d976d
+replace github.com/minio/console => github.com/pgsty/silo-console v0.0.0-20261015132217-e5b165e4c0ee
 
 replace github.com/minio/mc => github.com/pgsty/mc v0.0.0-20260916070421-e952aa78f10a
 

go.sum

@@ -547,8 +547,8 @@ github.com/pborman/getopt v0.0.0-20170112200414-7148bc3a4c30/go.mod h1:85jBQOZwp
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
 github.com/pgsty/mc v0.0.0-20260916070421-e952aa78f10a h1:FY1vcTQT67HGBe23ozlhOiQnWCzUCGmeSN5x6DT/kFs=
 github.com/pgsty/mc v0.0.0-20260916070421-e952aa78f10a/go.mod h1:6z9lX8kOvaWz2vLlP7BjjF3JWHnIH86XDA2HwcmpZH8=
-github.com/pgsty/silo-console v0.0.0-20260916075814-1360e26d976d h1:A9ou1GkUckGZd3BOgEXeZ393nbYST967SKcI4Lgr/Vs=
-github.com/pgsty/silo-console v0.0.0-20260916075814-1360e26d976d/go.mod h1:JS165CF2yoTBIx1X8k1IQ6vunJcZFG3pRZlWHJ8UT8o=
+github.com/pgsty/silo-console v0.0.0-20261015132217-e5b165e4c0ee h1:e51AnrGe0Zq+nHTYaGfv/GkehQMO9Mekuw/ew6zYwOM=
+github.com/pgsty/silo-console v0.0.0-20261015132217-e5b165e4c0ee/go.mod h1:jpt39bx4wjhXoFB+BTOa/bx/pmU3pXzp2cT7PnT21Lo=
 github.com/pgsty/silo-pkg/v3 v3.14.1 h1:8MONT3Hky9EOnsuPk29590pEBE6MILaOnJtaejPPi7M=
 github.com/pgsty/silo-pkg/v3 v3.14.1/go.mod h1:jxKWxi52jSsDaDhXXPqjOyPDYdxx2iqdBXyLfM/xSIw=
 github.com/philhofer/fwd v1.2.0 h1:e6DnBTl7vGY+Gz322/ASL4Gyp1FspeMvx1RNDoToZuM=

internal/hash/checksum-errors_test.go

@@ -0,0 +1,73 @@
+// Copyright (c) 2026 PGSTY
+//
+// This file is part of SILO Object Storage.
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//
+// 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 hash
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestChecksumMismatchAlgorithm(t *testing.T) {
+	content := []byte("actual content")
+	for _, typ := range []ChecksumType{ChecksumCRC32, ChecksumCRC32C, ChecksumSHA1, ChecksumSHA256, ChecksumCRC64NVME} {
+		t.Run(typ.String(), func(t *testing.T) {
+			wrong := NewChecksumFromData(typ, []byte("wrong content"))
+			check := func(t *testing.T, err error) {
+				t.Helper()
+				var mismatch ChecksumMismatch
+				if !errors.As(err, &mismatch) || mismatch.Algorithm != typ.String() {
+					t.Fatalf("error=%+v, want %s checksum mismatch", err, typ)
+				}
+			}
+			check(t, wrong.Matches(content, 0))
+			parts := NewChecksumFromData(typ, content)
+			parts.WantParts = 2
+			check(t, parts.Matches(content, 1))
+			for _, mode := range []string{"header", "trailer", "malformed-trailer"} {
+				t.Run(mode, func(t *testing.T) {
+					reader, err := NewReader(t.Context(), bytes.NewReader(content), int64(len(content)), "", "", int64(len(content)))
+					if err != nil {
+						t.Fatal(err)
+					}
+					req := httptest.NewRequest(http.MethodPut, "http://localhost/bucket/object", nil)
+					if mode == "header" {
+						req.Header.Set(typ.Key(), wrong.Encoded)
+					} else {
+						req.Header.Set("x-amz-trailer", typ.Key())
+						req.Trailer = make(http.Header)
+						value := wrong.Encoded
+						if mode == "malformed-trailer" {
+							value = "!invalid-base64!"
+						}
+						req.Trailer.Set(typ.Key(), value)
+					}
+					if err = reader.AddChecksum(req, false); err != nil {
+						t.Fatal(err)
+					}
+					_, err = io.ReadAll(reader)
+					check(t, err)
+				})
+			}
+		})
+	}
+}

internal/hash/checksum.go

@@ -584,15 +584,17 @@ func (c Checksum) Matches(content []byte, parts int) error {
 	sum := hasher.Sum(nil)
 	if c.WantParts > 0 && c.WantParts != parts {
 		return ChecksumMismatch{
-			Want: fmt.Sprintf("%s-%d", c.Encoded, c.WantParts),
-			Got:  fmt.Sprintf("%s-%d", base64.StdEncoding.EncodeToString(sum), parts),
+			Algorithm: c.Type.String(),
+			Want:      fmt.Sprintf("%s-%d", c.Encoded, c.WantParts),
+			Got:       fmt.Sprintf("%s-%d", base64.StdEncoding.EncodeToString(sum), parts),
 		}
 	}
 
 	if !bytes.Equal(sum, c.Raw) {
 		return ChecksumMismatch{
-			Want: c.Encoded,
-			Got:  base64.StdEncoding.EncodeToString(sum),
+			Algorithm: c.Type.String(),
+			Want:      c.Encoded,
+			Got:       base64.StdEncoding.EncodeToString(sum),
 		}
 	}
 	return nil

internal/hash/errors.go

@@ -74,8 +74,11 @@ func (e SizeMismatch) Error() string {
 
 // ChecksumMismatch - when content checksum does not match with what was sent from client.
 type ChecksumMismatch struct {
-	Want string
-	Got  string
+	// Algorithm carries the requested checksum type (e.g. "CRC32") so API
+	// error rendering can name it, matching S3's wording.
+	Algorithm string
+	Want      string
+	Got       string
 }
 
 func (e ChecksumMismatch) Error() string {

internal/hash/reader.go

@@ -302,13 +302,14 @@ func (r *Reader) Read(p []byte) (int, error) {
 				r.contentHash.Encoded = r.trailer.Get(r.contentHash.Type.Key())
 				r.contentHash.Raw, err = base64.StdEncoding.DecodeString(r.contentHash.Encoded)
 				if err != nil || len(r.contentHash.Raw) == 0 {
-					return 0, ChecksumMismatch{Got: r.contentHash.Encoded}
+					return 0, ChecksumMismatch{Algorithm: r.contentHash.Type.String(), Got: r.contentHash.Encoded}
 				}
 			}
 			if sum := r.contentHasher.Sum(nil); !bytes.Equal(r.contentHash.Raw, sum) {
 				err := ChecksumMismatch{
-					Want: r.contentHash.Encoded,
-					Got:  base64.StdEncoding.EncodeToString(sum),
+					Algorithm: r.contentHash.Type.String(),
+					Want:      r.contentHash.Encoded,
+					Got:       base64.StdEncoding.EncodeToString(sum),
 				}
 				return n, err
 			}