From 674c6f7a7b5cc00b0ac83fa141a63b28c92d0fdc Mon Sep 17 00:00:00 2001
From: Poorna Krishnamoorthy <poornas@users.noreply.github.com>
Date: Mon, 23 Aug 2021 17:48:22 -0400
Subject: [PATCH] fix: resync of replication of delete markers (#12932)

Fixes #12919
---
 cmd/bucket-replication.go | 16 +++++++++++++++-
 cmd/data-scanner.go       |  2 +-
 cmd/erasure-object.go     |  2 +-
 cmd/gateway-common.go     |  4 ++++
 cmd/object-api-errors.go  |  6 ++++++
 5 files changed, 27 insertions(+), 3 deletions(-)

diff --git a/cmd/bucket-replication.go b/cmd/bucket-replication.go
index a50019bb3..b0c1dc818 100644
--- a/cmd/bucket-replication.go
+++ b/cmd/bucket-replication.go
@@ -337,6 +337,17 @@ func replicateDelete(ctx context.Context, dobj DeletedObjectReplicationInfo, obj
 	}
 	ctx = lkctx.Context()
 	defer lk.Unlock(lkctx.Cancel)
+	// early return if already replicated delete marker for existing object replication
+	if dobj.DeleteMarkerVersionID != "" && dobj.OpType == replication.ExistingObjectReplicationType {
+		_, err := tgt.StatObject(ctx, rcfg.GetDestination().Bucket, dobj.ObjectName, miniogo.StatObjectOptions{
+			VersionID: versionID,
+			Internal: miniogo.AdvancedGetOptions{
+				ReplicationProxyRequest: "false",
+			}})
+		if isErrMethodNotAllowed(ErrorRespToObjectError(err, dobj.Bucket, dobj.ObjectName)) {
+			return
+		}
+	}
 
 	rmErr := tgt.RemoveObject(ctx, rcfg.GetDestination().Bucket, dobj.ObjectName, miniogo.RemoveObjectOptions{
 		VersionID: versionID,
@@ -766,7 +777,7 @@ func replicateObject(ctx context.Context, ri ReplicateObjectInfo, objectAPI Obje
 		if rtype == replicateNone {
 			// object with same VersionID already exists, replication kicked off by
 			// PutObject might have completed
-			if objInfo.ReplicationStatus == replication.Pending || objInfo.ReplicationStatus == replication.Failed {
+			if objInfo.ReplicationStatus == replication.Pending || objInfo.ReplicationStatus == replication.Failed || ri.OpType == replication.ExistingObjectReplicationType {
 				// if metadata is not updated for some reason after replication, such as
 				// 503 encountered while updating metadata - make sure to set ReplicationStatus
 				// as Completed.
@@ -782,6 +793,9 @@ func replicateObject(ctx context.Context, ri ReplicateObjectInfo, objectAPI Obje
 				for k, v := range objInfo.UserDefined {
 					popts.UserDefined[k] = v
 				}
+				if ri.OpType == replication.ExistingObjectReplicationType {
+					popts.UserDefined[xhttp.MinIOReplicationResetStatus] = fmt.Sprintf("%s;%s", UTCNow().Format(http.TimeFormat), ri.ResetID)
+				}
 				popts.UserDefined[xhttp.AmzBucketReplicationStatus] = replication.Completed.String()
 				if objInfo.UserTags != "" {
 					popts.UserDefined[xhttp.AmzObjectTagging] = objInfo.UserTags
diff --git a/cmd/data-scanner.go b/cmd/data-scanner.go
index a651a54f4..a579d1947 100644
--- a/cmd/data-scanner.go
+++ b/cmd/data-scanner.go
@@ -1173,7 +1173,7 @@ func (i *scannerItem) healReplication(ctx context.Context, o ObjectLayer, oi Obj
 			return
 		}
 		// if replication status is Complete on DeleteMarker and existing object resync required
-		if existingObjResync && oi.ReplicationStatus == replication.Completed {
+		if existingObjResync && (oi.ReplicationStatus == replication.Completed) {
 			i.healReplicationDeletes(ctx, o, oi, existingObjResync)
 			return
 		}
diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go
index 6ee107cb6..6d06b91da 100644
--- a/cmd/erasure-object.go
+++ b/cmd/erasure-object.go
@@ -441,7 +441,7 @@ func (er erasureObjects) getObjectInfo(ctx context.Context, bucket, object strin
 
 	}
 	objInfo = fi.ToObjectInfo(bucket, object)
-	if !fi.VersionPurgeStatus.Empty() {
+	if !fi.VersionPurgeStatus.Empty() && opts.VersionID != "" {
 		// Make sure to return object info to provide extra information.
 		return objInfo, toObjectErr(errMethodNotAllowed, bucket, object)
 	}
diff --git a/cmd/gateway-common.go b/cmd/gateway-common.go
index 1c74c29ff..d10bda345 100644
--- a/cmd/gateway-common.go
+++ b/cmd/gateway-common.go
@@ -333,6 +333,10 @@ func ErrorRespToObjectError(err error, params ...string) error {
 		err = PartTooSmall{}
 	}
 
+	switch minioErr.StatusCode {
+	case http.StatusMethodNotAllowed:
+		err = toObjectErr(errMethodNotAllowed, bucket, object)
+	}
 	return err
 }
 
diff --git a/cmd/object-api-errors.go b/cmd/object-api-errors.go
index effca57c0..4a2f7a539 100644
--- a/cmd/object-api-errors.go
+++ b/cmd/object-api-errors.go
@@ -686,3 +686,9 @@ func isErrPreconditionFailed(err error) bool {
 	_, ok := err.(PreConditionFailed)
 	return ok
 }
+
+// isErrMethodNotAllowed - Check if error type is MethodNotAllowed.
+func isErrMethodNotAllowed(err error) bool {
+	var methodNotAllowed MethodNotAllowed
+	return errors.As(err, &methodNotAllowed)
+}