Show Delete replication status header (#10946)

X-Minio-Replication-Delete-Status header shows the
status of the replication of a permanent delete of a version.

All GETs are disallowed and return 405 on this object version.
In the case of replicating delete markers.

X-Minio-Replication-DeleteMarker-Status shows the status 
of replication, and would similarly return 405.

Additionally, this PR adds reporting of delete marker event completion
and updates documentation

cmd/bucket-replication.go

@@ -222,21 +222,26 @@ func replicateDelete(ctx context.Context, dobj DeletedObjectVersionInfo, objectA
 			versionPurgeStatus = Complete
 		}
 	}
+	var eventName = event.ObjectReplicationComplete
 	if replicationStatus == string(replication.Failed) || versionPurgeStatus == Failed {
-		objInfo := ObjectInfo{
-			Name:              dobj.ObjectName,
-			DeleteMarker:      dobj.DeleteMarker,
-			VersionID:         versionID,
-			ReplicationStatus: replication.StatusType(dobj.DeleteMarkerReplicationStatus),
-		}
-		eventArg := &eventArgs{
-			BucketName: bucket,
-			Object:     objInfo,
-			Host:       "Internal: [Replication]",
-			EventName:  event.ObjectReplicationFailed,
-		}
-		sendEvent(*eventArg)
+		eventName = event.ObjectReplicationFailed
 	}
+	objInfo := ObjectInfo{
+		Name:               dobj.ObjectName,
+		DeleteMarker:       dobj.DeleteMarker,
+		VersionID:          versionID,
+		ReplicationStatus:  replication.StatusType(dobj.DeleteMarkerReplicationStatus),
+		VersionPurgeStatus: versionPurgeStatus,
+	}
+
+	eventArg := &eventArgs{
+		BucketName: bucket,
+		Object:     objInfo,
+		Host:       "Internal: [Replication]",
+		EventName:  eventName,
+	}
+	sendEvent(*eventArg)
+
 	// Update metadata on the delete marker or purge permanent delete if replication success.
 	if _, err = objectAPI.DeleteObject(ctx, bucket, dobj.ObjectName, ObjectOptions{
 		VersionID:                     versionID,

cmd/erasure-object.go

@@ -415,23 +415,26 @@ func (er erasureObjects) getObjectInfo(ctx context.Context, bucket, object strin
 	if err != nil {
 		return objInfo, toObjectErr(err, bucket, object)
 	}
-
+	objInfo = fi.ToObjectInfo(bucket, object)
+	if objInfo.TransitionStatus == lifecycle.TransitionComplete {
+		// overlay storage class for transitioned objects with transition tier SC Label
+		if sc := transitionSC(ctx, bucket); sc != "" {
+			objInfo.StorageClass = sc
+		}
+	}
+	if !fi.VersionPurgeStatus.Empty() {
+		// Make sure to return object info to provide extra information.
+		return objInfo, toObjectErr(errMethodNotAllowed, bucket, object)
+	}
 	if fi.Deleted {
-		objInfo = fi.ToObjectInfo(bucket, object)
 		if opts.VersionID == "" || opts.DeleteMarker {
 			return objInfo, toObjectErr(errFileNotFound, bucket, object)
 		}
 		// Make sure to return object info to provide extra information.
 		return objInfo, toObjectErr(errMethodNotAllowed, bucket, object)
 	}
-	oi := fi.ToObjectInfo(bucket, object)
-	if oi.TransitionStatus == lifecycle.TransitionComplete {
-		// overlay storage class for transitioned objects with transition tier SC Label
-		if sc := transitionSC(ctx, bucket); sc != "" {
-			oi.StorageClass = sc
-		}
-	}
-	return oi, nil
+
+	return objInfo, nil
 }
 
 func undoRename(disks []StorageAPI, srcBucket, srcEntry, dstBucket, dstEntry string, isDir bool, errs []error) {

cmd/http/headers.go

@@ -144,6 +144,11 @@ const (
 
 	// Header indicates if the delete marker version needs to be purged.
 	MinIOSourceDeleteMarkerDelete = "x-minio-source-deletemarker-delete"
+
+	// Header indicates permanent delete replication status.
+	MinIODeleteReplicationStatus = "X-Minio-Replication-Delete-Status"
+	// Header indicates delete-marker replication status.
+	MinIODeleteMarkerReplicationStatus = "X-Minio-Replication-DeleteMarker-Status"
 )
 
 // Common http query params S3 API

cmd/metacache-entries.go

@@ -151,7 +151,7 @@ func (e *metaCacheEntry) fileInfoVersions(bucket string) (FileInfoVersions, erro
 			},
 		}, nil
 	}
-	return getFileInfoVersions(e.metadata, bucket, e.name, false)
+	return getFileInfoVersions(e.metadata, bucket, e.name)
 }
 
 // metaCacheEntries is a slice of metacache entries.

