From 2a4db6a4f9d0dedf7229a3b1c5866d2a1bbb22ab Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Fri, 16 Oct 2026 01:23:29 +0800 Subject: [PATCH] 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 --- cmd/metacache-listing-cache_test.go | 253 ++++++++++++++++++--------- cmd/metacache-listing-delete_test.go | 4 +- cmd/metacache-listing-merge_test.go | 9 +- cmd/metacache-null-quorum_test.go | 6 +- cmd/metacache-server-pool.go | 8 +- 5 files changed, 187 insertions(+), 93 deletions(-) diff --git a/cmd/metacache-listing-cache_test.go b/cmd/metacache-listing-cache_test.go index ef6b24e55..68140fa8c 100644 --- a/cmd/metacache-listing-cache_test.go +++ b/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) - 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) - } + testListingCacheFallbackError(t, continuation, false) }) } } +func TestListingCacheCleanupKeepsOriginalID(t *testing.T) { + testListingCacheFallbackError(t, true, true) +} + +// Use actual peer RPCs to verify the abandoned cache is updated after the +// caller clears its own listing ID. The local-only path cannot observe this. +func recordListingPeerCleanup(t *testing.T) func() { + t.Helper() + tg, err := grid.SetupTestGrid(2) + if err != nil { + t.Fatal(err) + } + previous := globalNotificationSys + t.Cleanup(func() { + tg.Cleanup() + globalNotificationSys = previous + }) + getIDs := make(chan string, 8) + type cleanupResult struct { + id string + err *grid.RemoteErr + } + updated := make(chan cleanupResult, 1) + server := &peerRESTServer{} + if err := getMetacacheListingRPC.Register(tg.Managers[1], func(o *listPathOptions) (*metacache, *grid.RemoteErr) { + getIDs <- o.ID + return server.GetMetacacheListingHandler(o) + }); err != nil { + t.Fatal(err) + } + if err := updateMetacacheListingRPC.Register(tg.Managers[1], func(m *metacache) (*metacache, *grid.RemoteErr) { + result, err := server.UpdateMetacacheListingHandler(m) + if m.error == "no longer used" { + updated <- cleanupResult{id: m.id, err: err} + } + return result, err + }); err != nil { + t.Fatal(err) + } + client := &peerRESTClient{gridConn: func() *grid.Connection { return tg.Managers[0].Connection(tg.Hosts[1]) }} + globalNotificationSys = &NotificationSys{peerClients: []*peerRESTClient{client}, allPeerClients: []*peerRESTClient{client}} + return func() { + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + var ids []string + for range 2 { + select { + case id := <-getIDs: + ids = append(ids, id) + case <-ctx.Done(): + t.Fatal("peer did not receive the failed cache cleanup lookup") + } + } + if ids[0] == "" || ids[1] != ids[0] { + t.Fatalf("cleanup lookup lost the original cache ID: %q", ids) + } + select { + case result := <-updated: + if result.err != nil || result.id != ids[0] { + t.Fatalf("cleanup update: id=%q err=%v, want id=%q", result.id, result.err, ids[0]) + } + t.Logf("peer cleanup Get/Update retained cache ID %q", result.id) + case <-ctx.Done(): + t.Fatal("peer did not finish the failed cache cleanup update") + } + } +} + +func testListingCacheFallbackError(t *testing.T, continuation, peer bool) { + t.Helper() + z, bucket, router := nullQuorumBackend(t) + var waitPeerCleanup func() + if peer { + waitPeerCleanup = recordListingPeerCleanup(t) + } + set := z.serverPools[0].sets[0] + disks, switchDisks := installListingDiskSwitch(set) + body := bytes.Repeat([]byte{'a'}, 8192) + mtime := time.Now().UTC().Add(-time.Hour) + for _, object := range []string{"a", "b", "z"} { + nullQuorumPut(t, z, bucket, object, body, ObjectOptions{MTime: mtime}) + } + old := make([][]byte, 3) + for i := range old { + old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, "z") + } + nullQuorumPut(t, z, bucket, "z", body, ObjectOptions{MTime: mtime.Add(time.Minute)}) + var token string + if continuation { + rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1", "", "", "")) + var list ListObjectsV2Response + if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil || !list.IsTruncated || list.NextContinuationToken == "" { + t.Fatalf("first page: %d %s", rec.Code, rec.Body.String()) + } + token = list.NextContinuationToken + } + for i := range old { + if err := disks[i].WriteAll(t.Context(), bucket, "z/"+xlStorageFormatFile, old[i]); err != nil { + t.Fatal(err) + } + } + var reads atomic.Int64 + walkers := slices.Clone(disks) + for i := range walkers { + walkers[i] = &listingReadFailDisk{StorageAPI: disks[i], object: "z", reads: &reads} + if i >= 8 { + walkers[i] = &failedMetadataWalkDisk{StorageAPI: walkers[i]} + } + } + switchDisks(walkers) + target := getListObjectsV2URL("", bucket, "", "1000", "", "", "") + if token != "" { + target += "&continuation-token=" + url.QueryEscape(token) + } + rec := nullQuorumRequest(t, router, http.MethodGet, target) + t.Logf("LIST=%d failed metadata reads=%d body=%s", rec.Code, reads.Load(), rec.Body.String()) + if reads.Load() == 0 { + t.Fatal("fallback not exercised") + } + var apiErr APIErrorResponse + if rec.Code != http.StatusServiceUnavailable || xml.Unmarshal(rec.Body.Bytes(), &apiErr) != nil || apiErr.Code != "SlowDownRead" { + t.Fatalf("fallback failure lost its read-quorum error: %d %s", rec.Code, rec.Body.String()) + } + if waitPeerCleanup != nil { + waitPeerCleanup() + } + if continuation { + caches, _ := localMetacacheMgr.getBucket(t.Context(), bucket).cloneCaches() + if len(caches) == 0 { + t.Fatal("continuation did not create a cache") + } + for _, cache := range caches { + if cache.status != scanStateError { + t.Fatalf("incomplete cache was not marked failed: %+v", cache) + } + } + } + switchDisks(disks) + retry := nullQuorumRequest(t, router, http.MethodGet, target) + var recovered ListObjectsV2Response + if retry.Code != http.StatusOK || xml.Unmarshal(retry.Body.Bytes(), &recovered) != nil { + t.Fatalf("retry after metadata recovers: %d %s", retry.Code, retry.Body.String()) + } + var keys []string + for _, entry := range recovered.Contents { + keys = append(keys, entry.Key) + } + want := []string{"a", "b", "z"} + if continuation { + want = []string{"b", "z"} + } + if !slices.Equal(keys, want) || recovered.IsTruncated { + t.Fatalf("retry returned incomplete result: %+v", recovered) + } +} + func TestListingCachePagination(t *testing.T) { z, bucket, router := nullQuorumBackend(t) for _, key := range []string{"a", "b", "c", "d"} { @@ -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) diff --git a/cmd/metacache-listing-delete_test.go b/cmd/metacache-listing-delete_test.go index 9afe58ef3..74cfa2b5e 100644 --- a/cmd/metacache-listing-delete_test.go +++ b/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) diff --git a/cmd/metacache-listing-merge_test.go b/cmd/metacache-listing-merge_test.go index da71e8fb1..9fe4f349a 100644 --- a/cmd/metacache-listing-merge_test.go +++ b/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) diff --git a/cmd/metacache-null-quorum_test.go b/cmd/metacache-null-quorum_test.go index 8f1cc002c..fb0b18e5f 100644 --- a/cmd/metacache-null-quorum_test.go +++ b/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) } diff --git a/cmd/metacache-server-pool.go b/cmd/metacache-server-pool.go index 11bbb80a8..6510aaeba 100644 --- a/cmd/metacache-server-pool.go +++ b/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 = "" }