diff --git a/cmd/bucket-lifecycle.go b/cmd/bucket-lifecycle.go
index 4b22130b8..e7a6d99e5 100644
--- a/cmd/bucket-lifecycle.go
+++ b/cmd/bucket-lifecycle.go
@@ -37,6 +37,7 @@ import (
"github.com/minio/minio/internal/amztime"
sse "github.com/minio/minio/internal/bucket/encryption"
"github.com/minio/minio/internal/bucket/lifecycle"
+ "github.com/minio/minio/internal/bucket/replication"
"github.com/minio/minio/internal/event"
xhttp "github.com/minio/minio/internal/http"
"github.com/minio/minio/internal/logger"
@@ -631,31 +632,63 @@ func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *Ob
return err
}
- // Delete remote object from warm-tier
- //
- // Unless the expiry only puts a delete marker on top of the version, as
- // for an object that is not transitioned: with versioning enabled, or
- // suspended and the version is not the null version, the version stays
- // as a noncurrent version and must remain readable. Its remote object is
- // removed later, through the free version added when the version itself
- // is deleted.
+ // Recheck the lifecycle decision and create the hot-tier tombstone before
+ // touching the remote tier. If the lifecycle configuration changed after
+ // the scanner queued this task, DeleteObject rejects the stale action while
+ // the remote object is still protected.
opts.VersionSuspended = globalBucketVersioningSys.PrefixSuspended(oi.Bucket, oi.Name)
keepRemote := opts.VersionID == "" &&
(opts.Versioned || (opts.VersionSuspended && oi.VersionID != "" && oi.VersionID != nullVersionID))
+ if lcEvent.Action == lifecycle.DeleteAction || lcEvent.Action == lifecycle.DeleteVersionAction {
+ opts.SetEvalMetadataFn(func(goi *ObjectInfo, gerr error) (dsc ReplicateDecision, err error) {
+ return checkReplicateDelete(ctx, oi.Bucket, ObjectToDelete{
+ ObjectV: ObjectV{
+ ObjectName: goi.Name,
+ VersionID: opts.VersionID,
+ },
+ }, *goi, opts, gerr), nil
+ })
+ }
+
+ // Delete the hot-tier namespace first. For a transitioned version this
+ // leaves a durable free-version record until the remote delete succeeds;
+ // that record is the retry point if the process stops between the two
+ // operations.
+ delInfo, err := objectAPI.DeleteObject(ctx, oi.Bucket, oi.Name, opts)
+ if err != nil {
+ return err
+ }
+
if !keepRemote {
- err := deleteObjectFromRemoteTier(ctx, oi.TransitionedObject.Name, oi.TransitionedObject.VersionID, oi.TransitionedObject.Tier)
- if err == nil {
- // Skip adding free version since we successfully deleted the
- // remote object
- opts.SkipFreeVersion = true
- } else {
+ if err := deleteObjectFromRemoteTier(ctx, oi.TransitionedObject.Name, oi.TransitionedObject.VersionID, oi.TransitionedObject.Tier); err != nil {
transitionLogIf(ctx, err)
}
}
- // Now, delete object from hot-tier namespace
- if _, err := objectAPI.DeleteObject(ctx, oi.Bucket, oi.Name, opts); err != nil {
- return err
+ // Queue replication of the ILM-created delete marker or version purge,
+ // mirroring the DELETE handler's post-delete scheduling.
+ if delInfo.ReplicationStatus == replication.Pending || delInfo.VersionPurgeStatus == replication.VersionPurgePending {
+ dmVersionID, versionID := "", ""
+ if delInfo.DeleteMarker && delInfo.VersionPurgeStatus.Empty() {
+ dmVersionID = delInfo.VersionID
+ } else {
+ versionID = delInfo.VersionID
+ }
+ if delInfo.Name == "" {
+ delInfo.Name = oi.Name
+ }
+ scheduleReplicationDelete(ctx, DeletedObjectReplicationInfo{
+ DeletedObject: DeletedObject{
+ ObjectName: delInfo.Name,
+ VersionID: versionID,
+ DeleteMarkerVersionID: dmVersionID,
+ DeleteMarkerMTime: DeleteMarkerMTime{delInfo.ModTime},
+ DeleteMarker: delInfo.DeleteMarker,
+ ReplicationState: delInfo.ReplicationState(),
+ },
+ Bucket: oi.Bucket,
+ EventType: ReplicateILMDelete,
+ }, objectAPI)
}
// Send audit for the lifecycle delete operation
diff --git a/cmd/bucket-replication.go b/cmd/bucket-replication.go
index 492657e91..ee26ef355 100644
--- a/cmd/bucket-replication.go
+++ b/cmd/bucket-replication.go
@@ -2062,6 +2062,8 @@ const (
ReplicateIncoming = "replicate:incoming"
// ReplicateIncomingDelete - audit trail of inline replication of deletes.
ReplicateIncomingDelete = "replicate:incoming:delete"
+ // ReplicateILMDelete - audit trail of delete replication triggered by ILM expiry.
+ ReplicateILMDelete = "replicate:ilm:delete"
// ReplicateHeal - audit trail for healing of failed/pending replications
ReplicateHeal = "replicate:heal"
diff --git a/cmd/data-scanner.go b/cmd/data-scanner.go
index 13d0b99f0..329772df9 100644
--- a/cmd/data-scanner.go
+++ b/cmd/data-scanner.go
@@ -1130,15 +1130,9 @@ eventLoop:
applyExpiryRule(event, lcEventSrc_Scanner, oi)
case lifecycle.DeleteVersionAction: // noncurrent versions expiration
- opts := objOpts[idx]
remainingVersions--
size = 0
- toDel = append(toDel, ObjectToDelete{
- ObjectV: ObjectV{
- ObjectName: opts.Name,
- VersionID: opts.VersionID,
- },
- })
+ toDel = append(toDel, ilmPurgeToDelete(ctx, i.bucket, oi, i.replication.Config != nil))
noncurrentEvents = append(noncurrentEvents, event)
case lifecycle.TransitionAction, lifecycle.TransitionVersionAction:
@@ -1240,6 +1234,21 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay
// use prefix delete on exact object (this is an optimization to avoid fan-out calls)
opts.DeletePrefixObject = true
}
+
+ // ILM-created delete markers and version purges must carry the same
+ // replication decision as API-created deletes. DeleteAll is intentionally
+ // excluded: it is an object-local purge and has no single version to
+ // replicate.
+ if lcEvent.Action == lifecycle.DeleteAction || lcEvent.Action == lifecycle.DeleteVersionAction {
+ opts.SetEvalMetadataFn(func(oi *ObjectInfo, gerr error) (dsc ReplicateDecision, err error) {
+ return checkReplicateDelete(ctx, obj.Bucket, ObjectToDelete{
+ ObjectV: ObjectV{
+ ObjectName: oi.Name,
+ VersionID: opts.VersionID,
+ },
+ }, *oi, opts, gerr), nil
+ })
+ }
var (
dobj ObjectInfo
err error
@@ -1276,6 +1285,29 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay
dobj = obj
}
+ // Queue replication of the ILM-created delete marker or version purge,
+ // mirroring the DELETE handler's post-delete scheduling.
+ if dobj.ReplicationStatus == replication.Pending || dobj.VersionPurgeStatus == replication.VersionPurgePending {
+ dmVersionID, versionID := "", ""
+ if dobj.DeleteMarker && dobj.VersionPurgeStatus.Empty() {
+ dmVersionID = dobj.VersionID
+ } else {
+ versionID = dobj.VersionID
+ }
+ scheduleReplicationDelete(ctx, DeletedObjectReplicationInfo{
+ DeletedObject: DeletedObject{
+ ObjectName: dobj.Name,
+ VersionID: versionID,
+ DeleteMarkerVersionID: dmVersionID,
+ DeleteMarkerMTime: DeleteMarkerMTime{dobj.ModTime},
+ DeleteMarker: dobj.DeleteMarker,
+ ReplicationState: dobj.ReplicationState(),
+ },
+ Bucket: obj.Bucket,
+ EventType: ReplicateILMDelete,
+ }, objLayer)
+ }
+
tags := newLifecycleAuditEvent(src, lcEvent).Tags()
tags["version-id"] = dobj.VersionID
@@ -1309,6 +1341,33 @@ func applyExpiryRule(event lifecycle.Event, src lcEventSrc, obj ObjectInfo) {
globalExpiryState.enqueueByDays(obj, event, src)
}
+// ilmPurgeToDelete builds the bulk delete entry for an ILM-initiated removal
+// of a noncurrent version. The entry carries a pending version-purge state so
+// the purge is replicated and retried like an API version purge.
+func ilmPurgeToDelete(ctx context.Context, bucket string, oi ObjectInfo, replicate bool) ObjectToDelete {
+ otd := ObjectToDelete{
+ ObjectV: ObjectV{
+ ObjectName: oi.Name,
+ VersionID: oi.VersionID,
+ },
+ }
+ if !replicate {
+ return otd
+ }
+ delOpts := ObjectOptions{
+ Versioned: globalBucketVersioningSys.PrefixEnabled(bucket, oi.Name),
+ VersionSuspended: globalBucketVersioningSys.PrefixSuspended(bucket, oi.Name),
+ }
+ dsc := checkReplicateDelete(ctx, bucket, otd, oi, delOpts, nil)
+ if !dsc.ReplicateAny() {
+ return otd
+ }
+ otd.VersionPurgeStatus = replication.VersionPurgePending
+ otd.VersionPurgeStatuses = dsc.PendingStatus()
+ otd.ReplicateDecisionStr = dsc.String()
+ return otd
+}
+
// objectPath returns the prefix and object name.
func (i *scannerItem) objectPath() string {
return pathJoin(i.prefix, i.objectName)
diff --git a/cmd/erasure-server-pool-consistency.go b/cmd/erasure-server-pool-consistency.go
index 188ded74b..ad3e24d38 100644
--- a/cmd/erasure-server-pool-consistency.go
+++ b/cmd/erasure-server-pool-consistency.go
@@ -127,15 +127,119 @@ func (z *erasureServerPools) checkDataMovementCopies(ctx context.Context, bucket
}, nil
}
+type migrationExpiredVersion struct {
+ version FileInfo
+ event lifecycle.Event
+ lifecycleUpdatedAt time.Time
+ source lcEventSrc
+}
+
+// applyMigrationExpiry runs the logical expiry path before migration retires
+// any remaining source copies. This keeps replication and tier cleanup under
+// their normal ownership and makes lifecycle configuration changes fail safe.
+func (z *erasureServerPools) applyMigrationExpiry(ctx context.Context, bucket, object string, source int, expired []migrationExpiredVersion) error {
+ if len(expired) == 0 {
+ return nil
+ }
+
+ lc, lifecycleUpdatedAt, err := globalBucketMetadataSys.GetLifecycleConfig(bucket)
+ if err != nil {
+ return PreConditionFailed{}
+ }
+ for _, pending := range expired {
+ if !pending.lifecycleUpdatedAt.Equal(lifecycleUpdatedAt) {
+ return PreConditionFailed{}
+ }
+ }
+ lr, err := globalBucketObjectLockSys.Get(bucket)
+ if err != nil {
+ return err
+ }
+ rcfg, err := getReplicationConfig(ctx, bucket)
+ if err != nil {
+ return err
+ }
+ encodedObject := encodeDirObject(object)
+ seenDeleteAll := false
+ eventSource := lcEventSrc_Decom
+ for _, pending := range expired {
+ if pending.source != lcEventSrc_None {
+ eventSource = pending.source
+ break
+ }
+ }
+
+ for _, pending := range expired {
+ if pending.event.Action.DeleteAll() {
+ if seenDeleteAll {
+ continue
+ }
+ seenDeleteAll = true
+ }
+
+ versionID := pending.version.VersionID
+ if versionID == "" {
+ versionID = nullVersionID
+ }
+ copies, err := z.objectPoolInfos(ctx, bucket, encodedObject, ObjectOptions{
+ VersionID: versionID,
+ NoAuditLog: true,
+ })
+ if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
+ continue
+ }
+ if err != nil {
+ return err
+ }
+
+ current := copies[0].ObjInfo
+ for _, copy := range copies {
+ if copy.Index == source {
+ current = copy.ObjInfo
+ break
+ }
+ }
+ current.Bucket = bucket
+ current.Name = object
+ currentEvent := evalActionFromLifecycle(ctx, *lc, lr, rcfg, current)
+ if currentEvent.Action != pending.event.Action {
+ return PreConditionFailed{}
+ }
+
+ var ok bool
+ if current.TransitionedObject.Status != "" && !pending.event.Action.DeleteAll() {
+ ok = applyExpiryOnTransitionedObject(ctx, z, current, pending.event, eventSource)
+ } else {
+ ok = applyExpiryOnNonTransitionedObjects(ctx, z, current, pending.event, eventSource)
+ }
+ if ok {
+ continue
+ }
+
+ // A concurrent expiry may have removed the addressed version after
+ // the first lookup. Anything still present means the logical delete
+ // was not committed and migration must stop before purging its source.
+ if _, err := z.objectPoolInfos(ctx, bucket, encodedObject, ObjectOptions{VersionID: versionID, NoAuditLog: true}); err == nil {
+ return PreConditionFailed{}
+ } else if !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
+ return err
+ }
+ }
+ return nil
+}
+
// 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.
-// 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 {
+// Versions in expired were not moved because ILM deletes them. They must stay
+// in the source until the ordinary expiry path has deleted them: that path
+// owns replication state and tier cleanup, and it re-evaluates the current
+// lifecycle rule. A migration cleanup must never turn an asynchronous ILM
+// decision into an unconditional physical purge.
+func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, object string, source int, versions []FileInfo, expired []migrationExpiredVersion) error {
+ if err := z.applyMigrationExpiry(ctx, bucket, object, source, expired); err != nil {
+ return err
+ }
object = encodeDirObject(object)
lk := z.NewNSLock(bucket, object)
lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
@@ -171,9 +275,14 @@ func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, ob
if !src.ModTime.Equal(version.ModTime) || src.DeleteMarker != version.Deleted {
return PreConditionFailed{}
}
+ if isExpiredVersion(expired, version) {
+ // The logical expiry path owns this version. Keeping the source is
+ // also the retry point if that path failed or the pool check races it.
+ continue
+ }
// Successful movement is not enough if the destination changed again.
// Keep the source unless another copy still preserves its read result.
- if !isExpiredVersion(expired, version) && (len(retained) == 0 || retained[0].ObjInfo.ModTime.Before(src.ModTime)) {
+ if len(retained) == 0 || retained[0].ObjInfo.ModTime.Before(src.ModTime) {
return PreConditionFailed{}
}
_, err = z.serverPools[source].DeleteObject(ctx, bucket, object, ObjectOptions{
@@ -195,8 +304,9 @@ func ilmDeletesVersionOnly(action lifecycle.Action) bool {
return action == lifecycle.DeleteAction || action == lifecycle.DeleteVersionAction
}
-func isExpiredVersion(expired []FileInfo, version FileInfo) bool {
- for _, fi := range expired {
+func isExpiredVersion(expired []migrationExpiredVersion, version FileInfo) bool {
+ for _, pending := range expired {
+ fi := pending.version
if fi.VersionID == version.VersionID && fi.ModTime.Equal(version.ModTime) && fi.Deleted == version.Deleted {
return true
}
diff --git a/cmd/erasure-server-pool-decom.go b/cmd/erasure-server-pool-decom.go
index d5a557451..bc5e1c514 100644
--- a/cmd/erasure-server-pool-decom.go
+++ b/cmd/erasure-server-pool-decom.go
@@ -769,6 +769,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
var vc *versioning.Versioning
var lc *lifecycle.Lifecycle
+ var lifecycleUpdatedAt time.Time
var lr objectlock.Retention
var rcfg *replication.Config
if bi.Name != minioMetaBucket {
@@ -778,7 +779,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
}
// Check if the current bucket has a configured lifecycle policy
- lc, err = globalLifecycleSys.Get(bi.Name)
+ lc, lifecycleUpdatedAt, err = globalBucketMetadataSys.GetLifecycleConfig(bi.Name)
if err != nil && !errors.Is(err, BucketLifecycleNotFound{Bucket: bi.Name}) {
return err
}
@@ -796,8 +797,8 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
}
for setIdx, set := range pool.sets {
- // filterLifecycle returns the action of an ILM deletion it queued
- // for the version, which is then not moved, or NoneAction.
+ // filterLifecycle returns an ILM deletion that migration must handle
+ // before retiring the source copy, or NoneAction.
filterLifecycle := func(bucket, object string, fi FileInfo) lifecycle.Action {
if lc == nil {
return lifecycle.NoneAction
@@ -808,7 +809,9 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
evt := evalActionFromLifecycle(ctx, *lc, lr, rcfg, objInfo)
switch {
case evt.Action.DeleteRestored(): // if restored copy has expired, delete it synchronously
- applyExpiryOnTransitionedObject(ctx, z, objInfo, evt, lcEventSrc_Decom)
+ if !applyExpiryOnTransitionedObject(ctx, z, objInfo, evt, lcEventSrc_Decom) {
+ return evt.Action
+ }
return lifecycle.NoneAction
case evt.Action == lifecycle.DeleteAction && versioned:
// Expiring the latest version of a versioned bucket only puts
@@ -817,7 +820,6 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
// scanner applies the expiry once it lives in another pool.
return lifecycle.NoneAction
case evt.Action.Delete():
- globalExpiryState.enqueueByDays(objInfo, evt, lcEventSrc_Decom)
return evt.Action
default:
return lifecycle.NoneAction
@@ -841,17 +843,23 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
versionsSorter(fivs.Versions).reverse()
var decommissioned int
- var expired []FileInfo
+ var expired []migrationExpiredVersion
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 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)
+ if ilmDeletesVersionOnly(action) || (action.DeleteAll() && !version.IsRemote()) {
+ // Run the logical expiry before retiring the source.
+ // A full-object expiry is safe here only for hot-tier
+ // metadata; transitioned objects retain their tier cleanup
+ // ownership in the ordinary expiry worker.
+ expired = append(expired, migrationExpiredVersion{
+ version: version,
+ event: lifecycle.Event{Action: action},
+ lifecycleUpdatedAt: lifecycleUpdatedAt,
+ source: lcEventSrc_Decom,
+ })
}
stopFn(version.Size, errors.New("ILM expired object/version will be skipped"))
continue
@@ -1115,6 +1123,15 @@ func (z *erasureServerPools) decommissionInBackground(ctx context.Context, idx i
stopFn(0, err)
return err
}
+ // Do not retire the bucket from the resumable queue until every
+ // source version and free version is gone. In particular, an ILM
+ // action queued during the walk may still be pending; marking the
+ // bucket complete here would allow the source pool to be removed
+ // before that action performs its logical and tier cleanup.
+ if err := z.checkAfterDecomBucket(ctx, idx, bucket); err != nil {
+ stopFn(0, err)
+ return err
+ }
stopFn(0, nil)
z.poolMetaMutex.Lock()
@@ -1165,106 +1182,46 @@ func (z *erasureServerPools) removeFreeVersions(ctx context.Context, bucket, obj
return nil
}
+// checkAfterDecomBucket verifies that a source bucket is physically empty
+// before it is removed from the resumable decommission queue. ILM decisions
+// are deliberately not treated as an exemption: an expired version remains
+// until the expiry path has performed its logical delete, replication, and
+// tier cleanup.
+func (z *erasureServerPools) checkAfterDecomBucket(ctx context.Context, idx int, bi decomBucketInfo) error {
+ versionsFound := 0
+ for _, set := range z.serverPools[idx].sets {
+ if err := set.listObjectsToDecommission(ctx, bi, func(entry metaCacheEntry) {
+ if !entry.isObject() {
+ return
+ }
+ if bi.Name == minioMetaBucket && strings.Contains(entry.name, dataUsageCacheName) {
+ return
+ }
+ fivs, err := entry.fileInfoVersions(bi.Name)
+ if err != nil {
+ return
+ }
+ versionsFound += len(fivs.Versions) + len(fivs.FreeVersions)
+ }); err != nil {
+ return err
+ }
+ }
+ if versionsFound > 0 {
+ return fmt.Errorf("at least %d object(s)/version(s) were found in bucket `%s` after decommissioning", versionsFound, bi.Name)
+ }
+ return nil
+}
+
func (z *erasureServerPools) checkAfterDecom(ctx context.Context, idx int) error {
buckets, err := z.getBucketsToDecommission(ctx)
if err != nil {
return err
}
-
- pool := z.serverPools[idx]
- for _, set := range pool.sets {
- for _, bi := range buckets {
- var vc *versioning.Versioning
- var lc *lifecycle.Lifecycle
- var lr objectlock.Retention
- var rcfg *replication.Config
- if bi.Name != minioMetaBucket {
- vc, err = globalBucketVersioningSys.Get(bi.Name)
- if err != nil {
- return err
- }
-
- // Check if the current bucket has a configured lifecycle policy
- lc, err = globalLifecycleSys.Get(bi.Name)
- if err != nil && !errors.Is(err, BucketLifecycleNotFound{Bucket: bi.Name}) {
- return err
- }
-
- // Check if bucket is object locked.
- lr, err = globalBucketObjectLockSys.Get(bi.Name)
- if err != nil {
- return err
- }
-
- rcfg, err = getReplicationConfig(ctx, bi.Name)
- if err != nil {
- return err
- }
- }
-
- filterLifecycle := func(bucket, object string, fi FileInfo) bool {
- if lc == nil {
- return false
- }
- versioned := vc != nil && vc.Versioned(object)
- objInfo := fi.ToObjectInfo(bucket, object, versioned)
-
- evt := evalActionFromLifecycle(ctx, *lc, lr, rcfg, objInfo)
- switch {
- 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
- default:
- return false
- }
- }
-
- var versionsFound int
- if err = set.listObjectsToDecommission(ctx, bi, func(entry metaCacheEntry) {
- if !entry.isObject() {
- return
- }
-
- // `.usage-cache.bin` still exists, must be not readable ignore it.
- if bi.Name == minioMetaBucket && strings.Contains(entry.name, dataUsageCacheName) {
- // skipping bucket usage cache name, as its autogenerated.
- return
- }
-
- fivs, err := entry.fileInfoVersions(bi.Name)
- if err != nil {
- return
- }
-
- var ignored int
-
- for _, version := range fivs.Versions {
- // Apply lifecycle rules on the objects that are expired.
- if filterLifecycle(bi.Name, version.Name, version) {
- ignored++
- continue
- }
- }
-
- // 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
- }
-
- if versionsFound > 0 {
- return fmt.Errorf("at least %d object(s)/version(s) were found in bucket `%s` after decommissioning", versionsFound, bi.Name)
- }
+ for _, bi := range buckets {
+ if err := z.checkAfterDecomBucket(ctx, idx, bi); err != nil {
+ return err
}
}
-
return nil
}
diff --git a/cmd/erasure-server-pool-migration-ilm_test.go b/cmd/erasure-server-pool-migration-ilm_test.go
index 8eb108750..29791c994 100644
--- a/cmd/erasure-server-pool-migration-ilm_test.go
+++ b/cmd/erasure-server-pool-migration-ilm_test.go
@@ -18,6 +18,9 @@
package cmd
import (
+ "context"
+ "errors"
+ "net/http"
"slices"
"strings"
"testing"
@@ -26,6 +29,7 @@ import (
"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/replication"
"github.com/minio/minio/internal/bucket/versioning"
xhttp "github.com/minio/minio/internal/http"
)
@@ -234,16 +238,6 @@ func (m *migrationILM) catchUp() {
}
}
-// 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 {
@@ -384,12 +378,14 @@ func testMigrationLifecycleExpiry(t *testing.T, kind string) {
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", "1true", "obj-01", "v3"},
- {"expire-marker-all-versions", "1", "obj-dm", "dm"},
+ // Each rule deletes every version of one object. Migration must execute the
+ // logical DeleteAll action before it retires the source pool, so the object
+ // selected by the rule is gone from both pools before any later scan.
+ for _, tc := range []struct {
+ name, rule, deleted string
+ }{
+ {"expire-all-versions", "1true", "obj-01"},
+ {"expire-marker-all-versions", "1", "obj-dm"},
} {
t.Run("versioned/"+tc.name, func(t *testing.T) {
m := newMigrationILM(t, versioning.Enabled, false, migrationILMRule(tc.rule))
@@ -397,13 +393,17 @@ func testMigrationLifecycleExpiry(t *testing.T, kind string) {
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))
+ if len(m.queue) != 0 {
+ t.Fatalf("migration left stale expiry tasks queued: %d", len(m.queue))
}
+ want := map[string][]string{
+ "obj-01": {"v3", "v2", "v1"},
+ "keep-01": {"v3", "v2", "v1"},
+ "obj-dm": {"dm", "v1"},
+ }
+ delete(want, tc.deleted)
+ m.check(want)
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)
})
}
@@ -417,6 +417,138 @@ func testMigrationLifecycleExpiry(t *testing.T, kind string) {
})
}
+// A lifecycle action found during the migration walk must not be applied
+// after the bucket's lifecycle configuration changes. The source copy stays
+// available for a fresh scan under the new policy.
+func TestPoolsMigrationLifecycleChangePreservesSource(t *testing.T) {
+ m := newMigrationILM(t, versioning.Enabled, false, migrationILMExpireCurrent())
+ oi := m.put("obj-01", "v1", ObjectOptions{Versioned: true})
+ _, updatedAt, err := globalBucketMetadataSys.GetLifecycleConfig(m.bucket)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ meta, err := globalBucketMetadataSys.Get(m.bucket)
+ if err != nil {
+ t.Fatal(err)
+ }
+ meta.LifecycleConfigXML = []byte(`changedEnabledother-1`)
+ meta.LifecycleConfigUpdatedAt = updatedAt.Add(time.Second)
+ if err := meta.Save(t.Context(), m.z); err != nil {
+ t.Fatal(err)
+ }
+ globalBucketMetadataSys.Set(m.bucket, meta)
+
+ err = m.z.applyMigrationExpiry(t.Context(), m.bucket, oi.Name, 0, []migrationExpiredVersion{{
+ version: FileInfo{
+ Name: oi.Name,
+ VersionID: oi.VersionID,
+ ModTime: oi.ModTime,
+ },
+ event: lifecycle.Event{Action: lifecycle.DeleteAction},
+ lifecycleUpdatedAt: updatedAt,
+ source: lcEventSrc_Decom,
+ }})
+ if err == nil {
+ t.Fatal("migration expiry succeeded after lifecycle configuration changed")
+ }
+ if _, err := m.z.serverPools[0].GetObjectInfo(t.Context(), m.bucket, oi.Name, ObjectOptions{VersionID: oi.VersionID}); err != nil {
+ t.Fatalf("source version was removed after lifecycle configuration changed: %v", err)
+ }
+}
+
+// A migration must route an ILM version purge through the normal delete
+// replication path before it can retire the source copy. Otherwise the
+// source pool can disappear while the replica still retains the version.
+func TestPoolsMigrationLifecycleExpiryReplication(t *testing.T) {
+ m := newMigrationILM(t, versioning.Enabled, false,
+ migrationILMRule("1"))
+ m.aged(0, "obj-01")
+
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ stats := NewReplicationStats(ctx, nil)
+ oldStats := globalReplicationStats.Swap(stats)
+ defer globalReplicationStats.Store(oldStats)
+
+ target := newILMTestTarget(t)
+ registerILMTestTarget(t, m.z, m.bucket, target, func(w http.ResponseWriter, r *http.Request) {
+ opts := ObjectOptions{VersionID: r.URL.Query().Get("versionId"), Versioned: true}
+ switch r.Method {
+ case http.MethodHead:
+ oi, err := m.z.GetObjectInfo(r.Context(), target.bucket, "obj-01", opts)
+ if err != nil {
+ writeErrorResponseHeadersOnly(w, toAPIError(r.Context(), err))
+ return
+ }
+ if oi.DeleteMarker {
+ w.Header().Set(xhttp.AmzDeleteMarker, "true")
+ w.Header().Set(xhttp.AmzVersionID, oi.VersionID)
+ }
+ w.WriteHeader(http.StatusOK)
+ case http.MethodDelete:
+ target.deletes.Add(1)
+ opts.DeleteMarker = r.Header.Get(xhttp.MinIOSourceDeleteMarker) == "true"
+ opts.ReplicationRequest = true
+ opts.SetReplicaStatus(replication.Replica)
+ if _, err := m.z.DeleteObject(r.Context(), target.bucket, "obj-01", opts); err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
+ writeErrorResponse(r.Context(), w, toAPIError(r.Context(), err), r.URL)
+ return
+ }
+ w.WriteHeader(http.StatusNoContent)
+ default:
+ t.Errorf("unexpected remote method %s", r.Method)
+ }
+ })
+ setILMTestReplicationConfig(t, m.bucket, target.arn, target.bucket)
+ restorePool, receive := newILMTestReplicationPool(ctx, m.z, stats)
+ defer restorePool()
+
+ if err := m.z.StartDecommission(ctx, 0); err != nil {
+ 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 {
+ t.Fatalf("decommission completed before ILM purge replication: complete=%t failed=%t", info.Complete, info.Failed)
+ }
+
+ deletion := receive()
+ if deletion.ObjectName != "obj-01" || deletion.VersionID == "" || deletion.VersionPurgeStatus() != replication.VersionPurgePending {
+ t.Fatalf("unexpected migration purge: %+v", deletion)
+ }
+ if result := replicateDelete(ctx, deletion, m.z); result.VersionPurgeStatus() != replication.VersionPurgeComplete {
+ t.Fatalf("migration purge replication result: %+v", result)
+ }
+ if target.deletes.Load() != 1 {
+ t.Fatalf("replication target saw %d deletes, want 1", target.deletes.Load())
+ }
+
+ // A failed decommission may already have moved the source pool's bucket
+ // metadata. Recreate the metadata directory before the resumable retry,
+ // as the server's scanner would do in production.
+ for _, set := range m.z.serverPools[0].sets {
+ for _, disk := range set.getDisks() {
+ if err := disk.MakeVol(ctx, pathJoin(minioMetaBucket, bucketMetaPrefix)); err != nil && !errors.Is(err, errVolumeExists) {
+ t.Fatal(err)
+ }
+ }
+ }
+
+ if err := m.z.StartDecommission(ctx, 0); err != nil {
+ 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 {
+ t.Fatalf("decommission did not complete after ILM purge replication: complete=%t failed=%t", info.Complete, info.Failed)
+ }
+}
+
func TestPoolsDecommissionLifecycleExpiry(t *testing.T) {
testMigrationLifecycleExpiry(t, "decommission")
diff --git a/cmd/erasure-server-pool-rebalance.go b/cmd/erasure-server-pool-rebalance.go
index 0ff18b378..a23a9eb3d 100644
--- a/cmd/erasure-server-pool-rebalance.go
+++ b/cmd/erasure-server-pool-rebalance.go
@@ -542,6 +542,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
var vc *versioning.Versioning
var lc *lifecycle.Lifecycle
+ var lifecycleUpdatedAt time.Time
var lr objectlock.Retention
var rcfg *replication.Config
if bucket != minioMetaBucket {
@@ -551,7 +552,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
}
// Check if the current bucket has a configured lifecycle policy
- lc, err = globalLifecycleSys.Get(bucket)
+ lc, lifecycleUpdatedAt, err = globalBucketMetadataSys.GetLifecycleConfig(bucket)
if err != nil && !errors.Is(err, BucketLifecycleNotFound{Bucket: bucket}) {
return err
}
@@ -587,8 +588,8 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
}
for setIdx, set := range pool.sets {
- // filterLifecycle returns the action of an ILM deletion it queued
- // for the version, which is then not moved, or NoneAction.
+ // filterLifecycle returns an ILM deletion that migration must handle
+ // before retiring the source copy, or NoneAction.
filterLifecycle := func(bucket, object string, fi FileInfo) lifecycle.Action {
if lc == nil {
return lifecycle.NoneAction
@@ -605,7 +606,6 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
// scanner applies the expiry once it lives in another pool.
return lifecycle.NoneAction
case evt.Action.Delete():
- globalExpiryState.enqueueByDays(objInfo, evt, lcEventSrc_Rebal)
return evt.Action
}
return lifecycle.NoneAction
@@ -633,7 +633,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
versionsSorter(fivs.Versions).reverse()
var rebalanced int
- var expired []FileInfo
+ var expired []migrationExpiredVersion
for _, version := range fivs.Versions {
stopFn := globalRebalanceMetrics.log(rebalanceMetricRebalanceObject, poolIdx, bucket, version.Name, version.VersionID)
@@ -650,8 +650,13 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
// 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)
+ if ilmDeletesVersionOnly(action) || action.DeleteAll() {
+ expired = append(expired, migrationExpiredVersion{
+ version: version,
+ event: lifecycle.Event{Action: action},
+ lifecycleUpdatedAt: lifecycleUpdatedAt,
+ source: lcEventSrc_Rebal,
+ })
}
stopFn(version.Size, errors.New("ILM expired object/version will be skipped"))
continue
diff --git a/cmd/object-handlers-common.go b/cmd/object-handlers-common.go
index c7b4e9b94..ddfc8dde3 100644
--- a/cmd/object-handlers-common.go
+++ b/cmd/object-handlers-common.go
@@ -28,6 +28,7 @@ import (
"github.com/minio/minio/internal/amztime"
"github.com/minio/minio/internal/bucket/lifecycle"
+ "github.com/minio/minio/internal/bucket/replication"
"github.com/minio/minio/internal/crypto"
"github.com/minio/minio/internal/event"
"github.com/minio/minio/internal/hash"
@@ -499,6 +500,20 @@ func deleteObjectVersions(ctx context.Context, o ObjectLayer, bucket string, toD
}
}
sendEvent(evArgs)
+
+ // ILM purges carry a delete replication decision. Queue the purge
+ // after the local delete, just as the multi-object DELETE handler
+ // does for API-originated deletes.
+ if errs[i] == nil || isErrObjectNotFound(errs[i]) || isErrVersionNotFound(errs[i]) {
+ if dobj.VersionPurgeStatus() == replication.VersionPurgePending ||
+ dobj.DeleteMarkerReplicationStatus() == replication.Pending {
+ scheduleReplicationDelete(ctx, DeletedObjectReplicationInfo{
+ DeletedObject: dobj,
+ Bucket: bucket,
+ EventType: ReplicateILMDelete,
+ }, o)
+ }
+ }
}
}
}
diff --git a/cmd/replication-ilm-delete_test.go b/cmd/replication-ilm-delete_test.go
new file mode 100644
index 000000000..3ac04a871
--- /dev/null
+++ b/cmd/replication-ilm-delete_test.go
@@ -0,0 +1,333 @@
+// Copyright (c) 2015-2026 MinIO, Inc.
+// Copyright (c) 2026 PGSTY
+//
+// This file is part of MinIO 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 .
+
+package cmd
+
+import (
+ "context"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/minio/madmin-go/v3"
+ "github.com/minio/minio-go/v7"
+ "github.com/minio/minio/internal/auth"
+ "github.com/minio/minio/internal/bucket/lifecycle"
+ "github.com/minio/minio/internal/bucket/replication"
+ xhttp "github.com/minio/minio/internal/http"
+ "github.com/minio/minio/internal/once"
+)
+
+// ilmDeleteTarget models one replication destination for ILM expiry tests.
+type ilmDeleteTarget struct {
+ arn, bucket string
+ deletes atomic.Int32
+}
+
+func newILMTestTarget(t *testing.T) *ilmDeleteTarget {
+ return &ilmDeleteTarget{arn: "arn:minio:replication::" + mustGetUUID() + ":bucket", bucket: getRandomBucketName()}
+}
+
+func registerILMTestTarget(t *testing.T, obj ObjectLayer, bucket string, target *ilmDeleteTarget, handler http.HandlerFunc) {
+ t.Helper()
+ if err := obj.MakeBucket(t.Context(), target.bucket, MakeBucketOptions{}); err != nil {
+ t.Fatal(err)
+ }
+ remote := httptest.NewServer(handler)
+ t.Cleanup(remote.Close)
+ client, err := minio.New(strings.TrimPrefix(remote.URL, "http://"), &minio.Options{Region: "us-east-1", MaxRetries: 1})
+ if err != nil {
+ t.Fatal(err)
+ }
+ globalBucketTargetSys.Lock()
+ globalBucketTargetSys.arnRemotesMap[target.arn] = arnTarget{Client: &TargetClient{Client: client, ARN: target.arn, Bucket: target.bucket}, lastRefresh: UTCNow()}
+ globalBucketTargetSys.targetsMap[bucket] = append(globalBucketTargetSys.targetsMap[bucket], madmin.BucketTarget{Arn: target.arn, TargetBucket: target.bucket})
+ globalBucketTargetSys.Unlock()
+ globalBucketTargetSys.hMutex.Lock()
+ globalBucketTargetSys.hc[client.EndpointURL().Host] = epHealth{Online: true}
+ globalBucketTargetSys.hMutex.Unlock()
+}
+
+func setILMTestReplicationConfig(t *testing.T, bucket, arn, targetBucket string) {
+ t.Helper()
+ cfg := replication.Config{}
+ rule := configs[0].Rules[0]
+ rule.Priority = 1
+ rule.Destination = replication.Destination{ARN: arn, Bucket: targetBucket}
+ cfg.Rules = append(cfg.Rules, rule)
+ cfg.RoleArn = arn
+ meta, err := globalBucketMetadataSys.Get(bucket)
+ if err != nil {
+ t.Fatal(err)
+ }
+ meta.replicationConfig = &cfg
+ globalBucketMetadataSys.Set(bucket, meta)
+}
+
+func newILMTestReplicationPool(ctx context.Context, obj ObjectLayer, stats *ReplicationStats) (restore func(), receive func() DeletedObjectReplicationInfo) {
+ oldPool := globalReplicationPool
+ p := &ReplicationPool{ctx: ctx, objLayer: obj, workers: []chan ReplicationWorkerOperation{make(chan ReplicationWorkerOperation, 8)}, stats: stats, mrfSaveCh: make(chan MRFReplicateEntry, 8)}
+ globalReplicationPool = once.NewSingleton[ReplicationPool]()
+ globalReplicationPool.Set(p)
+ return func() { globalReplicationPool = oldPool }, func() DeletedObjectReplicationInfo {
+ select {
+ case op := <-p.workers[0]:
+ d, ok := op.(DeletedObjectReplicationInfo)
+ if !ok {
+ return DeletedObjectReplicationInfo{}
+ }
+ return d
+ case <-time.After(3 * time.Second):
+ return DeletedObjectReplicationInfo{}
+ }
+ }
+}
+
+// TestReplicateILMExpiryDeleteMarker verifies that a delete marker created by
+// ILM expiry carries delete replication state and is queued for replication,
+// exactly like a marker created through the S3 DELETE API.
+func TestReplicateILMExpiryDeleteMarker(t *testing.T) {
+ defer DetectTestLeak(t)()
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t,
+ objAPITest: testReplicateILMExpiryDeleteMarker,
+ })
+}
+
+func testReplicateILMExpiryDeleteMarker(obj ObjectLayer, instanceType, bucket string, _ http.Handler, creds auth.Credentials, t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ stats := NewReplicationStats(ctx, nil)
+ oldStats := globalReplicationStats.Swap(stats)
+ defer globalReplicationStats.Store(oldStats)
+
+ const name = "ilm-object"
+ if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketVersioningConfig, enabledBucketVersioningConfig); err != nil {
+ t.Fatal(err)
+ }
+ // DeleteObject refuses ILM expiry (Expiration.Expire) without lifecycle
+ // rules; a date in the past makes every version eligible immediately.
+ ilmXML := []byte(`ilm-expireEnabled2020-01-01T00:00:00Z`)
+ if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketLifecycleConfig, ilmXML); err != nil {
+ t.Fatal(err)
+ }
+ oi, err := obj.PutObject(ctx, bucket, name, mustGetPutObjReader(t, strings.NewReader("data"), 4, "", ""), ObjectOptions{Versioned: true})
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ target := newILMTestTarget(t)
+ registerILMTestTarget(t, obj, bucket, target, func(w http.ResponseWriter, r *http.Request) {
+ topts := ObjectOptions{VersionID: r.URL.Query().Get("versionId"), Versioned: true}
+ switch r.Method {
+ case http.MethodHead:
+ roi, rerr := obj.GetObjectInfo(r.Context(), target.bucket, name, topts)
+ if roi.DeleteMarker {
+ w.Header().Set(xhttp.AmzDeleteMarker, "true")
+ w.Header().Set(xhttp.AmzVersionID, roi.VersionID)
+ }
+ if rerr != nil {
+ writeErrorResponseHeadersOnly(w, toAPIError(r.Context(), rerr))
+ return
+ }
+ w.WriteHeader(http.StatusOK)
+ case http.MethodDelete:
+ target.deletes.Add(1)
+ if topts.VersionID == "" {
+ t.Error("missing remote versionId")
+ }
+ topts.DeleteMarker = r.Header.Get(xhttp.MinIOSourceDeleteMarker) == "true"
+ topts.ReplicationRequest = true
+ topts.SetReplicaStatus(replication.Replica)
+ if _, err := obj.DeleteObject(r.Context(), target.bucket, name, topts); err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
+ writeErrorResponse(r.Context(), w, toAPIError(r.Context(), err), r.URL)
+ return
+ }
+ w.WriteHeader(http.StatusNoContent)
+ default:
+ t.Errorf("unexpected remote method %s", r.Method)
+ }
+ })
+ setILMTestReplicationConfig(t, bucket, target.arn, target.bucket)
+ restorePool, receive := newILMTestReplicationPool(ctx, obj, stats)
+ defer restorePool()
+
+ if ok := applyExpiryOnNonTransitionedObjects(ctx, obj, oi, lifecycle.Event{Action: lifecycle.DeleteAction}, lcEventSrc_Scanner); !ok {
+ t.Fatalf("%s: ILM expiry did not apply", instanceType)
+ }
+
+ deletion := receive()
+ if deletion.ObjectName == "" {
+ t.Fatalf("%s: ILM delete marker was not queued for replication", instanceType)
+ }
+ if !deletion.DeleteMarker || deletion.VersionID != "" || deletion.DeleteMarkerVersionID == "" {
+ t.Fatalf("%s: unexpected queued delete: %+v", instanceType, deletion)
+ }
+ if deletion.EventType != ReplicateILMDelete {
+ t.Fatalf("%s: queued event type = %q, want %q", instanceType, deletion.EventType, ReplicateILMDelete)
+ }
+
+ // The ILM-created marker must carry pending replication state.
+ markerVID := deletion.DeleteMarkerVersionID
+ moi, merr := obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{VersionID: markerVID, Versioned: true})
+ if !isErrMethodNotAllowed(merr) || !moi.DeleteMarker {
+ t.Fatalf("%s: ILM marker missing: %v %v", instanceType, moi, merr)
+ }
+ if moi.ReplicationStatus != replication.Pending {
+ t.Fatalf("%s: ILM marker replication status = %q, want PENDING", instanceType, moi.ReplicationStatus)
+ }
+
+ result := replicateDelete(ctx, deletion, obj)
+ if result.ReplicationStatus() != replication.Completed {
+ t.Fatalf("%s: marker replication result: %+v", instanceType, result)
+ }
+ if target.deletes.Load() != 1 {
+ t.Fatalf("%s: target saw %d deletes, want 1", instanceType, target.deletes.Load())
+ }
+ // The marker must exist on the target.
+ roi, rerr := obj.GetObjectInfo(ctx, target.bucket, name, ObjectOptions{VersionID: markerVID, Versioned: true})
+ if !isErrMethodNotAllowed(rerr) || !roi.DeleteMarker {
+ t.Fatalf("%s: target marker missing: %v %v", instanceType, roi, rerr)
+ }
+ // The source marker records the completed replication per target. The
+ // composite ReplicationStatus field is unreliable on read-back for
+ // markers carrying a replication timestamp, so assert the raw state.
+ moi, merr = obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{VersionID: markerVID})
+ if !isErrMethodNotAllowed(merr) || !strings.Contains(moi.ReplicationStatusInternal, "=COMPLETED;") {
+ t.Fatalf("%s: source marker state = %q (err %v), want COMPLETED", instanceType, moi.ReplicationStatusInternal, merr)
+ }
+}
+
+// TestReplicateILMExpiryNoncurrentPurge verifies that a noncurrent version
+// removed by ILM carries a pending version purge state, is queued for delete
+// replication and converges on the target.
+func TestReplicateILMExpiryNoncurrentPurge(t *testing.T) {
+ defer DetectTestLeak(t)()
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t,
+ objAPITest: testReplicateILMExpiryNoncurrentPurge,
+ })
+}
+
+func testReplicateILMExpiryNoncurrentPurge(obj ObjectLayer, instanceType, bucket string, _ http.Handler, creds auth.Credentials, t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ stats := NewReplicationStats(ctx, nil)
+ oldStats := globalReplicationStats.Swap(stats)
+ defer globalReplicationStats.Store(oldStats)
+
+ const name = "ilm-object"
+ if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketVersioningConfig, enabledBucketVersioningConfig); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := obj.PutObject(ctx, bucket, name, mustGetPutObjReader(t, strings.NewReader("v1"), 2, "", ""), ObjectOptions{Versioned: true}); err != nil {
+ t.Fatal(err)
+ }
+ v1Info, err := obj.PutObject(ctx, bucket, name, mustGetPutObjReader(t, strings.NewReader("v2"), 2, "", ""), ObjectOptions{Versioned: true})
+ if err != nil {
+ t.Fatal(err)
+ }
+ v1 := v1Info.VersionID
+ if v1 == "" {
+ t.Fatalf("%s: missing version ID on second version", instanceType)
+ }
+
+ target := newILMTestTarget(t)
+ registerILMTestTarget(t, obj, bucket, target, func(w http.ResponseWriter, r *http.Request) {
+ topts := ObjectOptions{VersionID: r.URL.Query().Get("versionId"), Versioned: true}
+ switch r.Method {
+ case http.MethodDelete:
+ target.deletes.Add(1)
+ if topts.VersionID == "" {
+ t.Error("missing remote versionId")
+ }
+ topts.DeleteMarker = r.Header.Get(xhttp.MinIOSourceDeleteMarker) == "true"
+ topts.ReplicationRequest = true
+ topts.SetReplicaStatus(replication.Replica)
+ if _, err := obj.DeleteObject(r.Context(), target.bucket, name, topts); err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
+ writeErrorResponse(r.Context(), w, toAPIError(r.Context(), err), r.URL)
+ return
+ }
+ w.WriteHeader(http.StatusNoContent)
+ default:
+ t.Errorf("unexpected remote method %s", r.Method)
+ }
+ })
+ // Pre-create a marker version on the target with the same version ID as
+ // the noncurrent version being purged, so the purge can be observed.
+ if _, err := obj.DeleteObject(ctx, target.bucket, name, ObjectOptions{VersionID: v1, Versioned: true, DeleteMarker: true, ReplicationRequest: true}); err != nil {
+ t.Fatal(err)
+ }
+ setILMTestReplicationConfig(t, bucket, target.arn, target.bucket)
+ restorePool, receive := newILMTestReplicationPool(ctx, obj, stats)
+ defer restorePool()
+
+ // The scanner evaluates noncurrent version expiry per version; build the
+ // delete entry exactly the way the scanner does for the oldest version.
+ vi, verr := obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{VersionID: v1, Versioned: true})
+ if verr != nil {
+ t.Fatalf("%s: missing noncurrent version: %v", instanceType, verr)
+ }
+ otd := ilmPurgeToDelete(ctx, bucket, vi, true)
+ if otd.VersionPurgeStatus != replication.VersionPurgePending || otd.VersionPurgeStatuses == "" || otd.ReplicateDecisionStr == "" {
+ t.Fatalf("%s: ILM purge entry carries no replication state: %+v", instanceType, otd)
+ }
+ if plain := ilmPurgeToDelete(ctx, bucket, vi, false); plain.VersionPurgeStatus != "" || plain.ReplicateDecisionStr != "" {
+ t.Fatalf("%s: non-replicating purge entry carries state: %+v", instanceType, plain)
+ }
+
+ deleteObjectVersions(ctx, obj, bucket, []ObjectToDelete{otd}, []lifecycle.Event{{Action: lifecycle.DeleteVersionAction}})
+
+ // The version must remain locally as a tombstone with pending purge
+ // state until replication completes.
+ pi, perr := obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{VersionID: v1, Versioned: true})
+ if !isErrMethodNotAllowed(perr) {
+ t.Fatalf("%s: purged version metadata lost: %v", instanceType, perr)
+ }
+ if pi.VersionPurgeStatus != replication.VersionPurgePending {
+ t.Fatalf("%s: purged version state = %q, want PENDING", instanceType, pi.VersionPurgeStatus)
+ }
+
+ deletion := receive()
+ if deletion.ObjectName == "" {
+ t.Fatalf("%s: ILM noncurrent purge was not queued for replication", instanceType)
+ }
+ if deletion.VersionID != v1 || deletion.DeleteMarkerVersionID != "" || deletion.DeleteMarker {
+ t.Fatalf("%s: unexpected queued purge: %+v", instanceType, deletion)
+ }
+
+ result := replicateDelete(ctx, deletion, obj)
+ if result.VersionPurgeStatus() != replication.VersionPurgeComplete {
+ t.Fatalf("%s: purge replication result: %+v", instanceType, result)
+ }
+ if target.deletes.Load() != 1 {
+ t.Fatalf("%s: target saw %d deletes, want 1", instanceType, target.deletes.Load())
+ }
+ // The purge must have converged on the target.
+ if _, terr := obj.GetObjectInfo(ctx, target.bucket, name, ObjectOptions{VersionID: v1, Versioned: true}); !isErrVersionNotFound(terr) && !isErrObjectNotFound(terr) {
+ t.Fatalf("%s: target version still present: %v", instanceType, terr)
+ }
+ // Once the purge completes the local tombstone is freed and the version
+ // is gone on both sides.
+ if _, terr := obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{VersionID: v1}); !isErrVersionNotFound(terr) && !isErrObjectNotFound(terr) {
+ t.Fatalf("%s: purged version still present locally: %v", instanceType, terr)
+ }
+}