Avoid extra GetObjectInfo call in DeleteObject API (#17599)

Optimize DeleteObject API to avoid extra 
GetObjectInfo call on the replicating side.

For receiving side, it is just a regular
DeleteObject call.

Bonus: Fix a corner case where version purged is 
absent on target (either due to replication not yet
complete or target version already deleted in a
one-way replication or when replication was disabled). 

In such cases, mark version purge complete.

cmd/object-handlers.go

@@ -2326,33 +2326,29 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 		return
 	}
 
-	getObjectInfo := objectAPI.GetObjectInfo
-	if api.CacheAPI() != nil {
-		getObjectInfo = api.CacheAPI().GetObjectInfo
-	}
-
 	os := newObjSweeper(bucket, object).WithVersion(opts.VersionID).WithVersioning(opts.Versioned, opts.VersionSuspended)
-	// Mutations of objects on versioning suspended buckets
-	// affect its null version. Through opts below we select
-	// the null version's remote object to delete if
-	// transitioned.
-	goiOpts := os.GetOpts()
-	goi, gerr := getObjectInfo(ctx, bucket, object, goiOpts)
-	if gerr == nil {
-		os.SetTransitionState(goi.TransitionedObject)
-	}
 
-	dsc := checkReplicateDelete(ctx, bucket, ObjectToDelete{
-		ObjectV: ObjectV{
-			ObjectName: object,
-			VersionID:  opts.VersionID,
-		},
-	}, goi, opts, gerr)
+	opts.SetEvalMetadataFn(func(oi *ObjectInfo, gerr error) (dsc ReplicateDecision, err error) {
+		if replica { // no need to check replication on receiver
+			return dsc, nil
+		}
+		dsc = checkReplicateDelete(ctx, bucket, ObjectToDelete{
+			ObjectV: ObjectV{
+				ObjectName: object,
+				VersionID:  opts.VersionID,
+			},
+		}, *oi, opts, gerr)
+		// Mutations of objects on versioning suspended buckets
+		// affect its null version. Through opts below we select
+		// the null version's remote object to delete if
+		// transitioned.
+		if gerr == nil {
+			os.SetTransitionState(oi.TransitionedObject)
+		}
+		return dsc, nil
+	})
 
 	vID := opts.VersionID
-	if dsc.ReplicateAny() {
-		opts.SetDeleteReplicationState(dsc, opts.VersionID)
-	}
 	if replica {
 		opts.SetReplicaStatus(replication.Replica)
 		if opts.VersionPurgeStatus().Empty() {
@@ -2360,25 +2356,21 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 			vID = ""
 		}
 	}
-
-	apiErr := ErrNone
-	if vID != "" {
-		apiErr = enforceRetentionBypassForDelete(ctx, r, bucket, ObjectToDelete{
-			ObjectV: ObjectV{
-				ObjectName: object,
-				VersionID:  vID,
-			},
-		}, goi, gerr)
-		if apiErr != ErrNone && apiErr != ErrNoSuchKey {
-			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(apiErr), r.URL)
-			return
+	opts.SetEvalRetentionBypassFn(func(goi ObjectInfo, gerr error) (err error) {
+		err = nil
+		if vID != "" {
+			err := enforceRetentionBypassForDelete(ctx, r, bucket, ObjectToDelete{
+				ObjectV: ObjectV{
+					ObjectName: object,
+					VersionID:  vID,
+				},
+			}, goi, gerr)
+			if err != nil && !isErrObjectNotFound(err) {
+				return err
+			}
 		}
-	}
-
-	if apiErr == ErrNoSuchKey {
-		writeSuccessNoContent(w)
 		return
-	}
+	})
 
 	deleteObject := objectAPI.DeleteObject
 	if api.CacheAPI() != nil {
@@ -2393,6 +2385,12 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
 			return
 		}
+		if isErrObjectNotFound(err) {
+			writeSuccessNoContent(w)
+			return
+		}
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
 	}
 
 	if objInfo.Name == "" {
@@ -2419,7 +2417,7 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 		Host:         handlers.GetSourceIP(r),
 	})
 
-	if dsc.ReplicateAny() {
+	if objInfo.ReplicationStatus == replication.Pending || objInfo.VersionPurgeStatus == Pending {
 		dmVersionID := ""
 		versionID := ""
 		if objInfo.DeleteMarker {
@@ -2434,7 +2432,7 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 				DeleteMarkerVersionID: dmVersionID,
 				DeleteMarkerMTime:     DeleteMarkerMTime{objInfo.ModTime},
 				DeleteMarker:          objInfo.DeleteMarker,
-				ReplicationState:      objInfo.getReplicationState(dsc.String(), opts.VersionID, false),
+				ReplicationState:      objInfo.getReplicationState(),
 			},
 			Bucket:    bucket,
 			EventType: ReplicateIncomingDelete,
@@ -2505,7 +2503,7 @@ func (api objectAPIHandlers) PutObjectLegalHoldHandler(w http.ResponseWriter, r
 	popts := ObjectOptions{
 		MTime:     opts.MTime,
 		VersionID: opts.VersionID,
-		EvalMetadataFn: func(oi *ObjectInfo) error {
+		EvalMetadataFn: func(oi *ObjectInfo, gerr error) (ReplicateDecision, error) {
 			oi.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = strings.ToUpper(string(legalHold.Status))
 			oi.UserDefined[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = UTCNow().Format(time.RFC3339Nano)
 
@@ -2514,7 +2512,7 @@ func (api objectAPIHandlers) PutObjectLegalHoldHandler(w http.ResponseWriter, r
 				oi.UserDefined[ReservedMetadataPrefixLower+ReplicationTimestamp] = UTCNow().Format(time.RFC3339Nano)
 				oi.UserDefined[ReservedMetadataPrefixLower+ReplicationStatus] = dsc.PendingStatus()
 			}
-			return nil
+			return dsc, nil
 		},
 	}
 
@@ -2666,9 +2664,9 @@ func (api objectAPIHandlers) PutObjectRetentionHandler(w http.ResponseWriter, r
 	popts := ObjectOptions{
 		MTime:     opts.MTime,
 		VersionID: opts.VersionID,
-		EvalMetadataFn: func(oi *ObjectInfo) error {
+		EvalMetadataFn: func(oi *ObjectInfo, gerr error) (dsc ReplicateDecision, err error) {
 			if err := enforceRetentionBypassForPut(ctx, r, *oi, objRetention, cred, owner); err != nil {
-				return err
+				return dsc, err
 			}
 			if objRetention.Mode.Valid() {
 				oi.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = string(objRetention.Mode)
@@ -2678,12 +2676,12 @@ func (api objectAPIHandlers) PutObjectRetentionHandler(w http.ResponseWriter, r
 				oi.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = ""
 			}
 			oi.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = UTCNow().Format(time.RFC3339Nano)
-			dsc := mustReplicate(ctx, bucket, object, getMustReplicateOptions(*oi, replication.MetadataReplicationType, opts))
+			dsc = mustReplicate(ctx, bucket, object, getMustReplicateOptions(*oi, replication.MetadataReplicationType, opts))
 			if dsc.ReplicateAny() {
 				oi.UserDefined[ReservedMetadataPrefixLower+ReplicationTimestamp] = UTCNow().Format(time.RFC3339Nano)
 				oi.UserDefined[ReservedMetadataPrefixLower+ReplicationStatus] = dsc.PendingStatus()
 			}
-			return nil
+			return dsc, nil
 		},
 	}