Fix cross-parity version listing after pool removal

Signed-off-by: Feng Ruohang <rh@vonng.com>

.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

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) {

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(`<VersioningConfiguration><Status>Suspended</Status></VersioningConfiguration>`)); 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)

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) {
+			// Latest-only listing and listing all versions (decommission,
-				got := mergeXLV2Versions(tt.quorum, false, 1, streams...)
+			// rebalance) must select the same version.
-				if tt.want < 0 {
+			for _, requested := range []int{1, 0} {
-					if len(got) != 0 {
+				for order, streams := range quorumVersionOrders(input) {
-						t.Fatalf("order %d: subquorum groups were combined: %v", order, got)
+					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 {
+			for _, requested := range []int{1, 0} {
-				t.Fatalf("excluded input changed: %v", got)
+				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})
+			newTime := nullQuorumCrossParityOverwrite(t, z, 0, bucket, object, oldBody, newBody, mtime, ObjectOptions{})
-			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 {
+				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)
+			}
+		})
+	}
+}

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
+					// is offline. Only non-strict reads may recount these
-					// these ordinary null versions across separate modern EC groups.
+					// ordinary null versions across separate modern EC groups.
-					mixedECNull = !singleNull && !strict && requestedVersions == 1 &&
+					// Each stream holds just this version, so listing all
+					// versions (decommission, rebalance) must recount too.
+					mixedECNull = !singleNull && !strict &&
 						singleNullVersionStreams(originalVersions, true)
 					checkedSingleNull = true
 				}