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