fix(listing): preserve readable keys across pool and metadata disagreements

Continue PR #221 while retaining its original commits and authorship.
Route latest-only fallback reads through the namespace lock used by normal
writes, propagate terminal scan errors before cache completion, and retain
delete markers until the cross-pool merge has resolved them.

Recount narrowly eligible latest null-version metadata when mixed erasure
layouts prevent selection. Keep each EC group separate and preserve the
requested quorum, strict resolution, and version-listing behavior.

Add regression coverage for locking, terminal errors, cache completion,
delete markers, mixed EC layouts, and HTTP behavior; run focused race tests
in CI. Verified with make test, focused race tests, and local distributed
single- and multi-pool rolling-restart workloads with exact final oracles.

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

.github/workflows/go.yml

@@ -96,6 +96,9 @@ 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
 

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-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,233 @@
+// 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"
+)
+
+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)
+}
+
+func TestListingCachePreservesFallbackError(t *testing.T) {
+	for _, continuation := range []bool{false, true} {
+		name := "cold"
+		if continuation {
+			name = "continuation"
+		}
+		t.Run(name, func(t *testing.T) {
+			z, bucket, router := nullQuorumBackend(t)
+			set := z.serverPools[0].sets[0]
+			disks := set.getDisks()
+			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]}
+				}
+			}
+			set.getDisks = func() []StorageAPI { return 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 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)
+					}
+				}
+			}
+			set.getDisks = func() []StorageAPI { return 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)
+	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())
+	}
+	set := z.serverPools[0].sets[0]
+	resume := make(chan struct{})
+	release := sync.OnceFunc(func() { close(resume) })
+	defer release()
+	var blocked, active atomic.Int64
+	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.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]
+			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)
+				}
+			}
+			disks := deleted.getDisks()
+			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]}
+			}
+			deleted.getDisks = func() []StorageAPI { return 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,243 @@
+// 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]
+	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
+	brokenDisks := broken.getDisks()
+	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]}
+		}
+	}
+	broken.getDisks = func() []StorageAPI { return brokenWalkers }
+	resume := make(chan struct{})
+	defer close(resume)
+	healthyWalkers := slices.Clone(healthy.getDisks())
+	for i, disk := range healthyWalkers {
+		healthyWalkers[i] = &listingPausedTailDisk{StorageAPI: disk, bucket: bucket, tail: "zz", resume: resume, blocked: &blocked, active: &active}
+	}
+	healthy.getDisks = func() []StorageAPI { return 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 := set.getDisks()
+	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]}
+	}
+	set.getDisks = func() []StorageAPI { return 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)
+	}
+
+	set.getDisks = func() []StorageAPI { return 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

@@ -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/xl-storage-format-v2-cross-parity_test.go

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

cmd/xl-storage-format-v2.go

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