fix: DeleteObject() API with versionId under replication (#16325)

cmd/bucket-replication.go

@@ -309,7 +309,7 @@ func checkReplicateDelete(ctx context.Context, bucket string, dobj ObjectToDelet
 		for _, tgtArn := range tgtArns {
 			opts.TargetArn = tgtArn
 			replicate = rcfg.Replicate(opts)
-			// when incoming delete is removal of a delete marker( a.k.a versioned delete),
+			// when incoming delete is removal of a delete marker(a.k.a versioned delete),
 			// GetObjectInfo returns extra information even though it returns errFileNotFound
 			if gerr != nil {
 				validReplStatus := false

cmd/erasure-metadata.go

@@ -147,9 +147,9 @@ func (fi FileInfo) ToObjectInfo(bucket, object string, versioned bool) ObjectInf
 	// Add replication status to the object info
 	objInfo.ReplicationStatusInternal = fi.ReplicationState.ReplicationStatusInternal
 	objInfo.VersionPurgeStatusInternal = fi.ReplicationState.VersionPurgeStatusInternal
-	objInfo.ReplicationStatus = fi.ReplicationState.CompositeReplicationStatus()
+	objInfo.ReplicationStatus = fi.ReplicationStatus()
+	objInfo.VersionPurgeStatus = fi.VersionPurgeStatus()
 
-	objInfo.VersionPurgeStatus = fi.ReplicationState.CompositeVersionPurgeStatus()
 	objInfo.TransitionedObject = TransitionedObject{
 		Name:        fi.TransitionedObjName,
 		VersionID:   fi.TransitionVersionID,
@@ -176,7 +176,6 @@ func (fi FileInfo) ToObjectInfo(bucket, object string, versioned bool) ObjectInf
 		objInfo.StorageClass = globalMinioDefaultStorageClass
 	}
 
-	objInfo.VersionPurgeStatus = fi.VersionPurgeStatus()
 	// set restore status for transitioned object
 	restoreHdr, ok := fi.Metadata[xhttp.AmzRestore]
 	if ok {
@@ -534,6 +533,11 @@ func (fi *FileInfo) VersionPurgeStatus() VersionPurgeStatusType {
 	return fi.ReplicationState.CompositeVersionPurgeStatus()
 }
 
+// ReplicationStatus returns overall version replication status for this object version across targets
+func (fi *FileInfo) ReplicationStatus() replication.StatusType {
+	return fi.ReplicationState.CompositeReplicationStatus()
+}
+
 // DeleteMarkerReplicationStatus returns overall replication status for this delete marker version across targets
 func (fi *FileInfo) DeleteMarkerReplicationStatus() replication.StatusType {
 	if fi.Deleted {

cmd/erasure-object.go

@@ -1604,11 +1604,9 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string
 	markDelete := goi.VersionID != ""
 
 	// Default deleteMarker to true if object is under versioning
-	// versioning suspended means we add `null` version as
-	// delete marker, if its not decided already.
-	deleteMarker := (opts.Versioned || opts.VersionSuspended) && opts.VersionID == "" || (opts.DeleteMarker && opts.VersionID != "")
+	deleteMarker := opts.Versioned
 
-	if markDelete && opts.VersionID != "" {
+	if opts.VersionID != "" {
 		// case where replica version needs to be deleted on target cluster
 		if versionFound && opts.DeleteMarkerReplicationStatus() == replication.Replica {
 			markDelete = false
@@ -1619,6 +1617,22 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string
 		if opts.VersionPurgeStatus() == Complete {
 			markDelete = false
 		}
+		// now, since VersionPurgeStatus() is already set, we can let the
+		// lower layers decide this. This fixes a regression that was introduced
+		// in PR #14555 where !VersionPurgeStatus.Empty() is automatically
+		// considered as Delete marker true to avoid listing such objects by
+		// regular ListObjects() calls. However for delete replication this
+		// ends up being a problem because "upon" a successful delete this
+		// ends up creating a new delete marker that is spurious and unnecessary.
+		//
+		// Regression introduced by #14555 was reintroduced in #15564
+		if versionFound {
+			if !goi.VersionPurgeStatus.Empty() {
+				deleteMarker = false
+			} else if !goi.DeleteMarker { // implies a versioned delete of object
+				deleteMarker = false
+			}
+		}
 	}
 
 	modTime := opts.MTime
@@ -1628,9 +1642,14 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string
 	fvID := mustGetUUID()
 
 	if markDelete && (opts.Versioned || opts.VersionSuspended) {
+		if !deleteMarker {
+			// versioning suspended means we add `null` version as
+			// delete marker, if its not decided already.
+			deleteMarker = opts.VersionSuspended && opts.VersionID == ""
+		}
 		fi := FileInfo{
 			Name:             object,
-			Deleted:          true,
+			Deleted:          deleteMarker,
 			MarkDeleted:      markDelete,
 			ModTime:          modTime,
 			ReplicationState: opts.DeleteReplication,