From a7310d59fe307ae7fc4b57dc190b5e0c3de1a56a Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sun, 1 Nov 2026 14:08:09 +0800 Subject: [PATCH] Fix cross-parity version listing after pool removal (#267) Availability-optimized writes can leave a readable null version across separate EC groups when a disk is unavailable. Recount ordinary null versions for all non-strict reads and cover version listing and complete pool removal with end-to-end tests. Signed-off-by: Feng Ruohang --- .github/workflows/go.yml | 2 +- cmd/erasure-server-pool-migration_test.go | 83 +++++++++++ cmd/metacache-null-quorum_test.go | 65 ++++++++- cmd/xl-storage-format-v2-cross-parity_test.go | 133 +++++++++++++----- cmd/xl-storage-format-v2.go | 8 +- 5 files changed, 253 insertions(+), 38 deletions(-) diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index 6b7824a24..2bf27c930 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -102,7 +102,7 @@ jobs: run: go test -race ./cmd -run '^Test(Multipart|PaginateMultipartUploads|ListMultipartUploads|CleanupStaleUploads|ExtractModTime)' -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|ListingRecovery|ListingPending|ListingEntry|ListingSecondLock|StreamMetadataParts|LocalLockerListing|DistributedListing|MergeXLV2LatestNullCrossParity|CountXLV2MixedECGroups|ListCrossParity)' -count=1 -timeout=5m + run: go test -race ./cmd -run '^Test(ListFallback|ListingCache|ListingMerge|ListingRecovery|ListingPending|ListingEntry|ListingSecondLock|StreamMetadataParts|LocalLockerListing|DistributedListing|MergeXLV2LatestNullCrossParity|CountXLV2MixedECGroups|ListCrossParity|ListObjectVersionsCrossParityNull|MigrationCrossParity)' -count=1 -timeout=5m - name: Run lock cancellation and late-grant cleanup tests under race detector run: go test -race ./internal/dsync ./internal/lsync -run '^Test(DRWMutex|LRWMutex)' -count=1 -timeout=5m diff --git a/cmd/erasure-server-pool-migration_test.go b/cmd/erasure-server-pool-migration_test.go index cf3c7834c..2f6bba618 100644 --- a/cmd/erasure-server-pool-migration_test.go +++ b/cmd/erasure-server-pool-migration_test.go @@ -25,6 +25,7 @@ import ( "io" "maps" "net/http" + "slices" "strings" "sync" "testing" @@ -906,6 +907,88 @@ func TestPoolsMigrationRetiresSupersededCopy(t *testing.T) { } } +func TestPoolsMigrationAfterCompleteRemoval(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + dirs, err := getRandomDisks(32) + if err != nil { + t.Fatal(err) + } + previousEndpoints := globalEndpoints + endpoints := mustGetPoolEndpoints(0, dirs[:16]...) + endpoints = append(endpoints, mustGetPoolEndpoints(1, dirs[16:]...)...) + globalEndpoints = endpoints + obj, _, err := initObjectLayer(ctx, endpoints) + if err != nil { + globalEndpoints = previousEndpoints + removeRoots(dirs) + t.Fatal(err) + } + z := obj.(*erasureServerPools) + previous := newObjectLayerFn() + setObjectLayer(z) + var reopened *erasureServerPools + oldShutdown := false + t.Cleanup(func() { + if reopened != nil { + _ = reopened.Shutdown(context.Background()) + } + if !oldShutdown { + _ = z.Shutdown(context.Background()) + } + removeRoots(dirs) + globalEndpoints = previousEndpoints + setObjectLayer(previous) + }) + + bucket, _, err := initAPIHandlerTest(ctx, z, nil, MakeBucketOptions{}) + if err != nil { + t.Fatal(err) + } + const object = "object" + mtime := time.Now().UTC().Add(-time.Hour) + oldBody := bytes.Repeat([]byte{'a'}, 64<<10) + newBody := bytes.Repeat([]byte{'b'}, 256<<10) + nullQuorumCrossParityOverwrite(t, z, 0, bucket, object, oldBody, newBody, mtime, ObjectOptions{}) + + z.poolMetaMutex.Lock() + z.poolMeta.Pools[0].Decommission = &PoolDecommissionInfo{} + z.poolMetaMutex.Unlock() + if err := z.decommissionPool(ctx, 0, z.serverPools[0], decomBucketInfo{Name: bucket}); err != nil { + t.Fatal(err) + } + if err := z.CompleteDecommission(ctx, 0); err != nil { + t.Fatal(err) + } + if err := z.Shutdown(context.Background()); err != nil { + t.Fatal(err) + } + oldShutdown = true + + globalEndpoints = mustGetPoolEndpoints(0, dirs[16:]...) + reopenedObj, _, err := initObjectLayer(t.Context(), globalEndpoints) + if err != nil { + t.Fatal(err) + } + reopened = reopenedObj.(*erasureServerPools) + setObjectLayer(reopened) + _, router, err := initAPIHandlerTest(t.Context(), reopened, nil, MakeBucketOptions{}) + if err != nil { + t.Fatal(err) + } + + get := nullQuorumRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object)) + if get.Code != http.StatusOK || !bytes.Equal(get.Body.Bytes(), newBody) { + t.Fatalf("GET after pool removal: status=%d bytes=%d", get.Code, get.Body.Len()) + } + if code, keys := nullQuorumListV2Keys(t, router, bucket); code != http.StatusOK || !slices.Equal(keys, []string{object}) { + t.Fatalf("ListObjectsV2 after pool removal: status=%d keys=%v", code, keys) + } + if code, versions := nullQuorumListVersions(t, router, bucket); code != http.StatusOK || len(versions) != 1 || versions[0].Key != object || versions[0].VersionID != nullVersionID || versions[0].Size != int64(len(newBody)) { + t.Fatalf("ListObjectVersions after pool removal: status=%d versions=%+v", code, versions) + } +} + // A tiered move that fails, here because another pool cannot be read, must keep // its source like any other failed version. func TestPoolsMigrationTieredFailureKeepsSource(t *testing.T) { diff --git a/cmd/metacache-null-quorum_test.go b/cmd/metacache-null-quorum_test.go index ee88f0cfa..38cff9f9c 100644 --- a/cmd/metacache-null-quorum_test.go +++ b/cmd/metacache-null-quorum_test.go @@ -117,6 +117,10 @@ func (d *nullQuorumWalkDisk) WalkDir(ctx context.Context, opts WalkDirOptions, o } func nullQuorumBackend(t *testing.T) (*erasureServerPools, string, http.Handler) { + return nullQuorumBackendWithVersioning(t, false) +} + +func nullQuorumBackendWithVersioning(t *testing.T, suspended bool) (*erasureServerPools, string, http.Handler) { t.Helper() ctx, cancel := context.WithCancel(t.Context()) obj, dirs, err := prepareErasure16(ctx) @@ -133,10 +137,16 @@ func nullQuorumBackend(t *testing.T) (*erasureServerPools, string, http.Handler) removeRoots(dirs) setObjectLayer(previous) }) - bucket, router, err := initAPIHandlerTest(ctx, z, nil, MakeBucketOptions{}) + bucket, router, err := initAPIHandlerTest(ctx, z, nil, MakeBucketOptions{VersioningEnabled: suspended}) if err != nil { t.Fatal(err) } + if suspended { + if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketVersioningConfig, + []byte(`Suspended`)); err != nil { + t.Fatal(err) + } + } return z, bucket, router } @@ -158,6 +168,36 @@ func nullQuorumPut(t *testing.T, z *erasureServerPools, bucket, object string, b } } +func nullQuorumCrossParityOverwrite(t *testing.T, z *erasureServerPools, pool int, bucket, object string, oldBody, newBody []byte, oldTime time.Time, opts ObjectOptions) time.Time { + t.Helper() + set := z.serverPools[pool].getHashedSet(object) + disks := set.getDisks() + put := func(body []byte, opts ObjectOptions) { + if _, err := z.serverPools[pool].PutObject(t.Context(), bucket, object, + mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), opts); err != nil { + t.Fatal(err) + } + } + oldOpts := opts + oldOpts.MTime = oldTime + oldOpts.MaxParity = false + put(oldBody, oldOpts) + old := make([][]byte, len(disks)) + for i, disk := range disks { + old[i] = mustReadNullQuorumMeta(t, disk, bucket, object) + } + newTime := oldTime.Add(time.Minute) + opts.MTime = newTime + opts.MaxParity = true + put(newBody, opts) + for i := 12; i < len(disks); i++ { + if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, old[i]); err != nil { + t.Fatal(err) + } + } + return newTime +} + func nullQuorumListV2Keys(t *testing.T, router http.Handler, bucket string) (int, []string) { t.Helper() rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", "")) @@ -175,6 +215,29 @@ func nullQuorumListV2Keys(t *testing.T, router http.Handler, bucket string) (int return rec.Code, keys } +type nullQuorumListedVersion struct { + Key string + VersionID string `xml:"VersionId"` + LastModified string + Size int64 + IsLatest bool +} + +func nullQuorumListVersions(t *testing.T, router http.Handler, bucket string) (int, []nullQuorumListedVersion) { + t.Helper() + rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectVersionsURL("", bucket, "", "1000", "")) + if rec.Code != http.StatusOK { + return rec.Code, nil + } + var list struct { + Versions []nullQuorumListedVersion `xml:"Version"` + } + if err := xml.Unmarshal(rec.Body.Bytes(), &list); err != nil { + t.Fatal(err) + } + return rec.Code, list.Versions +} + func TestListObjectsSingleNullQuorumHTTP(t *testing.T) { z, bucket, router := nullQuorumBackend(t) oldBody := bytes.Repeat([]byte{'a'}, 8192) diff --git a/cmd/xl-storage-format-v2-cross-parity_test.go b/cmd/xl-storage-format-v2-cross-parity_test.go index 62dd6634f..ac96f8c1d 100644 --- a/cmd/xl-storage-format-v2-cross-parity_test.go +++ b/cmd/xl-storage-format-v2-cross-parity_test.go @@ -20,6 +20,7 @@ package cmd import ( "bytes" "fmt" + "io" "net/http" "slices" "testing" @@ -73,21 +74,25 @@ func TestMergeXLV2LatestNullCrossParity(t *testing.T) { } } 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) + // Latest-only listing and listing all versions (decommission, + // rebalance) must select the same version. + for _, requested := range []int{1, 0} { + for order, streams := range quorumVersionOrders(input) { + got := mergeXLV2Versions(tt.quorum, false, requested, streams...) + if tt.want < 0 { + if len(got) != 0 { + t.Fatalf("requested %d, order %d: subquorum groups were combined: %v", requested, order, got) + } + continue + } + if len(got) != 1 || got[0].header != versions[tt.want].header { + t.Fatalf("requested %d, order %d: got %v, want group %d", requested, 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("requested %d, order %d: selected metadata is invalid: %+v, %v", requested, order, fi, err) } - 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) } } }) @@ -97,15 +102,13 @@ func TestMergeXLV2LatestNullCrossParity(t *testing.T) { 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"} { + for _, kind := range []string{"strict", "uuid", "delete", "free", "history", "legacy-modern-ec", "partial-ec"} { t.Run(kind, func(t *testing.T) { a, b := old, newer - strict, requested := false, 1 + strict := false 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": @@ -140,8 +143,10 @@ func TestMergeXLV2LatestNullCrossParityScope(t *testing.T) { 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) + for _, requested := range []int{1, 0} { + if got := mergeXLV2Versions(8, strict, requested, input...); len(got) != 0 { + t.Fatalf("requested %d: excluded input changed: %v", requested, got) + } } }) } @@ -189,25 +194,14 @@ func TestListCrossParityReadableMajority(t *testing.T) { 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) - } - } + newTime := nullQuorumCrossParityOverwrite(t, z, 0, bucket, object, oldBody, newBody, mtime, ObjectOptions{}) 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 { + if err != nil || !fi.ModTime.Equal(newTime) || fi.Erasure.DataBlocks != 8 { t.Fatalf("readData=%v: full metadata read lost readable majority: %+v, %v", readData, fi, err) } } @@ -220,3 +214,76 @@ func TestListCrossParityReadableMajority(t *testing.T) { }) } } + +func TestListObjectVersionsCrossParityNull(t *testing.T) { + for _, tc := range []struct { + name string + suspended bool + }{ + {name: "unversioned"}, + {name: "suspended", suspended: true}, + } { + t.Run(tc.name, func(t *testing.T) { + z, bucket, router := nullQuorumBackendWithVersioning(t, tc.suspended) + const object = "object" + mtime := time.Now().UTC().Add(-time.Hour) + oldBody := bytes.Repeat([]byte{'a'}, 64<<10) + newBody := bytes.Repeat([]byte{'b'}, 256<<10) + nullQuorumCrossParityOverwrite(t, z, 0, bucket, object, oldBody, newBody, mtime, + ObjectOptions{VersionSuspended: tc.suspended}) + + code, versions := nullQuorumListVersions(t, router, bucket) + if code != http.StatusOK || len(versions) != 1 { + t.Fatalf("ListObjectVersions: status=%d versions=%+v", code, versions) + } + got := versions[0] + if got.Key != object || got.VersionID != nullVersionID || !got.IsLatest || got.Size != int64(len(newBody)) || got.LastModified == "" { + t.Fatalf("ListObjectVersions selected the wrong null version: %+v", got) + } + }) + } +} + +// Decommission and rebalance list every version. A readable degraded overwrite +// must not be skipped there, or removing the pool loses it. +func TestMigrationCrossParityReadableMajority(t *testing.T) { + z, bucket, _ := nullQuorumBackend(t) + set := z.serverPools[0].sets[0] + const object = "object" + mtime := time.Now().UTC().Add(-time.Hour) + oldBody := bytes.Repeat([]byte{'a'}, 256<<10) + newBody := bytes.Repeat([]byte{'b'}, 256<<10) + nullQuorumCrossParityOverwrite(t, z, 0, bucket, object, oldBody, newBody, mtime, ObjectOptions{}) + for _, kind := range []string{"rebalance", "decommission"} { + t.Run(kind, func(t *testing.T) { + var names []string + consume := func(entry metaCacheEntry) { + versions, err := entry.fileInfoVersions(bucket) + if err != nil || len(versions.Versions) != 1 || !versions.Versions[0].ModTime.Equal(mtime.Add(time.Minute)) { + t.Errorf("invalid migration metadata: %+v, %v", versions, err) + return + } + reader, err := set.GetObjectNInfo(t.Context(), bucket, entry.name, nil, nil, ObjectOptions{VersionID: nullVersionID, NoLock: true, NoDecryption: true}) + if err != nil { + t.Error(err) + return + } + body, err := io.ReadAll(reader) + reader.Close() + if err != nil || !bytes.Equal(body, newBody) { + t.Errorf("migration could not read selected object data: %v", err) + } + names = append(names, entry.name) + } + var err error + if kind == "rebalance" { + err = set.listObjectsToRebalance(t.Context(), bucket, consume) + } else { + err = set.listObjectsToDecommission(t.Context(), decomBucketInfo{Name: bucket}, consume) + } + if err != nil || !slices.Equal(names, []string{object}) { + t.Fatalf("migration listing skipped a readable object: %v, %v", names, err) + } + }) + } +} diff --git a/cmd/xl-storage-format-v2.go b/cmd/xl-storage-format-v2.go index 77248f72a..3bad89161 100644 --- a/cmd/xl-storage-format-v2.go +++ b/cmd/xl-storage-format-v2.go @@ -2027,9 +2027,11 @@ func mergeXLV2Versions(quorum int, strict bool, requestedVersions int, versions if !checkedSingleNull { 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 && + // is offline. Only non-strict reads may recount these + // ordinary null versions across separate modern EC groups. + // Each stream holds just this version, so listing all + // versions (decommission, rebalance) must recount too. + mixedECNull = !singleNull && !strict && singleNullVersionStreams(originalVersions, true) checkedSingleNull = true }