From fb6ab1cca20795f5a4798a71c394bbfd4562afcd Mon Sep 17 00:00:00 2001
From: Poorna <poornas@users.noreply.github.com>
Date: Wed, 8 Mar 2023 07:03:29 -0800
Subject: [PATCH] fix: allow replication of 'null' delete markers  (#16773)

---
 cmd/bucket-replication.go  | 12 +++++++-----
 cmd/erasure-server-pool.go | 11 ++++++++---
 2 files changed, 15 insertions(+), 8 deletions(-)

diff --git a/cmd/bucket-replication.go b/cmd/bucket-replication.go
index 24c7547fd..04f48374f 100644
--- a/cmd/bucket-replication.go
+++ b/cmd/bucket-replication.go
@@ -586,11 +586,13 @@ func replicateDeleteToTarget(ctx context.Context, dobj DeletedObjectReplicationI
 				return
 			}
 		}
-		// mark delete marker replication as failed if target cluster not ready to receive
-		// this request yet (object version not replicated yet)
-		if err != nil && !toi.ReplicationReady {
-			rinfo.ReplicationStatus = replication.Failed
-			return
+		if !isErrObjectNotFound(ErrorRespToObjectError(err, dobj.Bucket, dobj.ObjectName)) {
+			// mark delete marker replication as failed if target cluster not ready to receive
+			// this request yet (object version not replicated yet)
+			if err != nil && !toi.ReplicationReady {
+				rinfo.ReplicationStatus = replication.Failed
+				return
+			}
 		}
 	}
 
diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go
index eabe2e977..d8f5dfbbc 100644
--- a/cmd/erasure-server-pool.go
+++ b/cmd/erasure-server-pool.go
@@ -421,6 +421,7 @@ func (z *erasureServerPools) getPoolInfoExistingWithOpts(ctx context.Context, bu
 		return mtime1.After(mtime2)
 	})
 
+	defPool := PoolObjInfo{Index: -1}
 	for _, pinfo := range poolObjInfos {
 		// skip all objects from suspended pools if asked by the
 		// caller.
@@ -435,14 +436,13 @@ func (z *erasureServerPools) getPoolInfoExistingWithOpts(ctx context.Context, bu
 		if pinfo.Err != nil && !isErrObjectNotFound(pinfo.Err) {
 			return pinfo, pinfo.Err
 		}
-
+		defPool = pinfo
 		if isErrObjectNotFound(pinfo.Err) {
 			// No object exists or its a delete marker,
 			// check objInfo to confirm.
 			if pinfo.ObjInfo.DeleteMarker && pinfo.ObjInfo.Name != "" {
 				return pinfo, nil
 			}
-
 			// objInfo is not valid, truly the object doesn't
 			// exist proceed to next pool.
 			continue
@@ -450,7 +450,12 @@ func (z *erasureServerPools) getPoolInfoExistingWithOpts(ctx context.Context, bu
 
 		return pinfo, nil
 	}
-
+	if opts.ReplicationRequest && opts.DeleteMarker && defPool.Index >= 0 {
+		// If the request is a delete marker replication request, return a default pool
+		// in cases where the object does not exist.
+		// This is to ensure that the delete marker is replicated to the destination.
+		return defPool, nil
+	}
 	return PoolObjInfo{}, toObjectErr(errFileNotFound, bucket, object)
 }