Merge lifecycle migration fix (#266)

Merge the contributor's lifecycle migration fix after forward-porting it onto current main.

The contributor commits are retained in the merge commit. Maintainer follow-up work will address the remaining replication, tier cleanup, lifecycle configuration, and DeleteAll completion-fence cases.

cmd/erasure-server-pool-consistency.go

@@ -130,7 +130,12 @@ func (z *erasureServerPools) checkDataMovementCopies(ctx context.Context, bucket
 // removeMovedVersions retires only the source versions in the walker's
 // snapshot. A concurrent delete can create a new marker in the source pool;
 // deleting the whole prefix would remove that marker and expose moved data.
-func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, object string, source int, versions []FileInfo) error {
+// Versions in expired were not moved because ILM deletes them. They are
+// removed from the source without a moved copy, as the queued expiry would
+// remove them. Left behind, such a version can become the latest one in the
+// source pool once the moved versions are gone, and the expiry, which
+// re-evaluates the rule there, would no longer delete it.
+func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, object string, source int, versions, expired []FileInfo) error {
 	object = encodeDirObject(object)
 	lk := z.NewNSLock(bucket, object)
 	lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
@@ -168,7 +173,7 @@ func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, ob
 		}
 		// Successful movement is not enough if the destination changed again.
 		// Keep the source unless another copy still preserves its read result.
-		if len(retained) == 0 || retained[0].ObjInfo.ModTime.Before(src.ModTime) {
+		if !isExpiredVersion(expired, version) && (len(retained) == 0 || retained[0].ObjInfo.ModTime.Before(src.ModTime)) {
 			return PreConditionFailed{}
 		}
 		_, err = z.serverPools[source].DeleteObject(ctx, bucket, object, ObjectOptions{
@@ -182,6 +187,23 @@ func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, ob
 	return nil
 }
 
+// ilmDeletesVersionOnly reports whether the ILM action queued for a version
+// that data movement skipped deletes that version and nothing else, rather
+// than every version of the object or only the restored copy of a
+// transitioned version. Such a version is passed to removeMovedVersions.
+func ilmDeletesVersionOnly(action lifecycle.Action) bool {
+	return action == lifecycle.DeleteAction || action == lifecycle.DeleteVersionAction
+}
+
+func isExpiredVersion(expired []FileInfo, version FileInfo) bool {
+	for _, fi := range expired {
+		if fi.VersionID == version.VersionID && fi.ModTime.Equal(version.ModTime) && fi.Deleted == version.Deleted {
+			return true
+		}
+	}
+	return false
+}
+
 // Healing also commits object metadata. Both queued healing and drive healing
 // must use the same namespace as pooled writes, rather than racing them under
 // a destination set's independent namespace.

cmd/erasure-server-pool-decom.go

@@ -796,9 +796,11 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 	}
 
 	for setIdx, set := range pool.sets {
-		filterLifecycle := func(bucket, object string, fi FileInfo) bool {
+		// filterLifecycle returns the action of an ILM deletion it queued
+		// for the version, which is then not moved, or NoneAction.
+		filterLifecycle := func(bucket, object string, fi FileInfo) lifecycle.Action {
 			if lc == nil {
-				return false
+				return lifecycle.NoneAction
 			}
 			versioned := vc != nil && vc.Versioned(object)
 			objInfo := fi.ToObjectInfo(bucket, object, versioned)
@@ -807,12 +809,18 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 			switch {
 			case evt.Action.DeleteRestored(): // if restored copy has expired, delete it synchronously
 				applyExpiryOnTransitionedObject(ctx, z, objInfo, evt, lcEventSrc_Decom)
-				return false
+				return lifecycle.NoneAction
+			case evt.Action == lifecycle.DeleteAction && versioned:
+				// Expiring the latest version of a versioned bucket only puts
+				// a delete marker on top of it. The version itself survives as
+				// a noncurrent version and has to be moved like any other. The
+				// scanner applies the expiry once it lives in another pool.
+				return lifecycle.NoneAction
 			case evt.Action.Delete():
 				globalExpiryState.enqueueByDays(objInfo, evt, lcEventSrc_Decom)
-				return true
+				return evt.Action
 			default:
-				return false
+				return lifecycle.NoneAction
 			}
 		}
 
@@ -833,11 +841,18 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 			versionsSorter(fivs.Versions).reverse()
 
 			var decommissioned int
+			var expired []FileInfo
 			for _, version := range fivs.Versions {
 				stopFn := globalDecommissionMetrics.log(decomMetricDecommissionObject, idx, bi.Name, version.Name, version.VersionID)
 				// Apply lifecycle rules on the objects that are expired.
-				if filterLifecycle(bi.Name, version.Name, version) {
+				if action := filterLifecycle(bi.Name, version.Name, version); action.Delete() {
 					decommissioned++
+					if ilmDeletesVersionOnly(action) {
+						// Its source copy goes with those of the moved
+						// versions. A version that takes the whole object
+						// with it stays until the expiry deletes them all.
+						expired = append(expired, version)
+					}
 					stopFn(version.Size, errors.New("ILM expired object/version will be skipped"))
 					continue
 				}
@@ -981,7 +996,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 			// if all versions were decommissioned, then we can delete the object versions.
 			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)
+				err := z.removeMovedVersions(ctx, bi.Name, entry.name, idx, fivs.Versions, expired)
 				if err == nil {
 					err = z.removeFreeVersions(ctx, bi.Name, entry.name, idx, fivs.FreeVersions)
 				}
@@ -1199,6 +1214,9 @@ func (z *erasureServerPools) checkAfterDecom(ctx context.Context, idx int) error
 				case evt.Action.DeleteRestored(): // if restored copy has expired,delete it synchronously
 					applyExpiryOnTransitionedObject(ctx, z, objInfo, evt, lcEventSrc_Decom)
 					return false
+				case evt.Action == lifecycle.DeleteAction && versioned:
+					// The version survives as noncurrent; it must have moved.
+					return false
 				case evt.Action.Delete():
 					globalExpiryState.enqueueByDays(objInfo, evt, lcEventSrc_Decom)
 					return true

cmd/erasure-server-pool-migration-delete_test.go

@@ -132,7 +132,7 @@ func TestPoolsMigrationCleanupRechecksSource(t *testing.T) {
 				_, err := z.serverPools[1].DeleteObject(t.Context(), bucket, oi.Name, ObjectOptions{VersionID: nullVersionID})
 				mustIAM(t, err)
 			}
-			err := z.removeMovedVersions(t.Context(), bucket, oi.Name, 0, snapshot)
+			err := z.removeMovedVersions(t.Context(), bucket, oi.Name, 0, snapshot, nil)
 			if change != "delete-uuid" && err == nil {
 				t.Fatal("cleanup accepted changed or uncertain state")
 			}

cmd/erasure-server-pool-migration-ilm_test.go

@@ -0,0 +1,456 @@
+// 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 (
+	"slices"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/minio/minio/internal/amztime"
+	"github.com/minio/minio/internal/bucket/lifecycle"
+	objectlock "github.com/minio/minio/internal/bucket/object/lock"
+	"github.com/minio/minio/internal/bucket/versioning"
+	xhttp "github.com/minio/minio/internal/http"
+)
+
+// Decommission and rebalance leave a version that ILM expires to the expiry
+// workers. Expiring the current version of a versioned object only puts a
+// delete marker on top of it: the version stays as a noncurrent version and
+// has to be moved like any other. A version that ILM does delete is not moved,
+// and the source copies of the versions that were moved must still be removed.
+
+// migrationILM is a bucket in pool 0 of two pools. The ILM expiry workers are
+// replaced by a queue that nothing drains, so the expiries a decommission or
+// rebalance queues run after it, as when the workers lag behind the walk.
+type migrationILM struct {
+	t       *testing.T
+	z       *erasureServerPools
+	bucket  string
+	queue   chan expiryOp
+	objects []string
+	labels  map[string]string // object and version ID -> label of a version
+	markers map[string]string // object and version ID -> label of a delete marker
+}
+
+func newMigrationILM(t *testing.T, status versioning.State, lock bool, rules string) *migrationILM {
+	t.Helper()
+	z, _ := migrationTestPools(t)
+	ctx := t.Context()
+	m := &migrationILM{
+		t: t, z: z, bucket: "migration-ilm", queue: make(chan expiryOp, 100),
+		labels: map[string]string{}, markers: map[string]string{},
+	}
+	err := z.MakeBucket(ctx, m.bucket, MakeBucketOptions{LockEnabled: lock, VersioningEnabled: status == versioning.Enabled})
+	if err != nil {
+		t.Fatal(err)
+	}
+	meta, err := globalBucketMetadataSys.Get(m.bucket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if status == versioning.Suspended {
+		meta.VersioningConfigXML = []byte("<VersioningConfiguration><Status>Suspended</Status></VersioningConfiguration>")
+	}
+	if rules != "" {
+		meta.LifecycleConfigXML = []byte("<LifecycleConfiguration>" + rules + "</LifecycleConfiguration>")
+	}
+	if err := meta.Save(ctx, z); err != nil {
+		t.Fatal(err)
+	}
+	globalBucketMetadataSys.Set(m.bucket, meta)
+
+	es := &expiryState{ctx: ctx, objAPI: z}
+	workers := []chan expiryOp{m.queue}
+	es.workers.Store(&workers)
+	prev := globalExpiryState
+	globalExpiryState = es
+	t.Cleanup(func() { globalExpiryState = prev })
+	return m
+}
+
+func (m *migrationILM) key(object, versionID string) string {
+	if versionID == "" {
+		versionID = nullVersionID
+	}
+	return object + " " + versionID
+}
+
+func (m *migrationILM) track(object string) {
+	if !slices.Contains(m.objects, object) {
+		m.objects = append(m.objects, object)
+	}
+}
+
+// put writes a version of object to pool 0.
+func (m *migrationILM) put(object, label string, opts ObjectOptions) ObjectInfo {
+	m.t.Helper()
+	oi := putConsistencyObject(m.t, m.z, m.bucket, object, 0, object+" "+label, opts)
+	m.labels[m.key(object, oi.VersionID)] = label
+	m.track(object)
+	return oi
+}
+
+// marker writes a delete marker of object to pool 0.
+func (m *migrationILM) marker(object, label string, mtime time.Time) {
+	m.t.Helper()
+	versionID := mustGetUUID()
+	_, err := m.z.serverPools[0].DeleteObject(m.t.Context(), m.bucket, object,
+		ObjectOptions{Versioned: true, VersionID: versionID, DeleteMarker: true, MTime: mtime})
+	if err != nil {
+		m.t.Fatal(err)
+	}
+	m.markers[m.key(object, versionID)] = label
+	m.track(object)
+}
+
+// tiered writes a version of object to pool 0 whose data is in a remote tier.
+func (m *migrationILM) tiered(object, label string) ObjectInfo {
+	m.t.Helper()
+	ctx := m.t.Context()
+	oi := m.put(object, label, ObjectOptions{Versioned: true})
+	infos, errs := readAllFileInfo(ctx, m.z.serverPools[0].getHashedSet(object).getDisks(), "", m.bucket, object, oi.VersionID, false, false)
+	if errs[0] != nil {
+		m.t.Fatal(errs[0])
+	}
+	fi := infos[0]
+	fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName = lifecycle.TransitionComplete, "WARM", "remote"
+	if err := m.z.serverPools[0].DecomTieredObject(ctx, m.bucket, object, fi, ObjectOptions{}); err != nil {
+		m.t.Fatal(err)
+	}
+	stored, err := m.z.serverPools[0].GetObjectInfo(ctx, m.bucket, object, ObjectOptions{VersionID: oi.VersionID})
+	if err != nil || stored.TransitionedObject.Status != lifecycle.TransitionComplete {
+		m.t.Fatalf("fixture is not a transitioned version: %q, %v", stored.TransitionedObject.Status, err)
+	}
+	return oi
+}
+
+func (m *migrationILM) label(object, versionID string, deleted bool) string {
+	key := m.key(object, versionID)
+	if l, ok := m.markers[key]; ok && deleted {
+		return l
+	}
+	if l, ok := m.labels[key]; ok && !deleted {
+		return l
+	}
+	if deleted {
+		return "DM"
+	}
+	return "?"
+}
+
+// move moves the contents of pool 0 out with a decommission, as the admin API
+// runs it, including the check for versions left behind, or with a rebalance
+// pass, with the walker's accounting and the removal of the source copies.
+// The rebalance goal is never reached, so the walk visits every object. The
+// test fails unless a decommission completes. ILM has not caught up yet.
+func (m *migrationILM) move(kind string) {
+	m.t.Helper()
+	ctx := m.t.Context()
+	if kind == "rebalance" {
+		migrationTestWalk(m.t, m.z, m.bucket, 0, kind)
+		m.z.rebalMu.Lock()
+		m.z.rebalMeta.PoolStats[0].Info.Status = rebalCompleted
+		m.z.rebalMu.Unlock()
+		return
+	}
+	if err := m.z.StartDecommission(ctx, 0); err != nil {
+		m.t.Fatal(err)
+	}
+	m.z.doDecommissionInRoutine(ctx, 0)
+	m.z.poolMetaMutex.RLock()
+	info := *m.z.poolMeta.Pools[0].Decommission
+	m.z.poolMetaMutex.RUnlock()
+	if !info.Complete || info.Failed || info.ItemsDecommissionFailed > 0 {
+		m.t.Fatalf("decommission did not complete: complete=%t failed=%t itemsFailed=%d; pool 0 holds %v",
+			info.Complete, info.Failed, info.ItemsDecommissionFailed, m.view(0))
+	}
+}
+
+// migrate moves the contents of pool 0 out and lets ILM catch up.
+func (m *migrationILM) migrate(kind string) {
+	m.t.Helper()
+	m.move(kind)
+	m.catchUp()
+}
+
+// catchUp lets ILM catch up with data movement out of pool 0: the queued
+// expiries run as the expiry workers run them, and the rules are applied to
+// the current version of every object as the scanner applies them in the pool
+// that holds it now. A transitioned version is not expired: its remote tier is
+// not configured here.
+func (m *migrationILM) catchUp() {
+	m.t.Helper()
+	ctx := m.t.Context()
+	for len(m.queue) > 0 {
+		op := <-m.queue
+		task, ok := op.(expiryTask)
+		if !ok {
+			m.t.Fatalf("unexpected expiry operation %T", op)
+		}
+		oi := task.objInfo
+		m.t.Logf("%s queued %s of %s %s", task.src, task.event.Action, oi.Name, m.label(oi.Name, oi.VersionID, oi.DeleteMarker))
+		if oi.TransitionedObject.Status == "" {
+			applyExpiryOnNonTransitionedObjects(ctx, m.z, oi, task.event, task.src)
+		}
+	}
+
+	lc, err := globalLifecycleSys.Get(m.bucket)
+	if err != nil {
+		return // no rules
+	}
+	lr, err := globalBucketObjectLockSys.Get(m.bucket)
+	if err != nil {
+		m.t.Fatal(err)
+	}
+	for _, object := range m.objects {
+		oi, err := m.z.GetObjectInfo(ctx, m.bucket, object, ObjectOptions{})
+		if isErrObjectNotFound(err) || isErrMethodNotAllowed(err) {
+			continue // deleted
+		}
+		if err != nil {
+			m.t.Fatal(err)
+		}
+		evt := evalActionFromLifecycle(ctx, *lc, lr, nil, oi)
+		if evt.Action.Delete() && oi.TransitionedObject.Status == "" {
+			applyExpiryOnNonTransitionedObjects(ctx, m.z, oi, evt, lcEventSrc_Scanner)
+		}
+	}
+}
+
+// latest returns the label of the version a read of object returns.
+func (m *migrationILM) latest(object string) string {
+	m.t.Helper()
+	oi, err := m.z.GetObjectInfo(m.t.Context(), m.bucket, object, ObjectOptions{})
+	if err != nil && (!oi.DeleteMarker || !isErrObjectNotFound(err) && !isErrMethodNotAllowed(err)) {
+		m.t.Fatal(err)
+	}
+	return m.label(object, oi.VersionID, oi.DeleteMarker)
+}
+
+// view returns the versions of each object in a pool, newest first. Delete
+// markers and versions the test did not write are "DM" and "?".
+func (m *migrationILM) view(pool int) map[string][]string {
+	m.t.Helper()
+	view := map[string][]string{}
+	for _, set := range m.z.serverPools[pool].sets {
+		err := set.listObjectsToDecommission(m.t.Context(), decomBucketInfo{Name: m.bucket}, func(entry metaCacheEntry) {
+			fivs, err := entry.fileInfoVersions(m.bucket)
+			if err != nil {
+				m.t.Errorf("%s: %v", entry.name, err)
+				return
+			}
+			for _, fi := range fivs.Versions {
+				view[entry.name] = append(view[entry.name], m.label(entry.name, fi.VersionID, fi.Deleted))
+			}
+		})
+		if err != nil {
+			m.t.Fatal(err)
+		}
+	}
+	return view
+}
+
+// check fails the test unless pool 0 is empty and pool 1 holds the versions
+// of want.
+func (m *migrationILM) check(want map[string][]string) {
+	m.t.Helper()
+	if left := m.view(0); len(left) > 0 {
+		m.t.Errorf("pool 0 still holds %v", left)
+	}
+	got := m.view(1)
+	for _, object := range m.objects {
+		if !slices.Equal(got[object], want[object]) {
+			m.t.Errorf("%s: pool 1 holds %v, want %v", object, got[object], want[object])
+		}
+	}
+}
+
+// versions writes v1, v2 and v3 of each object.
+func (m *migrationILM) versions(opts ObjectOptions, objects ...string) {
+	for _, object := range objects {
+		for _, label := range []string{"v1", "v2", "v3"} {
+			m.put(object, label, opts)
+		}
+	}
+}
+
+// aged writes v1, v2 and v3 of each object, five, four and age days ago.
+func (m *migrationILM) aged(age int, objects ...string) {
+	for _, object := range objects {
+		m.put(object, "v1", ObjectOptions{Versioned: true, MTime: UTCNow().AddDate(0, 0, -5)})
+		m.put(object, "v2", ObjectOptions{Versioned: true, MTime: UTCNow().AddDate(0, 0, -4)})
+		m.put(object, "v3", ObjectOptions{Versioned: true, MTime: UTCNow().AddDate(0, 0, -age)})
+	}
+}
+
+// migrationILMRule returns a rule for the objects under obj-.
+func migrationILMRule(action string) string {
+	return "<Rule><ID>rule</ID><Status>Enabled</Status><Filter><Prefix>obj-</Prefix></Filter>" + action + "</Rule>"
+}
+
+// migrationILMExpireCurrent expires the current versions at a past date (S3
+// expiration dates are midnights).
+func migrationILMExpireCurrent() string {
+	date := UTCNow().Truncate(24*time.Hour).AddDate(0, 0, -2).Format(time.RFC3339)
+	return migrationILMRule("<Expiration><Date>" + date + "</Date></Expiration>")
+}
+
+// testMigrationLifecycleExpiry runs the cases that decommission and rebalance
+// share.
+func testMigrationLifecycleExpiry(t *testing.T, kind string) {
+	versioned := ObjectOptions{Versioned: true}
+
+	t.Run("versioned/no-lifecycle", func(t *testing.T) {
+		m := newMigrationILM(t, versioning.Enabled, false, "")
+		m.versions(versioned, "obj-01", "keep-01")
+		m.migrate(kind)
+		m.check(map[string][]string{"obj-01": {"v3", "v2", "v1"}, "keep-01": {"v3", "v2", "v1"}})
+	})
+
+	// The expiry puts a delete marker on top of v3, which must be moved.
+	t.Run("versioned/expire-current", func(t *testing.T) {
+		m := newMigrationILM(t, versioning.Enabled, false, migrationILMExpireCurrent())
+		m.versions(versioned, "obj-01", "keep-01")
+		m.migrate(kind)
+		m.check(map[string][]string{"obj-01": {"DM", "v3", "v2", "v1"}, "keep-01": {"v3", "v2", "v1"}})
+	})
+
+	// Object lock does not hold back the delete marker, and v3 is under
+	// retention.
+	t.Run("versioned/expire-current-locked", func(t *testing.T) {
+		m := newMigrationILM(t, versioning.Enabled, true, migrationILMExpireCurrent())
+		m.put("obj-01", "v1", versioned)
+		m.put("obj-01", "v2", versioned)
+		v3 := m.put("obj-01", "v3", ObjectOptions{Versioned: true, UserDefined: map[string]string{
+			strings.ToLower(xhttp.AmzObjectLockMode):            string(objectlock.RetCompliance),
+			strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): amztime.ISO8601Format(UTCNow().Add(24 * time.Hour)),
+		}})
+		m.migrate(kind)
+		m.check(map[string][]string{"obj-01": {"DM", "v3", "v2", "v1"}})
+		oi, err := m.z.serverPools[1].GetObjectInfo(t.Context(), m.bucket, "obj-01", ObjectOptions{VersionID: v3.VersionID})
+		if mode := objectlock.GetObjectRetentionMeta(oi.UserDefined).Mode; err != nil || mode != objectlock.RetCompliance {
+			t.Errorf("v3 retention in pool 1: %q, %v", mode, err)
+		}
+	})
+
+	// In a suspended bucket the expiry puts a null delete marker on top: a
+	// versioned v3 stays, a null current version is replaced.
+	t.Run("suspended/expire-current", func(t *testing.T) {
+		m := newMigrationILM(t, versioning.Suspended, false, migrationILMExpireCurrent())
+		m.versions(versioned, "obj-01")
+		m.put("obj-null", "v1", versioned)
+		m.put("obj-null", "null", ObjectOptions{VersionSuspended: true})
+		m.migrate(kind)
+		m.check(map[string][]string{"obj-01": {"DM", "v3", "v2", "v1"}, "obj-null": {"DM", "v1"}})
+	})
+
+	// v1 has been noncurrent for four days and is deleted by the queued
+	// expiry. The source copies of v2 and v3 must be removed although v1 has
+	// not been deleted yet.
+	t.Run("versioned/noncurrent-expiration", func(t *testing.T) {
+		m := newMigrationILM(t, versioning.Enabled, false,
+			migrationILMRule("<NoncurrentVersionExpiration><NoncurrentDays>1</NoncurrentDays></NoncurrentVersionExpiration>"))
+		m.aged(0, "obj-01", "keep-01")
+		m.migrate(kind)
+		m.check(map[string][]string{"obj-01": {"v3", "v2"}, "keep-01": {"v3", "v2", "v1"}})
+	})
+
+	// Only a delete marker without noncurrent versions is expired.
+	t.Run("versioned/expired-delete-marker", func(t *testing.T) {
+		m := newMigrationILM(t, versioning.Enabled, false,
+			migrationILMRule("<Expiration><ExpiredObjectDeleteMarker>true</ExpiredObjectDeleteMarker></Expiration>"))
+		m.marker("obj-dm", "dm", UTCNow())
+		m.put("obj-top", "v1", versioned)
+		m.marker("obj-top", "dm", UTCNow())
+		m.marker("keep-dm", "dm", UTCNow())
+		m.migrate(kind)
+		m.check(map[string][]string{"obj-top": {"dm", "v1"}, "keep-dm": {"dm"}})
+	})
+
+	// ILM deletes every version of obj-01 with v3, and of obj-dm with its
+	// marker. Until the queued expiry does, the moved older versions must
+	// stay hidden.
+	for _, tc := range []struct{ name, rule, object, latest string }{
+		{"expire-all-versions", "<Expiration><Days>1</Days><ExpiredObjectAllVersions>true</ExpiredObjectAllVersions></Expiration>", "obj-01", "v3"},
+		{"expire-marker-all-versions", "<DelMarkerExpiration><Days>1</Days></DelMarkerExpiration>", "obj-dm", "dm"},
+	} {
+		t.Run("versioned/"+tc.name, func(t *testing.T) {
+			m := newMigrationILM(t, versioning.Enabled, false, migrationILMRule(tc.rule))
+			m.aged(3, "obj-01", "keep-01")
+			m.put("obj-dm", "v1", ObjectOptions{Versioned: true, MTime: UTCNow().AddDate(0, 0, -5)})
+			m.marker("obj-dm", "dm", UTCNow().AddDate(0, 0, -3))
+			m.move(kind)
+			if got := m.latest(tc.object); got != tc.latest {
+				t.Errorf("%s reads as %s before ILM deletes it, want %s; pool 0 holds %v, pool 1 holds %v",
+					tc.object, got, tc.latest, m.view(0), m.view(1))
+			}
+			m.catchUp()
+			want := map[string][]string{"obj-01": {"v3", "v2", "v1"}, "keep-01": {"v3", "v2", "v1"}, "obj-dm": {"dm", "v1"}}
+			delete(want, tc.object)
+			m.check(want)
+		})
+	}
+
+	t.Run("unversioned/expire-current", func(t *testing.T) {
+		m := newMigrationILM(t, "", false, migrationILMExpireCurrent())
+		m.put("obj-01", "null", ObjectOptions{})
+		m.put("keep-01", "null", ObjectOptions{})
+		m.migrate(kind)
+		m.check(map[string][]string{"keep-01": {"null"}})
+	})
+}
+
+func TestPoolsDecommissionLifecycleExpiry(t *testing.T) {
+	testMigrationLifecycleExpiry(t, "decommission")
+
+	// A transitioned v3 is moved as metadata; its remote data stays.
+	// Rebalance does not move transitioned versions yet.
+	t.Run("versioned/expire-current-tiered", func(t *testing.T) {
+		m := newMigrationILM(t, versioning.Enabled, false, migrationILMExpireCurrent())
+		m.put("obj-01", "v1", ObjectOptions{Versioned: true})
+		m.put("obj-01", "v2", ObjectOptions{Versioned: true})
+		v3 := m.tiered("obj-01", "v3")
+		m.migrate("decommission")
+		m.check(map[string][]string{"obj-01": {"v3", "v2", "v1"}})
+		oi, err := m.z.serverPools[1].GetObjectInfo(t.Context(), m.bucket, "obj-01", ObjectOptions{VersionID: v3.VersionID})
+		if err != nil || oi.TransitionedObject.Status != lifecycle.TransitionComplete {
+			t.Errorf("v3 in pool 1 is not transitioned: %q, %v", oi.TransitionedObject.Status, err)
+		}
+	})
+}
+
+// A current version that ILM expires in a versioned bucket stays as a
+// noncurrent version. The check after decommission must count it when it is
+// left in the pool.
+func TestPoolsDecommissionCountsExpiredCurrentVersions(t *testing.T) {
+	m := newMigrationILM(t, versioning.Enabled, false, migrationILMExpireCurrent())
+	m.migrate("decommission")
+	putConsistencyObject(t, m.z, m.bucket, "obj-01", 0, "v1", ObjectOptions{Versioned: true})
+	if err := m.z.checkAfterDecom(t.Context(), 0); err == nil {
+		t.Fatal("decommission check passed with an expired current version left in the pool")
+	}
+}
+
+// Rebalance removes the source copies of an object only when it has counted
+// all of its versions. A version that ILM deletes must be counted, or the
+// moved versions stay in pool 0 as duplicates.
+func TestPoolsRebalanceLifecycleExpiry(t *testing.T) {
+	testMigrationLifecycleExpiry(t, "rebalance")
+}

cmd/erasure-server-pool-rebalance.go

@@ -587,19 +587,28 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
 	}
 
 	for setIdx, set := range pool.sets {
-		filterLifecycle := func(bucket, object string, fi FileInfo) bool {
+		// filterLifecycle returns the action of an ILM deletion it queued
+		// for the version, which is then not moved, or NoneAction.
+		filterLifecycle := func(bucket, object string, fi FileInfo) lifecycle.Action {
 			if lc == nil {
-				return false
+				return lifecycle.NoneAction
 			}
 			versioned := vc != nil && vc.Versioned(object)
 			objInfo := fi.ToObjectInfo(bucket, object, versioned)
 
 			evt := evalActionFromLifecycle(ctx, *lc, lr, rcfg, objInfo)
-			if evt.Action.Delete() {
+			switch {
+			case evt.Action == lifecycle.DeleteAction && versioned:
+				// Expiring the latest version of a versioned bucket only puts
+				// a delete marker on top of it. The version itself survives as
+				// a noncurrent version and has to be moved like any other. The
+				// scanner applies the expiry once it lives in another pool.
+				return lifecycle.NoneAction
+			case evt.Action.Delete():
 				globalExpiryState.enqueueByDays(objInfo, evt, lcEventSrc_Rebal)
-				return true
+				return evt.Action
 			}
-			return false
+			return lifecycle.NoneAction
 		}
 
 		rebalanceEntry := func(entry metaCacheEntry) {
@@ -624,6 +633,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
 			versionsSorter(fivs.Versions).reverse()
 
 			var rebalanced int
+			var expired []FileInfo
 			for _, version := range fivs.Versions {
 				stopFn := globalRebalanceMetrics.log(rebalanceMetricRebalanceObject, poolIdx, bucket, version.Name, version.VersionID)
 
@@ -634,7 +644,15 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
 				}
 
 				// Apply lifecycle rules on the objects that are expired.
-				if filterLifecycle(bucket, version.Name, version) {
+				if action := filterLifecycle(bucket, version.Name, version); action.Delete() {
+					// Count the version, or the source copies of the moved
+					// ones stay behind as duplicates. Its own source copy
+					// goes with them when ILM deletes it alone; any other
+					// version stays until the expiry has run.
+					rebalanced++
+					if ilmDeletesVersionOnly(action) {
+						expired = append(expired, version)
+					}
 					stopFn(version.Size, errors.New("ILM expired object/version will be skipped"))
 					continue
 				}
@@ -743,7 +761,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
 			// if all versions were rebalanced, we can delete the object versions.
 			if rebalanced == len(fivs.Versions) {
 				stopFn := globalRebalanceMetrics.log(rebalanceMetricRebalanceRemoveObject, poolIdx, bucket, entry.name)
-				err := z.removeMovedVersions(ctx, bucket, entry.name, poolIdx, fivs.Versions)
+				err := z.removeMovedVersions(ctx, bucket, entry.name, poolIdx, fivs.Versions, expired)
 				stopFn(0, err)
 				auditLogRebalance(ctx, "Rebalance:DeleteObject", bucket, entry.name, "", err)
 				if err != nil {