diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index 8752cd267..813bc3995 100644 --- a/.github/workflows/go.yml +++ b/.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 diff --git a/cmd/data-scanner.go b/cmd/data-scanner.go index 0475c4a53..8bbfd8f31 100644 --- a/cmd/data-scanner.go +++ b/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) { diff --git a/cmd/erasure-healing.go b/cmd/erasure-healing.go index 8fb371270..e02a075f1 100644 --- a/cmd/erasure-healing.go +++ b/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, } diff --git a/cmd/erasure-server-pool-decom.go b/cmd/erasure-server-pool-decom.go index 0e422c3ae..abae13c0d 100644 --- a/cmd/erasure-server-pool-decom.go +++ b/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, }) diff --git a/cmd/erasure-server-pool-rebalance.go b/cmd/erasure-server-pool-rebalance.go index b60c9c904..f352c5f93 100644 --- a/cmd/erasure-server-pool-rebalance.go +++ b/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, }) diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index 137204f97..5012379a1 100644 --- a/cmd/erasure-server-pool.go +++ b/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, } diff --git a/cmd/global-heal.go b/cmd/global-heal.go index ca78df818..e153418c1 100644 --- a/cmd/global-heal.go +++ b/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) diff --git a/cmd/metacache-entries.go b/cmd/metacache-entries.go index fceec0b43..b94de924c 100644 --- a/cmd/metacache-entries.go +++ b/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 { diff --git a/cmd/metacache-listing-cache_test.go b/cmd/metacache-listing-cache_test.go new file mode 100644 index 000000000..ef6b24e55 --- /dev/null +++ b/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 . + +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) + } +} diff --git a/cmd/metacache-listing-delete_test.go b/cmd/metacache-listing-delete_test.go new file mode 100644 index 000000000..9afe58ef3 --- /dev/null +++ b/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 . + +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) + } +} diff --git a/cmd/metacache-listing-http_test.go b/cmd/metacache-listing-http_test.go new file mode 100644 index 000000000..3af2f0ca0 --- /dev/null +++ b/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 . + +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) + } + }) + } +} diff --git a/cmd/metacache-listing-lock-routes_test.go b/cmd/metacache-listing-lock-routes_test.go new file mode 100644 index 000000000..040fc5139 --- /dev/null +++ b/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 . + +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) + }) + } +} diff --git a/cmd/metacache-listing-locks_test.go b/cmd/metacache-listing-locks_test.go new file mode 100644 index 000000000..bd5998467 --- /dev/null +++ b/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 . + +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 +} diff --git a/cmd/metacache-listing-merge_test.go b/cmd/metacache-listing-merge_test.go new file mode 100644 index 000000000..da71e8fb1 --- /dev/null +++ b/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 . + +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: + } + } +} diff --git a/cmd/metacache-null-quorum_test.go b/cmd/metacache-null-quorum_test.go index a0aa09308..8f1cc002c 100644 --- a/cmd/metacache-null-quorum_test.go +++ b/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] diff --git a/cmd/metacache-server-pool.go b/cmd/metacache-server-pool.go index c62f822ce..11bbb80a8 100644 --- a/cmd/metacache-server-pool.go +++ b/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 } diff --git a/cmd/metacache-set.go b/cmd/metacache-set.go index a2ab603c6..b7fa58580 100644 --- a/cmd/metacache-set.go +++ b/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 { diff --git a/cmd/xl-storage-format-v2-cross-parity_test.go b/cmd/xl-storage-format-v2-cross-parity_test.go new file mode 100644 index 000000000..62dd6634f --- /dev/null +++ b/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 . + +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) + } + }) + } +} diff --git a/cmd/xl-storage-format-v2.go b/cmd/xl-storage-format-v2.go index 3ce72da24..77248f72a 100644 --- a/cmd/xl-storage-format-v2.go +++ b/cmd/xl-storage-format-v2.go @@ -1944,7 +1944,7 @@ func mergeXLV2Versions(quorum int, strict bool, requestedVersions int, versions // Keep the original stream shapes: pruning must not make a versioned object // eligible for the single null-version recount below. originalVersions := versions - var checkedSingleNull, singleNull bool + var checkedSingleNull, singleNull, mixedECNull bool // Shallow copy input versions = append(make([][]xlMetaV2ShallowVersion, 0, len(versions)), versions...) @@ -2019,19 +2019,24 @@ func mergeXLV2Versions(quorum int, strict bool, requestedVersions int, versions // Version IDs match, but otherwise unable to resolve. // We are either strict, or don't have enough information to match. // Switch to a pure counting algo. - latest, latestCount = countXLV2Versions(tops, ver.header, latest, strict) + latest, latestCount = countXLV2Versions(tops, ver.header, latest, strict, false) break } } if latestCount < quorum { if !checkedSingleNull { - singleNull = singleNullVersionStreams(originalVersions) + singleNull = singleNullVersionStreams(originalVersions, false) + // Availability-optimized writes can change EC while a drive + // is offline. Only latest-only non-strict reads may recount + // these ordinary null versions across separate modern EC groups. + mixedECNull = !singleNull && !strict && requestedVersions == 1 && + singleNullVersionStreams(originalVersions, true) checkedSingleNull = true } - if singleNull { + if singleNull || mixedECNull { // A newer minority at the end can hide an older quorum from // the selection loop. Recount before discarding the null ID. - if candidate, count := countXLV2Versions(tops, latest.header, latest, strict); count >= quorum { + if candidate, count := countXLV2Versions(tops, latest.header, latest, strict, mixedECNull); count >= quorum { latest, latestCount = candidate, count } } @@ -2098,7 +2103,8 @@ func mergeXLV2Versions(quorum int, strict bool, requestedVersions int, versions // singleNullVersionStreams excludes histories and other version types from the // additional recount. Check the original inputs, before any stream is pruned. -func singleNullVersionStreams(versions [][]xlMetaV2ShallowVersion) bool { +// Mixed EC groups require known modern EC headers on every non-empty stream. +func singleNullVersionStreams(versions [][]xlMetaV2ShallowVersion, allowMixedEC bool) bool { var ec xlMetaV2VersionHeader var haveEC bool for _, stream := range versions { @@ -2112,7 +2118,10 @@ func singleNullVersionStreams(versions [][]xlMetaV2ShallowVersion) bool { if h.VersionID != [16]byte{} || h.Type != ObjectType || h.FreeVersion() { return false } - if haveEC && (h.EcN != ec.EcN || h.EcM != ec.EcM) { + if allowMixedEC && !h.hasEC() { + return false + } + if !allowMixedEC && haveEC && (h.EcN != ec.EcN || h.EcM != ec.EcM) { return false } ec, haveEC = h, true @@ -2121,16 +2130,18 @@ func singleNullVersionStreams(versions [][]xlMetaV2ShallowVersion) bool { } // countXLV2Versions selects the most frequent compatible header for reference's -// VersionID, retaining the existing sort tiebreak and last matching entry. -func countXLV2Versions(tops []xlMetaV2ShallowVersion, reference xlMetaV2VersionHeader, latest xlMetaV2ShallowVersion, strict bool) (xlMetaV2ShallowVersion, int) { +// VersionID and the last matching entry. The mixed-EC null-version recount adds +// a stable EC tiebreak without changing the ordinary counting path. +func countXLV2Versions(tops []xlMetaV2ShallowVersion, reference xlMetaV2VersionHeader, latest xlMetaV2ShallowVersion, strict, allowMixedEC bool) (xlMetaV2ShallowVersion, int) { x := make(map[xlMetaV2VersionHeader]int, len(tops)) for _, a := range tops { if a.header.VersionID != reference.VersionID { continue } if !strict { - // we must match EC, when we are not strict. - if !a.header.matchesEC(reference) { + // The narrow null-version recount can consider other EC groups, + // but EcN/EcM stay in the map key: their votes never combine. + if !allowMixedEC && !a.header.matchesEC(reference) { continue } a.header.Signature = [4]byte{} @@ -2142,9 +2153,19 @@ func countXLV2Versions(tops []xlMetaV2ShallowVersion, reference xlMetaV2VersionH if v < latestCount { continue } - if v == latestCount && latest.header.sortsBefore(k) { - // Tiebreak, use sort. - continue + if v == latestCount { + if allowMixedEC { + current := latest.header + current.Signature = [4]byte{} + // Signatures are not part of these groups. Use EC counts as + // the final tiebreak when the existing sort fields are equal. + if current.sortsBefore(k) || !k.sortsBefore(current) && + (current.EcM > k.EcM || current.EcM == k.EcM && current.EcN > k.EcN) { + continue + } + } else if latest.header.sortsBefore(k) { + continue + } } for _, a := range tops { hdr := a.header