Merge pull request #264 from botdeniska/fix/free-versions-not-versions

fix(ilm): stop taking the free versions of deleted tiered versions for versions

.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/data-scanner.go

@@ -776,7 +776,7 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
 					}
 
 					// We got an entry which we should be able to heal.
-					fiv, err := entry.fileInfoVersions(bucket)
+					fiv, err := entry.allFileInfoVersions(bucket)
 					if err != nil {
 						err := bgSeq.queueHealTask(healSource{
 							bucket:    bucket,

cmd/erasure-server-pool-decom.go

@@ -982,6 +982,9 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 			if decommissioned == len(fivs.Versions) {
 				stopFn := globalDecommissionMetrics.log(decomMetricDecommissionRemoveObject, idx, bi.Name, entry.name)
 				err := z.removeMovedVersions(ctx, bi.Name, entry.name, idx, fivs.Versions)
+				if err == nil {
+					err = z.removeFreeVersions(ctx, bi.Name, entry.name, idx, fivs.FreeVersions)
+				}
 				stopFn(0, err)
 				auditLogDecom(ctx, "DecomDeleteObject", bi.Name, entry.name, "", err)
 				if err != nil {
@@ -1108,6 +1111,45 @@ func (z *erasureServerPools) decommissionInBackground(ctx context.Context, idx i
 	return nil
 }
 
+// removeFreeVersions removes the free versions fvs of object from the pool
+// idx being decommissioned, as the scanner removes free versions: first the
+// remote object of each, then the free version. A free version is not moved.
+// In another pool it would be a delete marker on top of the object, and once
+// this pool is gone nothing would remove its remote object. A remote object
+// that is already gone is not an error.
+func (z *erasureServerPools) removeFreeVersions(ctx context.Context, bucket, object string, idx int, fvs []FileInfo) error {
+	if len(fvs) == 0 {
+		return nil
+	}
+	for _, fv := range fvs {
+		err := deleteObjectFromRemoteTier(ctx, fv.TransitionedObjName, fv.TransitionVersionID, fv.TransitionTier)
+		if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
+			return fmt.Errorf("unable to remove the remote object %s of a free version of %s/%s from tier %s: %w",
+				fv.TransitionedObjName, bucket, object, fv.TransitionTier, err)
+		}
+	}
+	object = encodeDirObject(object)
+	lk := z.NewNSLock(bucket, object)
+	lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
+	if err != nil {
+		return err
+	}
+	ctx = lkctx.Context()
+	defer lk.Unlock(lkctx)
+	for _, fv := range fvs {
+		_, err := z.serverPools[idx].DeleteObject(ctx, bucket, object, ObjectOptions{
+			VersionID:        fv.VersionID,
+			InclFreeVersions: true,
+			NoLock:           true,
+			NoAuditLog:       true,
+		})
+		if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
+			return err
+		}
+	}
+	return nil
+}
+
 func (z *erasureServerPools) checkAfterDecom(ctx context.Context, idx int) error {
 	buckets, err := z.getBucketsToDecommission(ctx)
 	if err != nil {
@@ -1192,7 +1234,9 @@ func (z *erasureServerPools) checkAfterDecom(ctx context.Context, idx int) error
 					}
 				}
 
-				versionsFound += len(fivs.Versions) - ignored
+				// A free version left behind keeps a remote object that
+				// nothing removes once the pool is gone.
+				versionsFound += len(fivs.Versions) - ignored + len(fivs.FreeVersions)
 			}); err != nil {
 				return err
 			}

cmd/erasure-server-pool-free-version_test.go

@@ -0,0 +1,526 @@
+// Copyright (c) 2026 Denis Babchuk
+//
+// 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 <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"context"
+	"errors"
+	"slices"
+	"testing"
+	"time"
+
+	madmin "github.com/minio/madmin-go/v3"
+	"github.com/minio/minio/internal/bucket/lifecycle"
+	objectlock "github.com/minio/minio/internal/bucket/object/lock"
+)
+
+// Deleting a version whose data is in a remote tier leaves a free version in
+// its place: a record of the remote object, kept until the scanner removes
+// both. A free version is not a version of the object. There is no delete
+// marker, and the version below it is the latest one; reads already skip it.
+
+// freeVersionNoncurrent expires versions one day after they become noncurrent.
+const freeVersionNoncurrent = "<Rule><ID>noncurrent</ID><Status>Enabled</Status><Filter><Prefix></Prefix></Filter>" +
+	"<NoncurrentVersionExpiration><NoncurrentDays>1</NoncurrentDays></NoncurrentVersionExpiration></Rule>"
+
+// freeVersionTier is a remote tier held in memory whose removals can fail.
+type freeVersionTier struct {
+	*tierExpiryTestBackend
+	removeErr error
+}
+
+func (b *freeVersionTier) Remove(ctx context.Context, object string, rv remoteVersionID) error {
+	if b.removeErr != nil {
+		return b.removeErr
+	}
+	return b.tierExpiryTestBackend.Remove(ctx, object, rv)
+}
+
+type freeVersionTest struct {
+	t      *testing.T
+	z      *erasureServerPools
+	bucket string
+	tier   *freeVersionTier
+	labels map[string]string // version ID -> label
+}
+
+// newFreeVersionTest returns two pools with a versioned bucket under rules and
+// a remote tier WARM. In pool 0, obj-01 has v1 and v2 and a tiered v3, written
+// five, four and three days ago; the client deleted v3. obj-02 had only a
+// tiered v1, deleted as well.
+func newFreeVersionTest(t *testing.T, rules string) *freeVersionTest {
+	t.Helper()
+	z, _ := migrationTestPools(t)
+	ctx := t.Context()
+	f := &freeVersionTest{t: t, z: z, bucket: "free-version", labels: map[string]string{}}
+	if err := z.MakeBucket(ctx, f.bucket, MakeBucketOptions{VersioningEnabled: true}); err != nil {
+		t.Fatal(err)
+	}
+	if rules != "" {
+		meta, err := globalBucketMetadataSys.Get(f.bucket)
+		if err != nil {
+			t.Fatal(err)
+		}
+		meta.LifecycleConfigXML = []byte("<LifecycleConfiguration>" + rules + "</LifecycleConfiguration>")
+		if err := meta.Save(ctx, z); err != nil {
+			t.Fatal(err)
+		}
+		globalBucketMetadataSys.Set(f.bucket, meta)
+	}
+	f.tier = &freeVersionTier{tierExpiryTestBackend: &tierExpiryTestBackend{objects: map[string][]byte{}}}
+	prevTier := globalTierConfigMgr
+	globalTierConfigMgr = NewTierConfigMgr()
+	globalTierConfigMgr.drivercache["WARM"] = f.tier
+	t.Cleanup(func() { globalTierConfigMgr = prevTier })
+
+	now := UTCNow()
+	f.put("obj-01", "v1", now.AddDate(0, 0, -5))
+	f.put("obj-01", "v2", now.AddDate(0, 0, -4))
+	f.deleteVersion(f.tiered("obj-01", "v3", now.AddDate(0, 0, -3)))
+	f.deleteVersion(f.tiered("obj-02", "v1", now.AddDate(0, 0, -3)))
+	return f
+}
+
+// put writes a version of object to pool 0.
+func (f *freeVersionTest) put(object, label string, mtime time.Time) ObjectInfo {
+	f.t.Helper()
+	oi := putConsistencyObject(f.t, f.z, f.bucket, object, 0, object+" "+label, ObjectOptions{Versioned: true, MTime: mtime})
+	f.labels[oi.VersionID] = label
+	return oi
+}
+
+// tiered writes a version of object to pool 0 whose data is in the tier.
+func (f *freeVersionTest) tiered(object, label string, mtime time.Time) ObjectInfo {
+	f.t.Helper()
+	oi := f.put(object, label, mtime)
+	infos, errs := readAllFileInfo(f.t.Context(), f.z.serverPools[0].getHashedSet(object).getDisks(), "", f.bucket, object, oi.VersionID, false, false)
+	if errs[0] != nil {
+		f.t.Fatal(errs[0])
+	}
+	fi := infos[0]
+	fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName = lifecycle.TransitionComplete, "WARM", f.remote(object, label)
+	f.tier.objects[fi.TransitionedObjName] = []byte(object + " " + label)
+	if err := f.z.serverPools[0].DecomTieredObject(f.t.Context(), f.bucket, object, fi, ObjectOptions{}); err != nil {
+		f.t.Fatal(err)
+	}
+	return oi
+}
+
+// remote is the name of the remote object of a tiered version.
+func (f *freeVersionTest) remote(object, label string) string {
+	return f.bucket + "/" + object + "/" + label
+}
+
+// deleteVersion deletes a tiered version by its ID, as a client does, which
+// leaves a free version in its place.
+func (f *freeVersionTest) deleteVersion(oi ObjectInfo) {
+	f.t.Helper()
+	if _, err := f.z.DeleteObject(f.t.Context(), f.bucket, oi.Name, ObjectOptions{Versioned: true, VersionID: oi.VersionID}); err != nil {
+		f.t.Fatal(err)
+	}
+	if got := f.stored(0, oi.Name); len(got) == 0 || got[0] != "free" {
+		f.t.Fatalf("fixture: %s holds %v, want a free version on top", oi.Name, got)
+	}
+}
+
+func (f *freeVersionTest) label(fi FileInfo) string {
+	switch {
+	case fi.TierFreeVersion():
+		return "free"
+	case fi.Deleted:
+		return "DM"
+	case f.labels[fi.VersionID] != "":
+		return f.labels[fi.VersionID]
+	}
+	return "?"
+}
+
+// xlMeta reads the metadata of object in pool from its first drive, nil if
+// the pool has none.
+func (f *freeVersionTest) xlMeta(pool int, object string) []byte {
+	f.t.Helper()
+	disk := f.z.serverPools[pool].getHashedSet(object).getDisks()[0]
+	buf, err := disk.ReadAll(f.t.Context(), f.bucket, pathJoin(object, xlStorageFormatFile))
+	if errors.Is(err, errFileNotFound) {
+		return nil
+	}
+	if err != nil {
+		f.t.Fatal(err)
+	}
+	return buf
+}
+
+// stored returns what pool holds of object, newest first, free versions
+// included.
+func (f *freeVersionTest) stored(pool int, object string) []string {
+	f.t.Helper()
+	buf := f.xlMeta(pool, object)
+	if buf == nil {
+		return nil
+	}
+	fivs, err := getFileInfoVersions(buf, f.bucket, object, true)
+	if err != nil {
+		f.t.Fatal(err)
+	}
+	var got []string
+	for _, fi := range fivs.Versions {
+		got = append(got, f.label(fi))
+	}
+	return got
+}
+
+// versions labels the versions of a listing as "<object> <label>[ latest]".
+func (f *freeVersionTest) versions(objects []ObjectInfo) []string {
+	var got []string
+	for _, oi := range objects {
+		label := oi.Name + " " + f.label(FileInfo{VersionID: oi.VersionID, Deleted: oi.DeleteMarker})
+		if oi.IsLatest {
+			label += " latest"
+		}
+		got = append(got, label)
+	}
+	return got
+}
+
+// listVersions returns the versions ListObjectVersions lists in one page.
+func (f *freeVersionTest) listVersions() []string {
+	f.t.Helper()
+	lv, err := f.z.ListObjectVersions(f.t.Context(), f.bucket, "", "", "", "", 100)
+	if err != nil {
+		f.t.Fatal(err)
+	}
+	return f.versions(lv.Objects)
+}
+
+// expiryQueue replaces the expiry workers with a queue for the test.
+func (f *freeVersionTest) expiryQueue() chan expiryOp {
+	queue := make(chan expiryOp, 100)
+	es := &expiryState{ctx: f.t.Context(), objAPI: f.z}
+	workers := []chan expiryOp{queue}
+	es.workers.Store(&workers)
+	prev := globalExpiryState
+	globalExpiryState = es
+	f.t.Cleanup(func() { globalExpiryState = prev })
+	return queue
+}
+
+// expired returns the labels of the versions queued for expiry.
+func (f *freeVersionTest) expired(queue chan expiryOp) []string {
+	var got []string
+	for len(queue) > 0 {
+		switch task := (<-queue).(type) {
+		case expiryTask:
+			got = append(got, f.labels[task.objInfo.VersionID])
+		case noncurrentVersionsTask:
+			for _, v := range task.versions {
+				got = append(got, f.labels[v.VersionID])
+			}
+		}
+	}
+	return got
+}
+
+// The scanner reads the versions of obj-01 without the free version. v2 is the
+// latest version, and a noncurrent expiry must not count it as noncurrent
+// since the deleted v3 was written. Otherwise the scanner deletes it.
+func TestFreeVersionScannerLatestVersion(t *testing.T) {
+	f := newFreeVersionTest(t, freeVersionNoncurrent)
+	fivs, err := getFileInfoVersions(f.xlMeta(0, "obj-01"), f.bucket, "obj-01", false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var got []string
+	for _, fi := range fivs.Versions {
+		got = append(got, f.label(fi))
+	}
+	if want := []string{"v2", "v1"}; !slices.Equal(got, want) || len(fivs.FreeVersions) != 1 {
+		t.Fatalf("scanner versions %v with %d free versions, want %v with 1", got, len(fivs.FreeVersions), want)
+	}
+	v2, v1 := fivs.Versions[0], fivs.Versions[1]
+	if !v2.IsLatest || !v2.SuccessorModTime.IsZero() {
+		t.Errorf("v2: latest %v, successor mod time %v; want latest without a successor", v2.IsLatest, v2.SuccessorModTime)
+	}
+	if v1.IsLatest || !v1.SuccessorModTime.Equal(v2.ModTime) {
+		t.Errorf("v1: latest %v, successor mod time %v; want noncurrent since v2 at %v", v1.IsLatest, v1.SuccessorModTime, v2.ModTime)
+	}
+	if !fivs.LatestModTime.Equal(v2.ModTime) {
+		t.Errorf("latest mod time %v, want that of v2 %v", fivs.LatestModTime, v2.ModTime)
+	}
+
+	// Apply the rules to these versions as the scanner does.
+	lc, err := globalLifecycleSys.Get(f.bucket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	objInfos := make([]ObjectInfo, len(fivs.Versions))
+	for i, fi := range fivs.Versions {
+		objInfos[i] = fi.ToObjectInfo(f.bucket, "obj-01", true)
+	}
+	queue := f.expiryQueue()
+	item := scannerItem{Path: "obj-01", bucket: f.bucket, objectName: "obj-01", lifeCycle: lc}
+	item.applyActions(t.Context(), f.z, objInfos, objectlock.Retention{}, &sizeSummary{}, nil)
+	if expired, want := f.expired(queue), []string{"v1"}; !slices.Equal(expired, want) {
+		t.Errorf("scanner queued %v for expiry, want %v", expired, want)
+	}
+}
+
+// ListObjectsV2 lists obj-01: its latest version is v2, not a delete marker.
+// obj-02 has no versions left.
+func TestFreeVersionListObjects(t *testing.T) {
+	f := newFreeVersionTest(t, "")
+	lo, err := f.z.ListObjectsV2(t.Context(), f.bucket, "", "", "", 100, false, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var got []string
+	for _, oi := range lo.Objects {
+		got = append(got, oi.Name)
+	}
+	if want := []string{"obj-01"}; !slices.Equal(got, want) {
+		t.Errorf("ListObjectsV2: %v, want %v", got, want)
+	}
+}
+
+// The listing checks of the latest version skip the free version: obj-01 is
+// shown, obj-02 has nothing to show. Both the encoded metadata and the
+// decoded form, which the listing caches, are checked.
+func TestFreeVersionLatestDeleteMarker(t *testing.T) {
+	f := newFreeVersionTest(t, "")
+	for _, tc := range []struct {
+		object string
+		hidden bool
+	}{{"obj-01", false}, {"obj-02", true}} {
+		entry := metaCacheEntry{name: tc.object, metadata: f.xlMeta(0, tc.object)}
+		meta, _, err := isIndexedMetaV2(entry.metadata)
+		if err != nil || meta == nil {
+			t.Fatalf("%s: metadata is not indexed: %v", tc.object, err)
+		}
+		if got := meta.IsLatestDeleteMarker(); got != tc.hidden {
+			t.Errorf("%s: IsLatestDeleteMarker %v, want %v", tc.object, got, tc.hidden)
+		}
+		if got := meta.AllHidden(true); got != tc.hidden {
+			t.Errorf("%s: AllHidden(true) %v, want %v", tc.object, got, tc.hidden)
+		}
+		if got := meta.AllHidden(false); got != tc.hidden {
+			t.Errorf("%s: AllHidden(false) %v, want %v", tc.object, got, tc.hidden)
+		}
+		if got := entry.isLatestDeletemarker(); got != tc.hidden {
+			t.Errorf("%s: isLatestDeletemarker %v, want %v", tc.object, got, tc.hidden)
+		}
+		if _, err := entry.xlmeta(); err != nil {
+			t.Fatal(err)
+		}
+		if got := entry.isLatestDeletemarker(); got != tc.hidden {
+			t.Errorf("%s: isLatestDeletemarker of the decoded metadata %v, want %v", tc.object, got, tc.hidden)
+		}
+	}
+}
+
+// ListObjectVersions lists v2 as the latest version of obj-01 and no delete
+// markers. Keys with nothing but a free version have no versions to list and
+// do not take up a page. Walk returns the same versions.
+func TestFreeVersionListObjectVersions(t *testing.T) {
+	f := newFreeVersionTest(t, "")
+	f.deleteVersion(f.tiered("obj-00", "v1", UTCNow().AddDate(0, 0, -3)))
+	want := []string{"obj-01 v2 latest", "obj-01 v1"}
+	if got := f.listVersions(); !slices.Equal(got, want) {
+		t.Errorf("ListObjectVersions: %v, want %v", got, want)
+	}
+
+	results := make(chan itemOrErr[ObjectInfo], 100)
+	if err := f.z.Walk(t.Context(), f.bucket, "", results, WalkOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	var walked []ObjectInfo
+	for r := range results {
+		if r.Err != nil {
+			t.Fatal(r.Err)
+		}
+		walked = append(walked, r.Item)
+	}
+	if got, want := f.versions(walked), []string{"obj-01 v1", "obj-01 v2 latest"}; !slices.Equal(got, want) {
+		t.Errorf("Walk: %v, want %v", got, want)
+	}
+
+	// Page by page, one version a page.
+	for _, delimiter := range []string{"", "/"} {
+		var got []string
+		marker, versionMarker := "", ""
+		for range 10 {
+			lv, err := f.z.ListObjectVersions(t.Context(), f.bucket, "", marker, versionMarker, delimiter, 1)
+			if err != nil {
+				t.Fatal(err)
+			}
+			got = append(got, f.versions(lv.Objects)...)
+			if !lv.IsTruncated {
+				break
+			}
+			marker, versionMarker = lv.NextMarker, lv.NextVersionIDMarker
+		}
+		if !slices.Equal(got, want) {
+			t.Errorf("ListObjectVersions page by page, delimiter %q: %v, want %v", delimiter, got, want)
+		}
+	}
+}
+
+// Healing still visits the free versions, so a free version lost on some
+// drives is healed and the scanner still finds it.
+func TestFreeVersionHealObjects(t *testing.T) {
+	f := newFreeVersionTest(t, "")
+	var got []string
+	err := f.z.HealObjects(t.Context(), f.bucket, "", madmin.HealOpts{Recursive: true}, func(_, object, versionID string, _ madmin.HealScanMode) error {
+		label := f.labels[versionID]
+		if label == "" {
+			label = "free"
+		}
+		got = append(got, object+" "+label)
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	slices.Sort(got)
+	if want := []string{"obj-01 free", "obj-01 v1", "obj-01 v2", "obj-02 free"}; !slices.Equal(got, want) {
+		t.Errorf("healed versions %v, want %v", got, want)
+	}
+}
+
+// Decommission and rebalance do not move a free version: in another pool it
+// would be a delete marker that hides v2. Decommission removes the remote
+// object and the free version, as the scanner would before the pool is
+// removed. Rebalance leaves both to the scanner of the source pool.
+func TestFreeVersionDataMovement(t *testing.T) {
+	for _, kind := range []string{"decommission", "rebalance"} {
+		for _, rules := range []string{"", freeVersionNoncurrent} {
+			name := kind
+			if rules != "" {
+				name += "/noncurrent-expiry"
+			}
+			t.Run(name, func(t *testing.T) {
+				f := newFreeVersionTest(t, rules)
+				queue := f.expiryQueue()
+				if failed := migrationTestWalk(t, f.z, f.bucket, 0, kind); failed != 0 {
+					t.Fatalf("%d versions failed to move", failed)
+				}
+				if expired := f.expired(queue); slices.Contains(expired, "v2") {
+					t.Errorf("%s queued the latest version v2 for expiry: %v", kind, expired)
+				}
+
+				// v1 has been noncurrent for days; whether a move skips an
+				// expired version is out of scope here.
+				if got := f.stored(1, "obj-01"); len(got) == 0 || got[0] != "v2" || slices.Contains(got, "DM") {
+					t.Errorf("pool 1 holds %v of obj-01, want v2 on top and no delete marker", got)
+				}
+				if got := f.stored(1, "obj-02"); len(got) != 0 {
+					t.Errorf("pool 1 holds %v of obj-02, want nothing", got)
+				}
+				if body := migrationTestBody(t, f.z.GetObjectNInfo, f.bucket, "obj-01", ""); body != "obj-01 v2" {
+					t.Errorf("obj-01 reads %q, want %q", body, "obj-01 v2")
+				}
+				if got := f.listVersions(); len(got) == 0 || got[0] != "obj-01 v2 latest" || slices.Contains(got, "obj-02 DM latest") {
+					t.Errorf("ListObjectVersions after %s: %v", kind, got)
+				}
+
+				if rules != "" {
+					// The expired v1 is not moved, so its source is kept.
+					return
+				}
+				var wantSource []string // decommission
+				removed := true
+				if kind == "rebalance" {
+					wantSource, removed = []string{"free"}, false
+				}
+				for _, object := range []string{"obj-01", "obj-02"} {
+					if got := f.stored(0, object); !slices.Equal(got, wantSource) {
+						t.Errorf("pool 0 holds %v of %s, want %v", got, object, wantSource)
+					}
+				}
+				for _, remote := range []string{f.remote("obj-01", "v3"), f.remote("obj-02", "v1")} {
+					if slices.Contains(f.tier.removed, remote) != removed {
+						t.Errorf("remote object %s of a deleted version removed: %v, want %v", remote, !removed, removed)
+					}
+				}
+			})
+		}
+	}
+}
+
+// decommission decommissions pool 0 as the admin API does, including the check
+// for versions left behind, and returns its state.
+func (f *freeVersionTest) decommission() PoolDecommissionInfo {
+	f.t.Helper()
+	ctx := f.t.Context()
+	if err := f.z.StartDecommission(ctx, 0); err != nil {
+		f.t.Fatal(err)
+	}
+	f.z.doDecommissionInRoutine(ctx, 0)
+	f.z.poolMetaMutex.RLock()
+	defer f.z.poolMetaMutex.RUnlock()
+	return *f.z.poolMeta.Pools[0].Decommission
+}
+
+// While the tier is unavailable, the decommission moves the versions but keeps
+// the free versions, and the check for versions left behind fails it. A retry
+// with the tier back completes it.
+func TestFreeVersionDecommissionTierUnavailable(t *testing.T) {
+	f := newFreeVersionTest(t, "")
+	f.tier.removeErr = errors.New("tier unavailable")
+	if info := f.decommission(); info.Complete || !info.Failed {
+		t.Fatalf("decommission with the tier unavailable: complete %v, failed %v; want failed", info.Complete, info.Failed)
+	}
+	if got, want := f.stored(1, "obj-01"), []string{"v2", "v1"}; !slices.Equal(got, want) {
+		t.Errorf("pool 1 holds %v of obj-01, want %v", got, want)
+	}
+	for _, object := range []string{"obj-01", "obj-02"} {
+		if got, want := f.stored(0, object), []string{"free"}; !slices.Equal(got, want) {
+			t.Errorf("pool 0 holds %v of %s, want %v", got, object, want)
+		}
+	}
+	if body := migrationTestBody(t, f.z.GetObjectNInfo, f.bucket, "obj-01", ""); body != "obj-01 v2" {
+		t.Errorf("obj-01 reads %q, want %q", body, "obj-01 v2")
+	}
+	if len(f.tier.removed) != 0 {
+		t.Errorf("remote objects removed: %v", f.tier.removed)
+	}
+
+	// The failed decommission moved the bucket metadata, so pool 0 has no
+	// .minio.sys/buckets left. On a server the scanner recreates it when it
+	// saves its usage cache there; it does not run here.
+	for _, set := range f.z.serverPools[0].sets {
+		for _, disk := range set.getDisks() {
+			if err := disk.MakeVol(t.Context(), pathJoin(minioMetaBucket, bucketMetaPrefix)); err != nil && !errors.Is(err, errVolumeExists) {
+				t.Fatal(err)
+			}
+		}
+	}
+	f.tier.removeErr = nil
+	if info := f.decommission(); !info.Complete || info.Failed {
+		t.Fatalf("retry: complete %v, failed %v; want complete", info.Complete, info.Failed)
+	}
+	if got, want := f.stored(1, "obj-01"), []string{"v2", "v1"}; !slices.Equal(got, want) {
+		t.Errorf("pool 1 holds %v of obj-01 after the retry, want %v", got, want)
+	}
+	for _, object := range []string{"obj-01", "obj-02"} {
+		if got := f.stored(0, object); len(got) != 0 {
+			t.Errorf("pool 0 holds %v of %s after the retry", got, object)
+		}
+	}
+	if want := []string{f.remote("obj-01", "v3"), f.remote("obj-02", "v1")}; !slices.Equal(slices.Sorted(slices.Values(f.tier.removed)), want) {
+		t.Errorf("remote objects removed: %v, want %v", f.tier.removed, want)
+	}
+}

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/erasure-server-pool.go

@@ -2868,7 +2868,7 @@ func (z *erasureServerPools) HealObjects(ctx context.Context, bucket, prefix str
 				return nil
 			}
 		}
-		fivs, err := entry.fileInfoVersions(bucket)
+		fivs, err := entry.allFileInfoVersions(bucket)
 		if err != nil {
 			return healObjectFn(bucket, entry.name, "", scanMode)
 		}

cmd/global-heal.go

@@ -423,7 +423,7 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
 			encodedEntryName := encodeDirObject(entry.name)
 
 			var result healEntryResult
-			fivs, err := entry.fileInfoVersions(bucket)
+			fivs, err := entry.allFileInfoVersions(bucket)
 			if err != nil {
 				res, err := healInPool(ctx, bucket, encodedEntryName, "",
 					madmin.HealOpts{

cmd/metacache-entries.go

@@ -177,14 +177,12 @@ func (e metaCacheEntry) isInDir(dir, separator string) bool {
 }
 
 // isLatestDeletemarker returns whether the latest version is a delete marker.
+// Free versions are not versions of the object and are skipped.
 // If metadata is NOT versioned false will always be returned.
 // If v2 and UNABLE to load metadata true will be returned.
 func (e *metaCacheEntry) isLatestDeletemarker() bool {
 	if e.cached != nil {
-		if len(e.cached.versions) == 0 {
-			return true
-		}
-		return e.cached.versions[0].header.Type == DeleteType
+		return e.cached.isLatestDeleteMarker()
 	}
 	if !isXL2V1Format(e.metadata) {
 		return false
@@ -196,10 +194,10 @@ func (e *metaCacheEntry) isLatestDeletemarker() bool {
 	}
 	// Fall back...
 	xlMeta, err := e.xlmeta()
-	if err != nil || len(xlMeta.versions) == 0 {
+	if err != nil {
 		return true
 	}
-	return xlMeta.versions[0].header.Type == DeleteType
+	return xlMeta.isLatestDeleteMarker()
 }
 
 // isAllFreeVersions returns if all objects are free versions.
@@ -287,9 +285,21 @@ func (e *metaCacheEntry) xlmeta() (*xlMetaV2, error) {
 	return e.cached, nil
 }
 
-// fileInfoVersions returns the metadata as FileInfoVersions.
+// fileInfoVersions returns the metadata as FileInfoVersions. The free
+// versions of deleted tiered versions are not versions of the object; they
+// are in FreeVersions.
 // If entry is a directory it is returned as that.
 func (e *metaCacheEntry) fileInfoVersions(bucket string) (FileInfoVersions, error) {
+	return e.getFileInfoVersions(bucket, false)
+}
+
+// allFileInfoVersions returns the metadata as FileInfoVersions with the free
+// versions among the versions, for healing.
+func (e *metaCacheEntry) allFileInfoVersions(bucket string) (FileInfoVersions, error) {
+	return e.getFileInfoVersions(bucket, true)
+}
+
+func (e *metaCacheEntry) getFileInfoVersions(bucket string, inclFreeVersions bool) (FileInfoVersions, error) {
 	if e.isDir() {
 		return FileInfoVersions{
 			Volume: bucket,
@@ -304,7 +314,7 @@ func (e *metaCacheEntry) fileInfoVersions(bucket string) (FileInfoVersions, erro
 		}, nil
 	}
 	// Too small gains to reuse cache here.
-	return getFileInfoVersions(e.metadata, bucket, e.name, true)
+	return getFileInfoVersions(e.metadata, bucket, e.name, inclFreeVersions)
 }
 
 // metaCacheEntries is a slice of metacache entries.

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/metacache-set.go

@@ -193,6 +193,11 @@ func (o *listPathOptions) shouldSkip(ctx context.Context, entry metaCacheEntry)
 	if !o.InclDeleted && entry.isObject() && entry.isLatestDeletemarker() && !entry.isObjectDir() {
 		return true
 	}
+	// Free versions are not versions of the object. Without other versions
+	// there is nothing to list, and the entry must not count toward the limit.
+	if o.InclDeleted && entry.isObject() && entry.isAllFreeVersions() {
+		return true
+	}
 	if o.Lifecycle != nil || o.Replication.Config != nil {
 		return triggerExpiryAndRepl(ctx, *o, entry)
 	}

cmd/xl-storage-format-utils.go

@@ -19,6 +19,7 @@ package cmd
 
 import (
 	"errors"
+	"time"
 
 	"github.com/zeebo/xxh3"
 )
@@ -28,8 +29,12 @@ import (
 //   - fivs.FreeVersions has all the free versions
 //
 // if inclFreeVersions is true all the versions are in fivs.Versions, free and non-free versions alike.
+// Otherwise IsLatest, SuccessorModTime and LatestModTime are those of the
+// non-free versions.
 //
-// Note: Only the scanner requires fivs.Versions to have exclusively non-free versions. This is used while enforcing NewerNoncurrentVersions lifecycle element.
+// Only healing needs the free versions among the versions. Free versions are
+// not versions of the object, and lifecycle evaluation, listings and data
+// movement must not count them.
 func getFileInfoVersions(xlMetaBuf []byte, volume, path string, inclFreeVersions bool) (FileInfoVersions, error) {
 	fivs, err := getAllFileInfoVersions(xlMetaBuf, volume, path, true)
 	if err != nil {
@@ -56,6 +61,19 @@ func getFileInfoVersions(xlMetaBuf []byte, volume, path string, inclFreeVersions
 	}
 	if !inclFreeVersions {
 		fivs.Versions = fivs.Versions[:n]
+		// A free version is not a version of the object. The latest
+		// version, and the successor of each other version, is among
+		// the rest.
+		for i := range fivs.Versions {
+			fivs.Versions[i].IsLatest = i == 0
+			fivs.Versions[i].SuccessorModTime = time.Time{}
+			if i > 0 {
+				fivs.Versions[i].SuccessorModTime = fivs.Versions[i-1].ModTime
+			}
+		}
+		if n > 0 {
+			fivs.LatestModTime = fivs.Versions[0].ModTime
+		}
 	}
 	// Update numversions
 	for i := range fivs.Versions {

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)
+			}
+		})
+	}
+}

cmd/xl-storage-format-v2.go

@@ -1138,6 +1138,17 @@ func (x *xlMetaV2) latestModtime() time.Time {
 	return time.Unix(0, x.versions[0].header.ModTime)
 }
 
+// isLatestDeleteMarker returns true if the latest version that is not a free
+// version is a delete marker, or if there is none.
+func (x *xlMetaV2) isLatestDeleteMarker() bool {
+	for _, ver := range x.versions {
+		if !ver.header.FreeVersion() {
+			return ver.header.Type == DeleteType
+		}
+	}
+	return true
+}
+
 func (x *xlMetaV2) addVersion(ver xlMetaV2Version) error {
 	modTime := ver.getModTime().UnixNano()
 	if !ver.Valid() {
@@ -2027,9 +2038,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
 				}
@@ -2289,6 +2302,7 @@ func (x xlMetaBuf) ListVersions(volume, path string, allParts bool) ([]FileInfo,
 }
 
 // IsLatestDeleteMarker returns true if latest version is a deletemarker or there are no versions.
+// Free versions are not versions of the object and are skipped.
 // If any error occurs false is returned.
 func (x xlMetaBuf) IsLatestDeleteMarker() bool {
 	vers, headerV, _, buf, err := decodeXLHeaders(x)
@@ -2298,13 +2312,17 @@ func (x xlMetaBuf) IsLatestDeleteMarker() bool {
 	if vers == 0 {
 		return true
 	}
-	isDeleteMarker := false
+	isDeleteMarker := true // only free versions
 
 	_ = decodeVersions(buf, vers, func(idx int, hdr, _ []byte) error {
 		var xl xlMetaV2VersionHeader
 		if _, err := xl.unmarshalV(headerV, hdr); err != nil {
+			isDeleteMarker = false
 			return errDoneForNow
 		}
+		if xl.FreeVersion() {
+			return nil
+		}
 		isDeleteMarker = xl.Type == DeleteType
 		return errDoneForNow
 	})
@@ -2312,7 +2330,7 @@ func (x xlMetaBuf) IsLatestDeleteMarker() bool {
 }
 
 // AllHidden returns true are no versions that would show up in a listing (ie all free markers)
-// Optionally also return early if top is a delete marker.
+// Optionally also return early if the top version that is not free is a delete marker.
 func (x xlMetaBuf) AllHidden(topDeleteMarker bool) bool {
 	vers, headerV, _, buf, err := decodeXLHeaders(x)
 	if err != nil {
@@ -2328,16 +2346,12 @@ func (x xlMetaBuf) AllHidden(topDeleteMarker bool) bool {
 		if _, err := xl.unmarshalV(headerV, hdr); err != nil {
 			return errDoneForNow
 		}
-		if topDeleteMarker && idx == 0 && xl.Type == DeleteType {
-			hidden = true
-			return errDoneForNow
+		if xl.FreeVersion() {
+			// Check next version
+			return nil
 		}
-		if !xl.FreeVersion() {
-			hidden = false
-			return errDoneForNow
-		}
-		// Check next version
-		return nil
+		hidden = topDeleteMarker && xl.Type == DeleteType
+		return errDoneForNow
 	})
 	return hidden
 }