// specify a versionID.
		objToDel := ObjectToDelete{
			ObjectV: ObjectV{
				ObjectName: decodeDirObject(dObjects[i].ObjectName),
				VersionID:  dObjects[i].VersionID,
			},
			VersionPurgeStatus:            dObjects[i].VersionPurgeStatus(),
			VersionPurgeStatuses:          dObjects[i].ReplicationState.VersionPurgeStatusInternal,
			DeleteMarkerReplicationStatus: dObjects[i].ReplicationState.ReplicationStatusInternal,

		RestoreOngoing:     oi.RestoreOngoing,
		RestoreExpires:     oi.RestoreExpires,
		TransitionStatus:   oi.TransitionedObject.Status,
		UserDefined:        oi.UserDefined,
		VersionPurgeStatus: oi.VersionPurgeStatus,
		ReplicationStatus:  oi.ReplicationStatus,
	}

			markDelete = false
		}
		if opts.VersionPurgeStatus().Empty() && opts.DeleteMarkerReplicationStatus().Empty() {
			markDelete = false
		}
		if opts.VersionPurgeStatus() == replication.VersionPurgeComplete {
			markDelete = false
		}
		// now, since VersionPurgeStatus() is already set, we can let the
		// lower layers decide this. This fixes a regression that was introduced

		}
	}
	updateVersion := false
	if fi.VersionPurgeStatus().Empty() && (fi.DeleteMarkerReplicationStatus() == "REPLICA" || fi.DeleteMarkerReplicationStatus().Empty()) {
		updateVersion = fi.MarkDeleted
	} else {
		// for replication scenario
		if fi.Deleted && fi.VersionPurgeStatus() != replication.VersionPurgeComplete {
			if !fi.VersionPurgeStatus().Empty() || fi.DeleteMarkerReplicationStatus().Empty() {

func (e *Evaluator) IsPendingReplication(obj ObjectOpts) bool {
	if e.replCfg == nil {
		return false
	}
	if e.replCfg.HasActiveRules(obj.Name, true) && !obj.VersionPurgeStatus.Empty() {
		return true
	}

	return false
}

// IsObjectLocked checks if it is appropriate to remove an
// object according to locking configuration when this is lifecycle/ bucket quota asking.

	if gerr != nil { // error from GetObjectInfo
		if _, ok := gerr.(MethodNotAllowed); ok {
			// This happens usually for a delete marker
			if oi.DeleteMarker || !oi.VersionPurgeStatus.Empty() {
				// Delete marker should be present and valid.
				return nil
			}
		}
		if isErrObjectNotFound(gerr) || isErrVersionNotFound(gerr) {
			return nil
		}
		return gerr
	}

				if vidMarkerIdx >= 0 {
					fiVersions = fiVersions[vidMarkerIdx+1:]
				}
				afterV = ""
			}

			for _, version := range fiVersions {
				if !version.VersionPurgeStatus().Empty() {
					continue
				}
				versioned := vcfg != nil && vcfg.Versioned(entry.name)
				versions = append(versions, version.ToObjectInfo(bucket, entry.name, versioned))
			}

			continue
		}

	tgtPrefix := r.Target.Prefix
	srcObject := srcObjInfo.Name
	s3Type := r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3

	if srcObjInfo.DeleteMarker || !srcObjInfo.VersionPurgeStatus.Empty() {
		if retry && !s3Type {
			if _, err := c.StatObject(ctx, tgtBucket, pathJoin(tgtPrefix, srcObject), minio.StatObjectOptions{
				VersionID: srcObjInfo.VersionID,
				Internal: minio.AdvancedGetOptions{

func (r *BatchJobKeyRotateV1) KeyRotate(ctx context.Context, api ObjectLayer, objInfo ObjectInfo) error {
	srcBucket := r.Bucket
	srcObject := objInfo.Name

	if objInfo.DeleteMarker || !objInfo.VersionPurgeStatus.Empty() {
		return nil
	}
	sseKMS := crypto.S3KMS.IsEncrypted(objInfo.UserDefined)
	sseS3 := crypto.S3.IsEncrypted(objInfo.UserDefined)
	if !sseKMS && !sseS3 { // neither sse-s3 nor sse-kms disallowed

				writeErrorResponse(ctx, w, toAPIError(ctx, proxy.Err), r.URL)
				return
			}
			if gr != nil {
				if !gr.ObjInfo.VersionPurgeStatus.Empty() {
					// Shows the replication status of a permanent delete of a version
					w.Header()[xhttp.MinIODeleteReplicationStatus] = []string{string(gr.ObjInfo.VersionPurgeStatus)}
				}
				if !gr.ObjInfo.ReplicationStatus.Empty() && gr.ObjInfo.DeleteMarker {