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>

cmd/metacache-listing-cache_test.go

@@ -29,6 +29,8 @@ import (
 	"sync/atomic"
 	"testing"
 	"time"
+
+	"github.com/minio/minio/internal/grid"
 )
 
 type listingReadFailDisk struct {
@@ -45,6 +47,17 @@ func (d *listingReadFailDisk) ReadXL(ctx context.Context, bucket, object string,
 	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"
@@ -52,87 +65,167 @@ func TestListingCachePreservesFallbackError(t *testing.T) {
 			name = "continuation"
 		}
 		t.Run(name, func(t *testing.T) {
-			z, bucket, router := nullQuorumBackend(t)
+			testListingCacheFallbackError(t, continuation, false)
-			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 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"} {
@@ -172,6 +265,7 @@ func TestListingCachePagination(t *testing.T) {
 // 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{})
 	}
@@ -180,16 +274,15 @@ func TestListingCacheFullPageDoesNotWaitForTail(t *testing.T) {
 	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())
+	walkers := slices.Clone(disks)
 	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 }
+	switchDisks(walkers)
 	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
 	defer cancel()
 	target := getListObjectsV2URL("", bucket, "", "1", "", "", "") + "&continuation-token=" + url.QueryEscape(page.NextContinuationToken)

cmd/metacache-listing-delete_test.go

@@ -33,6 +33,7 @@ func TestListFallbackPreservesPoolDeleteMarker(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)
@@ -42,7 +43,6 @@ func TestListFallbackPreservesPoolDeleteMarker(t *testing.T) {
 					t.Fatal(err)
 				}
 			}
-			disks := deleted.getDisks()
 			old := make([][]byte, 3)
 			for i := range old {
 				old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, object)
@@ -69,7 +69,7 @@ func TestListFallbackPreservesPoolDeleteMarker(t *testing.T) {
 			for i := 8; i < len(walkers); i++ {
 				walkers[i] = &failedMetadataWalkDisk{StorageAPI: disks[i]}
 			}
-			deleted.getDisks = func() []StorageAPI { return walkers }
+			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)

cmd/metacache-listing-merge_test.go

@@ -116,6 +116,8 @@ func TestListingMergeDoesNotSkipFailedPool(t *testing.T) {
 	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) {
@@ -135,7 +137,6 @@ func TestListingMergeDoesNotSkipFailedPool(t *testing.T) {
 		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")
@@ -154,14 +155,14 @@ func TestListingMergeDoesNotSkipFailedPool(t *testing.T) {
 			brokenWalkers[i] = &failedMetadataWalkDisk{StorageAPI: brokenWalkers[i]}
 		}
 	}
-	broken.getDisks = func() []StorageAPI { return brokenWalkers }
+	switchBrokenDisks(brokenWalkers)
 	resume := make(chan struct{})
 	defer close(resume)
-	healthyWalkers := slices.Clone(healthy.getDisks())
+	healthyWalkers := slices.Clone(healthyDisks)
 	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 }
+	switchHealthyDisks(healthyWalkers)
 
 	target := getListObjectsV2URL("", bucket, "", "2", "", "", "") + "&continuation-token=" + url.QueryEscape(firstToken)
 	req, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)

cmd/metacache-null-quorum_test.go

@@ -596,7 +596,7 @@ func TestListFallbackDoesNotTrustNotFound(t *testing.T) {
 func TestListCrossParityMinorityStaysVisible(t *testing.T) {
 	z, bucket, router := nullQuorumBackend(t)
 	set := z.serverPools[0].sets[0]
-	disks := set.getDisks()
+	disks, switchDisks := installListingDiskSwitch(set)
 	const object = "object"
 	mtime := time.Now().UTC().Add(-time.Hour)
 
@@ -619,7 +619,7 @@ func TestListCrossParityMinorityStaysVisible(t *testing.T) {
 	for i := 12; i < len(online); i++ {
 		online[i] = &offlineNullQuorumDisk{StorageAPI: disks[i]}
 	}
-	set.getDisks = func() []StorageAPI { return online }
+	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")
 	}
@@ -627,7 +627,7 @@ func TestListCrossParityMinorityStaysVisible(t *testing.T) {
 		t.Errorf("ListObjectsV2 with a node away: got %d %v; want 200 [%s]", code, keys, object)
 	}
 
-	set.getDisks = func() []StorageAPI { return disks }
+	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)
 	}

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 = ""
 	}