Merge commit '8c1473a618f89fda3b89462c69c2a3c1a320d4c6' into codex/silo-lifecycle-r1-r4

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

cmd/erasure-object.go

@@ -2151,7 +2151,7 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string
 		}
 	}
 
-	if opts.Expiration.Expire {
+	if opts.Expiration.Expire && !opts.Expiration.CheckedAllPools {
 		if gerr == nil {
 			evt := evalActionFromLifecycle(ctx, *lc, rcfg, replcfg, goi)
 			var isErr bool

cmd/erasure-server-pool-decom.go

@@ -819,6 +819,10 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 				// 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() && !z.migrationExpiryApplies(ctx, objInfo, evt):
+				// Another pool holds versions with which the rule does not
+				// apply. The version moves like any other.
+				return lifecycle.NoneAction
 			case evt.Action.Delete():
 				return evt.Action
 			default:

cmd/erasure-server-pool-lifecycle.go

@@ -0,0 +1,281 @@
+// 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"
+	"sort"
+	"sync"
+	"time"
+
+	"github.com/minio/minio/internal/bucket/lifecycle"
+)
+
+// ILM decides on the versions of an object that one pool holds: the scanner
+// and the drive heal read the drives of one set, decommission and rebalance
+// walk one pool, and a GET or HEAD of a version reads the pool that holds it.
+// IsLatest, NumVersions and SuccessorModTime count only these. An object can
+// have versions in several pools: from two concurrent first writes, from
+// overwrites that went to another pool, from data movement. The expiries
+// these decisions run delete in every pool, so a version that is the latest
+// in its own pool but not of the object had a newer version hidden under a
+// delete marker or deleted with all versions, and a delete marker without
+// noncurrent versions in its pool was removed over the versions of another
+// pool. Deciding on fewer versions only delays the expiry of a noncurrent
+// version: its successor in the pool is not older than the one of the object,
+// and fewer newer versions count against NewerNoncurrentVersions.
+//
+// Before an expiry deletes in any pool, it is evaluated again with the
+// versions of the object in all pools, as the scanner would evaluate it if
+// one pool held them all. An expiry that no longer applies is not run, as the
+// recheck of a set does it: the scanner queues it again once the rule
+// applies to the object.
+
+// objectVersions returns the versions of object that a read quorum of the
+// drives of the set agree on, newest first, without free versions. It returns
+// no versions when the set does not hold the object, and an error when too
+// few drives answer to tell.
+func (er erasureObjects) objectVersions(ctx context.Context, bucket, object string) ([]FileInfo, error) {
+	disks := er.getDisks()
+	rawFileInfos, errs := readAllRawFileInfo(ctx, disks, bucket, object, false)
+	readQuorum := (len(disks) + 1) / 2
+
+	var (
+		metas   [][]xlMetaV2ShallowVersion
+		metaV   uint8
+		unknown int
+	)
+	for i, rf := range rawFileInfos {
+		if errs[i] != nil {
+			if !errors.Is(errs[i], errFileNotFound) && !errors.Is(errs[i], errVolumeNotFound) {
+				unknown++
+			}
+			continue
+		}
+		var xl xlMetaV2
+		if err := xl.LoadOrConvert(rf.Buf); err != nil {
+			unknown++
+			continue
+		}
+		metas = append(metas, xl.versions)
+		metaV = xl.metaV
+	}
+	if len(metas) < readQuorum {
+		if len(metas)+unknown < readQuorum {
+			return nil, nil // not in this set
+		}
+		return nil, toObjectErr(errErasureReadQuorum, bucket, object)
+	}
+
+	merged := mergeXLV2Versions(readQuorum, false, 0, metas...)
+	if len(merged) == 0 {
+		for _, versions := range metas {
+			if len(versions) > 0 {
+				return nil, toObjectErr(errErasureReadQuorum, bucket, object)
+			}
+		}
+		return nil, nil
+	}
+	all, err := xlMetaV2{versions: merged, metaV: metaV}.ListVersions(bucket, object, false)
+	if err != nil {
+		return nil, toObjectErr(err, bucket, object)
+	}
+	versions := all[:0]
+	for _, fi := range all {
+		if !fi.TierFreeVersion() {
+			versions = append(versions, fi)
+		}
+	}
+	return versions, nil
+}
+
+// lifecycleEventsAllPools returns the versions of object in all pools,
+// newest first, and the lifecycle event of each as the scanner evaluates it
+// on them. A version with copies in several pools, as an interrupted move
+// leaves it, counts once.
+func (z *erasureServerPools) lifecycleEventsAllPools(ctx context.Context, bucket, object string) ([]ObjectInfo, []lifecycle.Event, error) {
+	lc, err := globalLifecycleSys.Get(bucket)
+	if err != nil {
+		if errors.Is(err, BucketLifecycleNotFound{Bucket: bucket}) {
+			return nil, nil, nil
+		}
+		return nil, nil, err
+	}
+	lr, err := globalBucketObjectLockSys.Get(bucket)
+	if err != nil {
+		return nil, nil, err
+	}
+	rcfg, err := getReplicationConfig(ctx, bucket)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	poolVersions := make([][]FileInfo, len(z.serverPools))
+	errs := make([]error, len(z.serverPools))
+	var wg sync.WaitGroup
+	for i, pool := range z.serverPools {
+		wg.Add(1)
+		go func(i int, set *erasureObjects) {
+			defer wg.Done()
+			poolVersions[i], errs[i] = set.objectVersions(ctx, bucket, object)
+		}(i, pool.getHashedSet(object))
+	}
+	wg.Wait()
+	for _, err := range errs {
+		if err != nil {
+			return nil, nil, err
+		}
+	}
+
+	var all []FileInfo
+	for _, versions := range poolVersions {
+		all = append(all, versions...)
+	}
+	// Each pool lists its versions newest first. On equal times the lower
+	// pool comes first, as reads choose it.
+	sort.SliceStable(all, func(i, j int) bool { return all[i].ModTime.After(all[j].ModTime) })
+	seen := make(map[string]struct{}, len(all))
+	versions := all[:0]
+	for _, fi := range all {
+		if _, ok := seen[fi.VersionID]; ok {
+			continue
+		}
+		seen[fi.VersionID] = struct{}{}
+		versions = append(versions, fi)
+	}
+
+	vc, _ := globalBucketVersioningSys.Get(bucket)
+	versioned := vc != nil && vc.Versioned(decodeDirObject(object))
+	objInfos := make([]ObjectInfo, len(versions))
+	objOpts := make([]lifecycle.ObjectOpts, len(versions))
+	for i := range versions {
+		fi := versions[i]
+		fi.IsLatest = i == 0
+		fi.NumVersions = len(versions)
+		fi.SuccessorModTime = time.Time{}
+		if i > 0 {
+			fi.SuccessorModTime = versions[i-1].ModTime
+		}
+		objInfos[i] = fi.ToObjectInfo(bucket, object, versioned)
+		objOpts[i] = objInfos[i].ToLifecycleOpts()
+	}
+	events, err := lifecycle.NewEvaluator(*lc).WithLockRetention(&lr).WithReplicationConfig(rcfg).Eval(objOpts)
+	if err != nil {
+		return nil, nil, err
+	}
+	return objInfos, events, nil
+}
+
+// lifecycleAllPools holds the versions of an object in all pools and their
+// lifecycle events, see lifecycleEventsAllPools.
+type lifecycleAllPools struct {
+	versions []ObjectInfo
+	events   []lifecycle.Event
+	err      error
+}
+
+// checkExpiryAllPools returns nil when the ILM expiry that opts describe
+// still applies to object with the versions it has in all pools, see
+// lifecycleAllPools.expiryApplies.
+func (z *erasureServerPools) checkExpiryAllPools(ctx context.Context, bucket, object string, opts ObjectOptions) error {
+	var all lifecycleAllPools
+	all.versions, all.events, all.err = z.lifecycleEventsAllPools(ctx, bucket, object)
+	return all.expiryApplies(bucket, object, opts)
+}
+
+// expiryApplies returns nil when the ILM expiry that opts describe applies
+// to the versions, and otherwise the error a set returns for an expiry that
+// no longer applies:
+//   - an expiry of all versions (DeletePrefix) needs the latest version to
+//     expire with all versions;
+//   - a delete marker (no VersionID) hides the latest version, which has to
+//     expire;
+//   - a version (VersionID) has to expire by itself: a noncurrent version,
+//     or a delete marker without noncurrent versions.
+func (all lifecycleAllPools) expiryApplies(bucket, object string, opts ObjectOptions) error {
+	if all.err != nil {
+		if _, ok := all.err.(InsufficientReadQuorum); ok {
+			return InsufficientWriteQuorum{}
+		}
+		return all.err
+	}
+	name := decodeDirObject(object)
+	switch {
+	case opts.DeletePrefix:
+		if len(all.events) > 0 && all.events[0].Action.DeleteAll() {
+			return nil
+		}
+	case opts.VersionID == "":
+		if len(all.events) > 0 && (all.events[0].Action == lifecycle.DeleteAction || all.events[0].Action == lifecycle.DeleteAllVersionsAction) {
+			return nil
+		}
+	default:
+		for i, oi := range all.versions {
+			if oi.VersionID == opts.VersionID || oi.VersionID == "" && opts.VersionID == nullVersionID {
+				if all.events[i].Action == lifecycle.DeleteVersionAction {
+					return nil
+				}
+				break
+			}
+		}
+		return VersionNotFound{Bucket: bucket, Object: name, VersionID: opts.VersionID}
+	}
+	return ObjectNotFound{Bucket: bucket, Object: name}
+}
+
+// expiryNeedsAllPools reports whether an ILM expiry in opts deletes or hides
+// versions by a rule that depends on the other versions of the object. A
+// restored copy expires by its own restore date. Without versions each copy
+// of an object is a write of its own, and a set rechecks the rule on its
+// copy (deleteObjectFromAllPools).
+func (z *erasureServerPools) expiryNeedsAllPools(opts ObjectOptions) bool {
+	if !opts.Expiration.Expire || z.SinglePool() || opts.Transition.ExpireRestored {
+		return false
+	}
+	return opts.DeletePrefix || opts.VersionID != "" || opts.Versioned || opts.VersionSuspended
+}
+
+// migrationExpiryApplies reports whether the ILM expiry evt, which a
+// decommission or rebalance decided for oi on the versions of the pool it
+// walks, applies with the versions of the object in all pools too. Only then
+// may the walk leave the version to the expiry instead of moving it. A
+// version that is the latest in the walked pool can be a noncurrent version
+// of the object, and a delete marker alone in that pool can hide versions in
+// another one. Such a version is moved like any other, and the scanner
+// applies the rule in the pool that holds the object then.
+func (z *erasureServerPools) migrationExpiryApplies(ctx context.Context, oi ObjectInfo, evt lifecycle.Event) bool {
+	// The options the expiry deletes with, see
+	// applyExpiryOnNonTransitionedObjects and expireTransitionedObject.
+	opts := ObjectOptions{
+		Expiration:       ExpirationOptions{Expire: true},
+		Versioned:        globalBucketVersioningSys.PrefixEnabled(oi.Bucket, oi.Name),
+		VersionSuspended: globalBucketVersioningSys.PrefixSuspended(oi.Bucket, oi.Name),
+	}
+	if evt.Action.DeleteVersioned() {
+		opts.VersionID = oi.VersionID
+	}
+	if evt.Action.DeleteAll() {
+		opts.DeletePrefix = true
+	}
+	opts.Transition.ExpireRestored = evt.Action.DeleteRestored()
+	if !z.expiryNeedsAllPools(opts) {
+		return true
+	}
+	return z.checkExpiryAllPools(ctx, oi.Bucket, encodeDirObject(oi.Name), opts) == nil
+}

cmd/erasure-server-pool-lifecycle_test.go

@@ -0,0 +1,501 @@
+// 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 (
+	"fmt"
+	"io"
+	"slices"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/minio/minio/internal/bucket/lifecycle"
+	"github.com/minio/minio/internal/bucket/versioning"
+)
+
+// The scanner and the walks of decommission and rebalance evaluate ILM on the
+// versions of a key in one pool: IsLatest, NumVersions and SuccessorModTime
+// count only these. A key can have versions in several pools: two concurrent
+// first writes, overwrites that went to another pool. The expiry they decide
+// runs in every pool.
+
+// splitILM lays out the versions of keys in the two pools of migrationILM.
+type splitILM struct {
+	*migrationILM
+	ids    map[string]string // object and label -> version ID
+	bodies map[string]string // body -> label
+}
+
+func newSplitILM(t *testing.T, status versioning.State, rules string) *splitILM {
+	t.Helper()
+	return &splitILM{migrationILM: newMigrationILM(t, status, false, rules), ids: map[string]string{}, bodies: map[string]string{}}
+}
+
+// splitVersion is a version of a key in a test layout.
+type splitVersion struct {
+	pools  []int // the pools that hold a copy of it
+	label  string
+	days   float64 // its age
+	marker bool
+	tiered bool // its data is in a remote tier
+	null   bool // a null version (unversioned or suspended bucket)
+	size   int  // its size, at least the size of its label
+}
+
+func (s *splitILM) write(object string, v splitVersion) {
+	s.t.Helper()
+	ctx := s.t.Context()
+	mtime := UTCNow().Add(-time.Duration(v.days * float64(24*time.Hour)))
+	versionID := mustGetUUID()
+	if v.null {
+		versionID = ""
+	}
+	for _, pool := range v.pools {
+		if v.marker {
+			_, err := s.z.serverPools[pool].DeleteObject(ctx, s.bucket, object,
+				ObjectOptions{Versioned: true, VersionID: versionID, DeleteMarker: true, MTime: mtime})
+			if err != nil {
+				s.t.Fatal(err)
+			}
+			continue
+		}
+		opts := ObjectOptions{Versioned: !v.null, VersionID: versionID, MTime: mtime}
+		if v.null && s.status() == versioning.Suspended {
+			opts = ObjectOptions{VersionSuspended: true, MTime: mtime}
+		}
+		body := object + " " + v.label
+		if len(body) < v.size {
+			body += strings.Repeat(".", v.size-len(body))
+		}
+		putConsistencyObject(s.t, s.z, s.bucket, object, pool, body, opts)
+		s.bodies[body] = v.label
+		if v.tiered {
+			vid := versionID
+			if vid == "" {
+				vid = nullVersionID
+			}
+			infos, errs := readAllFileInfo(ctx, s.z.serverPools[pool].getHashedSet(object).getDisks(), "", s.bucket, object, vid, false, false)
+			if errs[0] != nil {
+				s.t.Fatal(errs[0])
+			}
+			fi := infos[0]
+			fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName = lifecycle.TransitionComplete, "WARM", "remote-"+v.label
+			if err := s.z.serverPools[pool].DecomTieredObject(ctx, s.bucket, object, fi, ObjectOptions{}); err != nil {
+				s.t.Fatal(err)
+			}
+		}
+	}
+	if v.marker {
+		s.markers[s.key(object, versionID)] = v.label
+	} else {
+		s.labels[s.key(object, versionID)] = v.label
+	}
+	s.ids[object+" "+v.label] = versionID
+	s.track(object)
+}
+
+func (s *splitILM) status() versioning.State {
+	vc, _ := globalBucketVersioningSys.Get(s.bucket)
+	if vc == nil {
+		return ""
+	}
+	return vc.Status
+}
+
+// scan applies ILM to every object in a pool as the scanner of a drive of
+// that pool does, from the xl.meta of that drive, and runs the queued
+// expiries as an expiry worker does.
+func (s *splitILM) scan(pool int) {
+	s.t.Helper()
+	ctx := s.t.Context()
+	lc, err := globalLifecycleSys.Get(s.bucket)
+	if err != nil {
+		s.t.Fatal(err)
+	}
+	lr, err := globalBucketObjectLockSys.Get(s.bucket)
+	if err != nil {
+		s.t.Fatal(err)
+	}
+	vc, _ := globalBucketVersioningSys.Get(s.bucket)
+	for _, object := range s.objects {
+		disk := s.z.serverPools[pool].getHashedSet(object).getDisks()[0]
+		rf, err := disk.ReadXL(ctx, s.bucket, object, false)
+		if err != nil {
+			continue // not in this pool
+		}
+		fivs, err := getFileInfoVersions(rf.Buf, s.bucket, object, false)
+		if err != nil {
+			s.t.Fatal(err)
+		}
+		objInfos := make([]ObjectInfo, len(fivs.Versions))
+		for i, fi := range fivs.Versions {
+			objInfos[i] = fi.ToObjectInfo(s.bucket, object, vc != nil && vc.Versioned(object))
+		}
+		item := scannerItem{bucket: s.bucket, objectName: object, lifeCycle: lc}
+		item.applyActions(ctx, s.z, objInfos, lr, &sizeSummary{}, nil)
+	}
+	s.runQueue()
+}
+
+// head applies ILM to a version as a HEAD or GET with its versionId does.
+func (s *splitILM) head(object, label string) lifecycle.Action {
+	s.t.Helper()
+	ctx := s.t.Context()
+	oi, err := s.z.GetObjectInfo(ctx, s.bucket, object, ObjectOptions{VersionID: s.ids[object+" "+label]})
+	if err != nil {
+		s.t.Fatal(err)
+	}
+	lc, err := globalLifecycleSys.Get(s.bucket)
+	if err != nil {
+		s.t.Fatal(err)
+	}
+	lr, err := globalBucketObjectLockSys.Get(s.bucket)
+	if err != nil {
+		s.t.Fatal(err)
+	}
+	evt := evalActionFromLifecycle(ctx, *lc, lr, nil, oi)
+	if evt.Action.Delete() {
+		applyExpiryRule(evt, lcEventSrc_s3HeadObject, oi)
+	}
+	s.runQueue()
+	return evt.Action
+}
+
+// runQueue runs the queued expiries with an expiry worker.
+func (s *splitILM) runQueue() {
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		globalExpiryState.Worker(s.queue)
+	}()
+	s.queue <- nil // the worker returns once it ran every expiry queued before
+	<-done
+}
+
+// get returns the label of the version a GET of object reads, or "-" when it
+// reads as deleted.
+func (s *splitILM) get(object string) string {
+	s.t.Helper()
+	ctx := s.t.Context()
+	oi, err := s.z.GetObjectInfo(ctx, s.bucket, object, ObjectOptions{})
+	if err != nil {
+		if isErrObjectNotFound(err) || isErrVersionNotFound(err) || isErrMethodNotAllowed(err) {
+			return "-"
+		}
+		return fmt.Sprintf("<%v>", err)
+	}
+	if oi.VersionID != "" && oi.VersionID != nullVersionID {
+		return s.label(object, oi.VersionID, false)
+	}
+	// Copies of a null version in two pools are told apart by their data.
+	gr, err := s.z.GetObjectNInfo(ctx, s.bucket, object, nil, nil, ObjectOptions{})
+	if err != nil {
+		return fmt.Sprintf("<%v>", err)
+	}
+	defer gr.Close()
+	data, err := io.ReadAll(gr)
+	if err != nil {
+		s.t.Fatal(err)
+	}
+	if l, ok := s.bodies[string(data)]; ok {
+		return l
+	}
+	return "?" + string(data)
+}
+
+// views returns the versions of each object in each pool, newest first.
+func (s *splitILM) views() []map[string][]string {
+	views := make([]map[string][]string, len(s.z.serverPools))
+	for pool := range s.z.serverPools {
+		views[pool] = s.view(pool)
+	}
+	return views
+}
+
+type splitKey struct {
+	name     string
+	versions []splitVersion
+	before   string     // what a GET reads before ILM ran
+	after    string     // what a GET reads once ILM ran
+	want     [][]string // versions in each pool once ILM ran, nil to skip
+}
+
+// run lays out keys, runs two scanner cycles over all pools and checks that
+// a GET of each key never reads anything but before or after, and the
+// versions each pool holds at the end. The second cycle must not change
+// anything.
+func (s *splitILM) run(keys []splitKey) {
+	s.t.Helper()
+	for _, k := range keys {
+		for _, v := range k.versions {
+			s.write(k.name, v)
+		}
+		if got := s.get(k.name); got != k.before {
+			s.t.Fatalf("%s: fixture reads %s, want %s", k.name, got, k.before)
+		}
+	}
+	var first []map[string][]string
+	for cycle := 1; cycle <= 2; cycle++ {
+		for pool := range s.z.serverPools {
+			s.scan(pool)
+			for _, k := range keys {
+				if got := s.get(k.name); got != k.before && got != k.after {
+					s.t.Errorf("%s: reads %s after the scanner of pool %d (cycle %d), want %s", k.name, got, pool, cycle, k.after)
+				}
+			}
+		}
+		if cycle == 1 {
+			first = s.views()
+		}
+	}
+	last := s.views()
+	for _, k := range keys {
+		if got := s.get(k.name); got != k.after {
+			s.t.Errorf("%s: reads %s once ILM ran, want %s", k.name, got, k.after)
+		}
+		for pool := range last {
+			if !slices.Equal(first[pool][k.name], last[pool][k.name]) {
+				s.t.Errorf("%s: the second scanner cycle changed pool %d from %v to %v", k.name, pool, first[pool][k.name], last[pool][k.name])
+			}
+			if k.want != nil && !slices.Equal(last[pool][k.name], k.want[pool]) {
+				s.t.Errorf("%s: pool %d holds %v, want %v", k.name, pool, last[pool][k.name], k.want[pool])
+			}
+		}
+	}
+}
+
+const splitILMRules = "<Rule><ID>exp</ID><Status>Enabled</Status><Filter><Prefix>exp-</Prefix></Filter>" +
+	"<Expiration><Days>1</Days></Expiration></Rule>" +
+	"<Rule><ID>eav</ID><Status>Enabled</Status><Filter><Prefix>eav-</Prefix></Filter>" +
+	"<Expiration><Days>1</Days><ExpiredObjectAllVersions>true</ExpiredObjectAllVersions></Expiration></Rule>" +
+	"<Rule><ID>edm</ID><Status>Enabled</Status><Filter><Prefix>edm-</Prefix></Filter>" +
+	"<Expiration><ExpiredObjectDeleteMarker>true</ExpiredObjectDeleteMarker></Expiration></Rule>" +
+	"<Rule><ID>dme</ID><Status>Enabled</Status><Filter><Prefix>dme-</Prefix></Filter>" +
+	"<DelMarkerExpiration><Days>1</Days></DelMarkerExpiration></Rule>" +
+	"<Rule><ID>ncv</ID><Status>Enabled</Status><Filter><Prefix>ncv-</Prefix></Filter>" +
+	"<NoncurrentVersionExpiration><NoncurrentDays>1</NoncurrentDays></NoncurrentVersionExpiration></Rule>" +
+	"<Rule><ID>exs</ID><Status>Enabled</Status><Filter><And><Prefix>exs-</Prefix><ObjectSizeGreaterThan>100</ObjectSizeGreaterThan></And></Filter>" +
+	"<Expiration><Days>1</Days></Expiration></Rule>" +
+	"<Rule><ID>nnv</ID><Status>Enabled</Status><Filter><Prefix>nnv-</Prefix></Filter>" +
+	"<NoncurrentVersionExpiration><NoncurrentDays>1</NoncurrentDays><NewerNoncurrentVersions>1</NewerNoncurrentVersions></NoncurrentVersionExpiration></Rule>"
+
+func splitAt(label string, days float64, pools ...int) splitVersion {
+	return splitVersion{pools: pools, label: label, days: days}
+}
+
+func splitMarker(label string, days float64, pools ...int) splitVersion {
+	return splitVersion{pools: pools, label: label, days: days, marker: true}
+}
+
+func splitTiered(label string, days float64, pools ...int) splitVersion {
+	return splitVersion{pools: pools, label: label, days: days, tiered: true}
+}
+
+func splitNull(label string, days float64, pools ...int) splitVersion {
+	return splitVersion{pools: pools, label: label, days: days, null: true}
+}
+
+// splitILMOnePool are keys in one pool. ILM applies to them as usual.
+func splitILMOnePool(pool int) []splitKey {
+	in := func(vs ...string) [][]string {
+		want := [][]string{nil, nil}
+		want[pool] = vs
+		return want
+	}
+	return []splitKey{
+		{"exp-one", []splitVersion{splitAt("v1", 5, pool), splitAt("v2", 4, pool)}, "v2", "-", in("DM", "v2", "v1")},
+		{"exp-one-tiered", []splitVersion{splitAt("v1", 5, pool), splitTiered("v2", 4, pool)}, "v2", "-", in("DM", "v2", "v1")},
+		{"eav-one", []splitVersion{splitAt("v1", 5, pool), splitAt("v2", 4, pool)}, "v2", "-", in()},
+		{"edm-one", []splitVersion{splitMarker("dm", 3, pool)}, "-", "-", in()},
+		{"dme-one", []splitVersion{splitAt("v1", 5, pool), splitMarker("dm", 3, pool)}, "-", "-", in()},
+		{"ncv-one", []splitVersion{splitAt("v1", 5, pool), splitAt("v2", 4, pool), splitAt("v3", 0.01, pool)}, "v3", "v3", in("v3", "v2")},
+		{"nnv-one", []splitVersion{splitAt("v1", 5, pool), splitAt("v2", 4, pool), splitAt("v3", 3, pool), splitAt("v4", 2, pool)}, "v4", "v4", in("v4", "v3")},
+	}
+}
+
+func TestPoolsLifecycleSplitObject(t *testing.T) {
+	t.Run("versioned", func(t *testing.T) {
+		s := newSplitILM(t, versioning.Enabled, splitILMRules)
+		s.run([]splitKey{
+			// Expiration with Days applies to the current version v2 only.
+			{"exp-old-in-0", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 0.01, 1)}, "v2", "v2", [][]string{{"v1"}, {"v2"}}},
+			{"exp-old-in-1", []splitVersion{splitAt("v2", 0.01, 0), splitAt("v1", 5, 1)}, "v2", "v2", [][]string{{"v2"}, {"v1"}}},
+			{"exp-tiered-old-in-0", []splitVersion{splitTiered("v1", 5, 0), splitAt("v2", 0.01, 1)}, "v2", "v2", [][]string{{"v1"}, {"v2"}}},
+			// v2 is old enough: one delete marker hides the key.
+			{"exp-both-old", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 4, 1)}, "v2", "-", [][]string{{"DM", "v1"}, {"v2"}}},
+			// The rule applies to v2 only by its size.
+			{"exs-big-in-1", []splitVersion{splitAt("v1", 5, 0), {pools: []int{1}, label: "v2", days: 4, size: 200}}, "v2", "-", [][]string{{"DM", "v1"}, {"v2"}}},
+			// The marker hides v1 and is not a marker without noncurrent
+			// versions; v1 is not the current version.
+			{"exp-marker-in-0", []splitVersion{splitAt("v1", 5, 1), splitMarker("dm", 3, 0)}, "-", "-", [][]string{{"dm"}, {"v1"}}},
+			// ExpiredObjectAllVersions applies to the current version v2 only.
+			{"eav-old-in-0", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 0.01, 1)}, "v2", "v2", [][]string{{"v1"}, {"v2"}}},
+			{"eav-old-in-1", []splitVersion{splitAt("v2", 0.01, 0), splitAt("v1", 5, 1)}, "v2", "v2", [][]string{{"v2"}, {"v1"}}},
+			{"eav-tiered-old-in-0", []splitVersion{splitTiered("v1", 5, 0), splitAt("v2", 0.01, 1)}, "v2", "v2", [][]string{{"v1"}, {"v2"}}},
+			{"eav-both-old", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 4, 1)}, "v2", "-", [][]string{nil, nil}},
+			// A copy of v1 in both pools, as an interrupted move leaves it.
+			{"eav-copies", []splitVersion{splitAt("v1", 5, 0, 1)}, "v1", "-", [][]string{nil, nil}},
+			// The marker hides v1 in the other pool.
+			{"edm-marker-in-0", []splitVersion{splitAt("v1", 5, 1), splitMarker("dm", 3, 0)}, "-", "-", [][]string{{"dm"}, {"v1"}}},
+			{"edm-marker-in-1", []splitVersion{splitAt("v1", 5, 0), splitMarker("dm", 3, 1)}, "-", "-", [][]string{{"v1"}, {"dm"}}},
+			{"edm-copies", []splitVersion{splitMarker("dm", 3, 0, 1)}, "-", "-", [][]string{nil, nil}},
+			// v2 in the other pool is the current version, the marker does
+			// not expire.
+			{"dme-new-in-1", []splitVersion{splitMarker("dm", 3, 0), splitAt("v2", 0.01, 1)}, "v2", "v2", [][]string{{"dm"}, {"v2"}}},
+			{"dme-new-in-0", []splitVersion{splitAt("v2", 0.01, 0), splitMarker("dm", 3, 1)}, "v2", "v2", [][]string{{"v2"}, {"dm"}}},
+			// The marker is the current version and expires with v1.
+			{"dme-old-in-0", []splitVersion{splitAt("v1", 5, 0), splitMarker("dm", 3, 1)}, "-", "-", [][]string{nil, nil}},
+			// v1 is noncurrent since v2 was written, v2 since v3 was.
+			{"ncv-split", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 4, 1), splitAt("v3", 2, 0)}, "v3", "v3", [][]string{{"v3"}, {"v2"}}},
+		})
+	})
+	t.Run("versioned/one-pool", func(t *testing.T) {
+		for pool := range 2 {
+			s := newSplitILM(t, versioning.Enabled, splitILMRules)
+			s.run(splitILMOnePool(pool))
+		}
+	})
+	t.Run("versioned/head-version", func(t *testing.T) {
+		s := newSplitILM(t, versioning.Enabled, splitILMRules)
+		keys := []splitKey{
+			{"exp-old-in-0", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 0.01, 1)}, "v2", "v2", [][]string{{"v1"}, {"v2"}}},
+			{"eav-old-in-0", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 0.01, 1)}, "v2", "v2", [][]string{{"v1"}, {"v2"}}},
+		}
+		for _, k := range keys {
+			for _, v := range k.versions {
+				s.write(k.name, v)
+			}
+			act := s.head(k.name, "v1")
+			t.Logf("%s: HEAD of v1 evaluates %s", k.name, act)
+			if got := s.get(k.name); got != k.after {
+				t.Errorf("%s: reads %s after a HEAD of v1, want %s", k.name, got, k.after)
+			}
+			views := s.views()
+			for pool := range views {
+				if !slices.Equal(views[pool][k.name], k.want[pool]) {
+					t.Errorf("%s: pool %d holds %v, want %v", k.name, pool, views[pool][k.name], k.want[pool])
+				}
+			}
+		}
+	})
+	t.Run("suspended", func(t *testing.T) {
+		s := newSplitILM(t, versioning.Suspended, splitILMRules)
+		s.run([]splitKey{
+			{"exp-null-old-in-0", []splitVersion{splitNull("v1", 5, 0), splitAt("v2", 0.01, 1)}, "v2", "v2", [][]string{{"v1"}, {"v2"}}},
+		})
+	})
+	t.Run("unversioned", func(t *testing.T) {
+		s := newSplitILM(t, "", splitILMRules)
+		s.run([]splitKey{
+			{"eav-old-in-0", []splitVersion{splitNull("old", 5, 0), splitNull("new", 0.01, 1)}, "new", "new", nil},
+			{"exp-old-in-0", []splitVersion{splitNull("old", 5, 0), splitNull("new", 0.01, 1)}, "new", "new", nil},
+		})
+		views := s.views()
+		for object, want := range map[string]int{"eav-old-in-0": 1, "exp-old-in-0": 0} {
+			if len(views[0][object]) != want || len(views[1][object]) != 1 {
+				t.Errorf("%s: pools hold %v and %v, want %d and 1 copies", object, views[0][object], views[1][object], want)
+			}
+		}
+	})
+}
+
+// walk moves pool 0 out with a decommission or a rebalance pass, as move
+// does, but reports a decommission that did not complete instead of stopping
+// the test, so the reads are still checked.
+func (s *splitILM) walk(kind string) {
+	s.t.Helper()
+	ctx := s.t.Context()
+	if kind == "rebalance" {
+		migrationTestWalk(s.t, s.z, s.bucket, 0, kind)
+		s.z.rebalMu.Lock()
+		s.z.rebalMeta.PoolStats[0].Info.Status = rebalCompleted
+		s.z.rebalMu.Unlock()
+		return
+	}
+	if err := s.z.StartDecommission(ctx, 0); err != nil {
+		s.t.Fatal(err)
+	}
+	s.z.doDecommissionInRoutine(ctx, 0)
+	s.z.poolMetaMutex.RLock()
+	info := *s.z.poolMeta.Pools[0].Decommission
+	s.z.poolMetaMutex.RUnlock()
+	if !info.Complete || info.Failed || info.ItemsDecommissionFailed > 0 {
+		s.t.Errorf("decommission did not complete: complete=%t failed=%t itemsFailed=%d; pool 0 holds %v",
+			info.Complete, info.Failed, info.ItemsDecommissionFailed, s.view(0))
+	}
+}
+
+// Rules that expire a whole object (DelMarkerExpiration,
+// ExpiredObjectAllVersions) or a delete marker without noncurrent versions
+// (ExpiredObjectDeleteMarker) depend on every version of the object. A walk
+// sees the versions in its own pool only.
+func TestPoolsMigrationLifecycleWholeObject(t *testing.T) {
+	recent := 1.0 / 24
+	objects := []struct {
+		name     string
+		versions []splitVersion // oldest first
+		before   string         // what a GET reads before the walk
+		after    string         // what a GET reads once ILM caught up
+		want     []string       // versions in pool 1 once ILM caught up, newest first
+	}{
+		// The marker expires and takes v1 with it.
+		{"dme-one-pool", []splitVersion{splitAt("v1", 5, 0), splitMarker("dm", 3, 0)}, "-", "-", nil},
+		{"dme-older-in-pool-1", []splitVersion{splitAt("v1", 5, 1), splitMarker("dm", 3, 0)}, "-", "-", nil},
+		// v2 is the current version, the marker does not expire.
+		{"dme-newer-in-pool-1", []splitVersion{splitMarker("dm", 3, 0), splitAt("v2", recent, 1)}, "v2", "v2", []string{"v2", "dm"}},
+		// v2 expires and takes v1 with it.
+		{"eav-one-pool", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 4, 0)}, "v2", "-", nil},
+		{"eav-older-in-pool-1", []splitVersion{splitAt("v1", 5, 1), splitAt("v2", 4, 0)}, "v2", "-", nil},
+		// v2 is the current version and does not expire.
+		{"eav-newer-in-pool-1", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", recent, 1)}, "v2", "v2", []string{"v2", "v1"}},
+		// The marker hides v1, it is not a marker without noncurrent versions.
+		{"edm-older-in-pool-1", []splitVersion{splitAt("v1", 5, 1), splitMarker("dm", 3, 0)}, "-", "-", []string{"dm", "v1"}},
+		{"keep-marker", []splitVersion{splitAt("v1", 5, 0), splitMarker("dm", 3, 0)}, "-", "-", []string{"dm", "v1"}},
+		{"keep-versions", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 4, 0)}, "v2", "v2", []string{"v2", "v1"}},
+	}
+	for _, kind := range []string{"decommission", "rebalance"} {
+		t.Run(kind, func(t *testing.T) {
+			s := newSplitILM(t, versioning.Enabled, splitILMRules)
+			for _, o := range objects {
+				for _, v := range o.versions {
+					s.write(o.name, v)
+				}
+				if got := s.get(o.name); got != o.before {
+					t.Fatalf("%s: fixture reads %s, want %s", o.name, got, o.before)
+				}
+			}
+			s.walk(kind)
+			for _, o := range objects {
+				if got := s.get(o.name); got != o.before && got != o.after {
+					t.Errorf("%s: reads %s after the %s, before ILM caught up; want %s or %s", o.name, got, kind, o.before, o.after)
+				}
+			}
+			s.runQueue()
+			s.scan(1)
+			want := map[string][]string{}
+			for _, o := range objects {
+				want[o.name] = o.want
+				if got := s.get(o.name); got != o.after {
+					t.Errorf("%s: reads %s once ILM caught up, want %s", o.name, got, o.after)
+				}
+			}
+			s.check(want)
+		})
+	}
+}

