From 90112b56446a402bf6f3ab4e67e5677b8f62eff9 Mon Sep 17 00:00:00 2001
From: Poorna Krishnamoorthy <poornas@users.noreply.github.com>
Date: Thu, 29 Apr 2021 16:46:26 -0700
Subject: [PATCH] Update ReplicationStatus if metadata not updated correctly
 (#12191)

There can be situations where replication completed but the
`X-Amz-Replication-Status` metadata update failed such as
when the server returns 503 under high load. This object version will
continue to be picked up by the scanner and replicateObject would perform
no action since the versions match between source and target.
The metadata would never reflect that replication was successful
without this fix, leading to repeated re-queuing.
---
 cmd/bucket-replication.go | 27 +++++++++++++++++++++++++++
 1 file changed, 27 insertions(+)

diff --git a/cmd/bucket-replication.go b/cmd/bucket-replication.go
index 51266b950..4b1936d6f 100644
--- a/cmd/bucket-replication.go
+++ b/cmd/bucket-replication.go
@@ -646,6 +646,33 @@ 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 metadata is not updated for some reason after replication, such as 503 encountered while updating metadata - make sure
+				// to set ReplicationStatus as Completed.Note that replication Stats would have been updated despite metadata update failure.
+				z, ok := objectAPI.(*erasureServerPools)
+				if !ok {
+					return
+				}
+				// This lower level implementation is necessary to avoid write locks from CopyObject.
+				poolIdx, err := z.getPoolIdx(ctx, bucket, object, objInfo.Size)
+				if err != nil {
+					logger.LogIf(ctx, fmt.Errorf("Unable to update replication metadata for %s/%s(%s): %w", bucket, objInfo.Name, objInfo.VersionID, err))
+				} else {
+					fi := FileInfo{}
+					fi.VersionID = objInfo.VersionID
+					fi.Metadata = make(map[string]string, len(objInfo.UserDefined))
+					for k, v := range objInfo.UserDefined {
+						fi.Metadata[k] = v
+					}
+					fi.Metadata[xhttp.AmzBucketReplicationStatus] = replication.Completed.String()
+					if objInfo.UserTags != "" {
+						fi.Metadata[xhttp.AmzObjectTagging] = objInfo.UserTags
+					}
+					if err = z.serverPools[poolIdx].getHashedSet(object).updateObjectMeta(ctx, bucket, object, fi); err != nil {
+						logger.LogIf(ctx, fmt.Errorf("Unable to update replication metadata for %s/%s(%s): %w", bucket, objInfo.Name, objInfo.VersionID, err))
+					}
+				}
+			}
 			return
 		}
 	}