cmd/object-handlers.go

@@ -575,6 +575,13 @@ func (api objectAPIHandlers) HeadObjectHandler(w http.ResponseWriter, r *http.Re
 	objInfo, err := getObjectInfo(ctx, bucket, object, opts)
 	if err != nil {
 		if globalBucketVersioningSys.Enabled(bucket) {
+			if !objInfo.VersionPurgeStatus.Empty() {
+				// Shows the replication status of a permanent delete of a version
+				w.Header()[xhttp.MinIODeleteReplicationStatus] = []string{string(objInfo.VersionPurgeStatus)}
+			}
+			if !objInfo.ReplicationStatus.Empty() && objInfo.DeleteMarker {
+				w.Header()[xhttp.MinIODeleteMarkerReplicationStatus] = []string{string(objInfo.ReplicationStatus)}
+			}
 			// Versioning enabled quite possibly object is deleted might be delete-marker
 			// if present set the headers, no idea why AWS S3 sets these headers.
 			if objInfo.VersionID != "" && objInfo.DeleteMarker {

cmd/xl-storage-format-utils.go

@@ -34,13 +34,13 @@ func (v versionsSorter) Less(i, j int) bool {
 	return v[i].ModTime.After(v[j].ModTime)
 }
 
-func getFileInfoVersions(xlMetaBuf []byte, volume, path string, showPendingDeletes bool) (FileInfoVersions, error) {
+func getFileInfoVersions(xlMetaBuf []byte, volume, path string) (FileInfoVersions, error) {
 	if isXL2V1Format(xlMetaBuf) {
 		var xlMeta xlMetaV2
 		if err := xlMeta.Load(xlMetaBuf); err != nil {
 			return FileInfoVersions{}, err
 		}
-		versions, latestModTime, err := xlMeta.ListVersions(volume, path, showPendingDeletes)
+		versions, latestModTime, err := xlMeta.ListVersions(volume, path)
 		if err != nil {
 			return FileInfoVersions{}, err
 		}

cmd/xl-storage-format-v2.go

@@ -598,7 +598,7 @@ func (z xlMetaV2) TotalSize() int64 {
 // versions returns error for unexpected entries.
 // showPendingDeletes is set to true if ListVersions needs to list objects marked deleted
 // but waiting to be replicated
-func (z xlMetaV2) ListVersions(volume, path string, showPendingDeletes bool) (versions []FileInfo, modTime time.Time, err error) {
+func (z xlMetaV2) ListVersions(volume, path string) (versions []FileInfo, modTime time.Time, err error) {
 	var latestModTime time.Time
 	var latestVersionID string
 	for _, version := range z.Versions {
@@ -611,9 +611,6 @@ func (z xlMetaV2) ListVersions(volume, path string, showPendingDeletes bool) (ve
 			fi, err = version.ObjectV2.ToFileInfo(volume, path)
 		case DeleteType:
 			fi, err = version.DeleteMarker.ToFileInfo(volume, path)
-			if !fi.VersionPurgeStatus.Empty() && !showPendingDeletes {
-				continue
-			}
 		case LegacyType:
 			fi, err = version.ObjectV1.ToFileInfo(volume, path)
 		}

cmd/xl-storage.go

@@ -380,7 +380,7 @@ func (s *xlStorage) CrawlAndGetDataUsage(ctx context.Context, cache dataUsageCac
 		// Remove filename which is the meta file.
 		item.transformMetaDir()
 
-		fivs, err := getFileInfoVersions(buf, item.bucket, item.objectPath(), true)
+		fivs, err := getFileInfoVersions(buf, item.bucket, item.objectPath())
 		if err != nil {
 			return 0, errSkipFile
 		}
@@ -898,7 +898,7 @@ func (s *xlStorage) WalkVersions(ctx context.Context, volume, dirPath, marker st
 					continue
 				}
 
-				fiv, err = getFileInfoVersions(xlMetaBuf, volume, walkResult.entry, false)
+				fiv, err = getFileInfoVersions(xlMetaBuf, volume, walkResult.entry)
 				if err != nil {
 					continue
 				}