cmd/erasure-server-pool-rebalance.go

@@ -605,6 +605,10 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
 				// 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() && !z.migrationExpiryApplies(ctx, objInfo, evt):
+				// Another pool holds versions with which the rule does not
+				// apply. The version moves like any other.
+				return lifecycle.NoneAction
 			case evt.Action.Delete():
 				return evt.Action
 			}

cmd/erasure-server-pool.go

@@ -1248,6 +1248,16 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
 	ctx = lkctx.Context()
 	defer lk.Unlock(lkctx)
 
+	// An ILM expiry is decided on the versions in one pool. Under the lock,
+	// check it with the versions in all pools; the sets then take it as
+	// decided, without a recheck by the versions each one holds.
+	if z.expiryNeedsAllPools(opts) {
+		if err = z.checkExpiryAllPools(ctx, bucket, object, opts); err != nil {
+			return ObjectInfo{Name: decodeDirObject(object), VersionID: opts.VersionID}, err
+		}
+		opts.Expiration.CheckedAllPools = true
+	}
+
 	if opts.DeletePrefix {
 		return ObjectInfo{}, z.deletePrefix(ctx, bucket, object)
 	}
@@ -1517,10 +1527,28 @@ func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, o
 		batch = make([]ObjectToDelete, 0, len(objects))
 		batchIdx = make([]int, 0, len(objects))
 		data, dataIdx := make([]ObjectToDelete, 0, len(objects)), make([]int, 0, len(objects))
