fix(pools): preserve ILM cleanup during migration

Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
Feng Ruohang
2026-11-02 14:20:17 +08:00
parent 8dac7f7c14
commit 0a546c7a76
9 changed files with 811 additions and 155 deletions
+66 -7
View File
@@ -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)