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

CHANGELOG.md

@@ -2,8 +2,11 @@
 
 ## RELEASE.2026-10-26T00-00-00Z
 
-Package version: `20261026000000.0.0`. Publication status and artifacts are
+Package version: `20261026000000.0.0`; published from source `c3ad836aed0d`.
-listed in [GitHub Releases](https://github.com/pgsty/silo/releases).
+Artifacts are listed in the [GitHub release](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-10-26T00-00-00Z).
+The coordinated [English release notes](https://silo.pgsty.com/blog/release/silo-20261026/)
+and [Chinese release notes](https://silo.pgsty.com/zh/blog/release/silo-20261026/)
+cover all components, upgrade guidance, and validation scope.
 
 - Fix LDAP login for long normalized DNs without hiding unreadable IAM
   revocation records. Only length errors for valid keys with an ASCII path

README.md

@@ -24,7 +24,7 @@
 <p align="center">
   <a href="https://silo.pgsty.com/"><img alt="Website" src="https://img.shields.io/badge/Website-silo.pgsty.com-1d588c"></a>
   <a href="https://github.com/pgsty/silo/releases"><img alt="GitHub Release" src="https://img.shields.io/github/v/release/pgsty/silo?include_prereleases&label=release&logo=github"></a>
-  <a href="https://hub.docker.com/r/pgsty/silo"><img alt="Docker Pulls" src="https://img.shields.io/docker/pulls/pgsty/minio?logo=docker"></a>
+  <a href="https://hub.docker.com/r/pgsty/silo"><img alt="Docker Pulls" src="https://img.shields.io/docker/pulls/pgsty/silo?logo=docker"></a>
   <a href="go.mod"><img alt="Go Version" src="https://img.shields.io/github/go-mod/go-version/pgsty/silo?logo=go"></a>
   <a href="LICENSE"><img alt="License" src="https://img.shields.io/badge/license-AGPLv3-blue"></a>
 </p>
@@ -35,20 +35,13 @@
 > [!NOTE]
 > Renamed from `pgsty/minio` to `pgsty/silo`, default branch `master` → `main`, on 2026-08-06. Artifacts under the original MinIO identity stay published on the archived [`minio`](https://github.com/pgsty/silo/tree/minio) branch and in releases up to [`RELEASE.2026-08-04T00-00-00Z`](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-08-04T00-00-00Z).
 
-## SILO 20261018000000
+## SILO 20261026000000
 
-[SILO 20261018000000](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-10-18T00-00-00Z) is available as platform archives, Linux packages, and standard or distroless Docker images for amd64 and arm64. The release tag is `RELEASE.2026-10-18T00-00-00Z`; the package version is `20261018000000.0.0`.
+[SILO 20261026000000](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-10-26T00-00-00Z) is available as platform archives, Linux packages, and standard or distroless Docker images for amd64 and arm64. The release tag is `RELEASE.2026-10-26T00-00-00Z`; the package version is `20261026000000.0.0`, built from source `c3ad836aed0d`.
 
-This release strengthens Object Lock and IAM authorization, makes scoped STS revocation durable across processes and sites, protects concurrent updates during pool migration, and improves LIST recovery, multipart cleanup, and S3 error handling. It includes Console v2.4.2 and silo-pkg v3.14.3, and is built with Go 1.27.2. Use mcli 20261018 for administration.
+This release fixes LDAP long-DN login, IAM policy refresh, multi-pool deletion safety, tiered-object expiry, multipart replication checksums, and remote rollback-option propagation. It embeds Console v2.5.1 (`fff1618d9abb`) and uses silo-pkg v3.15.2, Go 1.27.3, and upstream minio-go `v7.4.2-0.20261025165926-320af050c445`. Use mcli 20261026 for administration; the container images bundle it as `mcli`.
 
-Read the [release notes](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-10-18T00-00-00Z) before upgrading, especially the removal of strict multipart listing, coordinated server upgrades, and measured performance costs. See [CONTRIBUTORS.md](CONTRIBUTORS.md) for the people behind the code, reports, and documentation.
+The Server IAM cache repair enables Console's **Reset to default** action. Existing storage timeouts and cancellation behavior remain: a timeout does not prove that a disk operation stopped or a write did not commit. Read the [coordinated release notes](https://silo.pgsty.com/blog/release/silo-20261026/) and [remaining storage limits](CHANGELOG.md#release2026-10-26t00-00-00z) before upgrading. See [CONTRIBUTORS.md](CONTRIBUTORS.md) for the people behind the code, reports, and documentation.
-
-The 20261026 maintenance release is in preparation with Go 1.27.3,
-silo-pkg v3.15.2, mcli 20261026, Console v2.5.1 (`fff1618d9abb`),
-and upstream minio-go `320af050c445`. It includes the Server IAM cache repair
-needed by Console's **Reset to default** action and remote rollback-option propagation, preserving existing storage
-timeouts and cancellation behavior. Read the [changes and remaining storage limits](CHANGELOG.md#release2026-10-26t00-00-00z).
-The published artifacts above retain their original component versions.
 
 ## Overview
 
@@ -69,7 +62,7 @@ docker run -d --name silo -p 9000:9000 -p 9001:9001 \
   -e MINIO_ROOT_USER=minioadmin \
   -e MINIO_ROOT_PASSWORD=change-me-long-password \
   -v "$PWD/data:/data" \
-  docker.io/pgsty/silo:RELEASE.2026-10-18T00-00-00Z server /data --console-address ":9001"
+  docker.io/pgsty/silo:RELEASE.2026-10-26T00-00-00Z server /data --console-address ":9001"
 ```
 
 <p align="center">

README_ZH.md

@@ -24,7 +24,7 @@
 <p align="center">
   <a href="https://silo.pgsty.com/zh/"><img alt="官网" src="https://img.shields.io/badge/%E5%AE%98%E7%BD%91-silo.pgsty.com%2Fzh-1d588c"></a>
   <a href="https://github.com/pgsty/silo/releases"><img alt="GitHub Release" src="https://img.shields.io/github/v/release/pgsty/silo?include_prereleases&label=release&logo=github"></a>
-  <a href="https://hub.docker.com/r/pgsty/silo"><img alt="Docker Pulls" src="https://img.shields.io/docker/pulls/pgsty/minio?logo=docker"></a>
+  <a href="https://hub.docker.com/r/pgsty/silo"><img alt="Docker Pulls" src="https://img.shields.io/docker/pulls/pgsty/silo?logo=docker"></a>
   <a href="go.mod"><img alt="Go Version" src="https://img.shields.io/github/go-mod/go-version/pgsty/silo?logo=go"></a>
   <a href="LICENSE"><img alt="License" src="https://img.shields.io/badge/license-AGPLv3-blue"></a>
 </p>
@@ -35,20 +35,13 @@
 > [!NOTE]
 > 2026-08-06，本仓库由 `pgsty/minio` 更名为 `pgsty/silo`，默认分支由 `master` 更名为 `main`。以原 MinIO 形态维持的归档构件仍位于归档的 [`minio`](https://github.com/pgsty/silo/tree/minio) 分支，以及截止 [`RELEASE.2026-08-04T00-00-00Z`](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-08-04T00-00-00Z) 的历次发布中。
 
-## SILO 20261018000000
+## SILO 20261026000000
 
-[SILO 20261018000000](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-10-18T00-00-00Z) 已正式发布，提供各平台归档包、Linux 软件包，以及 amd64／arm64 标准版和 distroless Docker 镜像。版本标签为 `RELEASE.2026-10-18T00-00-00Z`，包版本为 `20261018000000.0.0`。
+[SILO 20261026000000](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-10-26T00-00-00Z) 已正式发布，提供各平台归档包、Linux 软件包，以及 amd64／arm64 标准版和 distroless Docker 镜像。版本标签为 `RELEASE.2026-10-26T00-00-00Z`，包版本为 `20261026000000.0.0`，构建源码为 `c3ad836aed0d`。
 
-本版加强 Object Lock 与 IAM 鉴权，实现跨进程和站点的持久 STS 撤销，保护多池迁移中的并发更新，并改进 LIST 恢复、分片清理与 S3 错误处理。内置 Console v2.4.2、silo-pkg v3.14.3，使用 Go 1.27.2 构建，配套管理客户端为 mcli 20261018。
+本版修复 LDAP 长 DN 登录、IAM 策略刷新、多池删除安全、分层对象过期、分片复制校验和及远端回滚参数透传。内置 Console v2.5.1（`fff1618d9abb`），使用 silo-pkg v3.15.2、Go 1.27.3 与上游 minio-go `v7.4.2-0.20261025165926-320af050c445`。配套管理客户端为 mcli 20261026，容器镜像内置该版本的 `mcli`。
 
-升级前请阅读[发行说明](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-10-18T00-00-00Z)，特别是 strict 分片列表的移除、服务端协调升级要求与实测性能成本。代码、问题报告和文档贡献者见 [CONTRIBUTORS.md](CONTRIBUTORS.md)。
+Server IAM 缓存修复使 Console **Reset to default（恢复默认）** 操作即时生效。存储超时与取消行为保持既有语义：超时不代表底层磁盘操作已停止，也不代表写入未提交。升级前请阅读[联合发布说明](https://silo.pgsty.com/zh/blog/release/silo-20261026/)与[仍存的存储限制](CHANGELOG.md#release2026-10-26t00-00-00z)。代码、问题报告和文档贡献者见 [CONTRIBUTORS.md](CONTRIBUTORS.md)。
-
-20261026 维护版本正在准备中：使用 Go 1.27.3、silo-pkg v3.15.2、
-mcli 20261026、Console v2.5.1（`fff1618d9abb`）与上游 minio-go
-`320af050c445`。其中包含 Console **Reset to default（恢复默认）** 所需的
-Server IAM 缓存修复，以及远端回滚参数透传修复；保持既有存储超时与取消行为。
-请阅读[变更与仍存的存储限制](CHANGELOG.md#release2026-10-26t00-00-00z)；
-上述已发布制品仍保留原有组件版本。
 
 ## 概述
 
@@ -69,7 +62,7 @@ docker run -d --name silo -p 9000:9000 -p 9001:9001 \
   -e MINIO_ROOT_USER=minioadmin \
   -e MINIO_ROOT_PASSWORD=change-me-long-password \
   -v "$PWD/data:/data" \
-  docker.io/pgsty/silo:RELEASE.2026-10-18T00-00-00Z server /data --console-address ":9001"
+  docker.io/pgsty/silo:RELEASE.2026-10-26T00-00-00Z server /data --console-address ":9001"
 ```
 
 <p align="center">

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 e.cached.isLatestDeleteMarker()
-			return true
-		}
-		return e.cached.versions[0].header.Type == DeleteType
 	}
 	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) {
+			// 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

@@ -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
+					// 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
 				}
@@ -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 {
+		if xl.FreeVersion() {
-			hidden = true
+			// Check next version
-			return errDoneForNow
+			return nil
 		}
-		if !xl.FreeVersion() {
+		hidden = topDeleteMarker && xl.Type == DeleteType
-			hidden = false
+		return errDoneForNow
-			return errDoneForNow
-		}
-		// Check next version
-		return nil
 	})
 	return hidden
 }