+		ilm := make(map[string]lifecycleAllPools)
 		for i, obj := range objects {
 			if derrs[i] != nil {
 				continue
 			}
+			// A version that ILM expires is checked with the versions of
+			// its object in all pools, see checkExpiryAllPools. When the
+			// expiry no longer applies, nothing is deleted and nothing is
+			// left to replicate.
+			vopts := ObjectOptions{Expiration: opts.Expiration, VersionID: obj.VersionID}
+			if obj.VersionID != "" && z.expiryNeedsAllPools(vopts) {
+				all, ok := ilm[obj.ObjectName]
+				if !ok {
+					all.versions, all.events, all.err = z.lifecycleEventsAllPools(ctx, bucket, obj.ObjectName)
+					ilm[obj.ObjectName] = all
+				}
+				if err := all.expiryApplies(bucket, obj.ObjectName, vopts); err != nil {
+					dobjects[i] = DeletedObject{ObjectName: obj.ObjectName, VersionID: obj.VersionID}
+					derrs[i] = err
+					continue
+				}
+			}
 			if deletesData(obj, opts) {
 				if z.copyWithoutDeleteQuorum(ctx, bucket, obj.ObjectName, obj.VersionID) {
 					derrs[i] = InsufficientWriteQuorum{}

cmd/object-api-interface.go

@@ -159,6 +159,10 @@ type WalkOptions struct {
 // ExpirationOptions represents object options for object expiration at objectLayer.
 type ExpirationOptions struct {
 	Expire bool
+	// CheckedAllPools: the expiry was checked with the versions of the
+	// object in all pools (checkExpiryAllPools). A set does not check it
+	// again with the versions it holds alone.
+	CheckedAllPools bool
 }
 
 // TransitionOptions represents object options for transition ObjectLayer operation

cmd/object-api-interface_gen.go

@@ -79,10 +79,13 @@ func (z BucketOptions) Msgsize() (s int) {
 // MarshalMsg implements msgp.Marshaler
 func (z ExpirationOptions) MarshalMsg(b []byte) (o []byte, err error) {
 	o = msgp.Require(b, z.Msgsize())
-	// map header, size 1
+	// map header, size 2
 	// string "Expire"
-	o = append(o, 0x81, 0xa6, 0x45, 0x78, 0x70, 0x69, 0x72, 0x65)
+	o = append(o, 0x82, 0xa6, 0x45, 0x78, 0x70, 0x69, 0x72, 0x65)
 	o = msgp.AppendBool(o, z.Expire)
+	// string "CheckedAllPools"
+	o = append(o, 0xaf, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x65, 0x64, 0x41, 0x6c, 0x6c, 0x50, 0x6f, 0x6f, 0x6c, 0x73)
+	o = msgp.AppendBool(o, z.CheckedAllPools)
 	return
 }
 
@@ -110,6 +113,12 @@ func (z *ExpirationOptions) UnmarshalMsg(bts []byte) (o []byte, err error) {
 				err = msgp.WrapError(err, "Expire")
 				return
 			}
+		case "CheckedAllPools":
+			z.CheckedAllPools, bts, err = msgp.ReadBoolBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "CheckedAllPools")
+				return
+			}
 		default:
 			bts, err = msgp.Skip(bts)
 			if err != nil {
@@ -124,7 +133,7 @@ func (z *ExpirationOptions) UnmarshalMsg(bts []byte) (o []byte, err error) {
 
 // Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
 func (z ExpirationOptions) Msgsize() (s int) {
-	s = 1 + 7 + msgp.BoolSize
+	s = 1 + 7 + msgp.BoolSize + 16 + msgp.BoolSize
 	return
 }
 

cmd/object-handlers-common.go

@@ -467,6 +467,7 @@ func deleteObjectVersions(ctx context.Context, o ObjectLayer, bucket string, toD
 		deletedObjs, errs := o.DeleteObjects(ctx, bucket, toDel, ObjectOptions{
 			PrefixEnabledFn:  vc.PrefixEnabled,
 			VersionSuspended: vc.Suspended(),
+			Expiration:       ExpirationOptions{Expire: true},
 		})
 
 		for i, dobj := range deletedObjs {