fix: replication deleteObject() regression and CopyObject() behavior (#14780)

This PR fixes two issues

- The first fix is a regression from #14555, the fix itself in #14555
  is correct but the interpretation of that information by the
  object layer code for "replication" was not correct. This PR
  tries to fix this situation by making sure the "Delete" replication
  works as expected when "VersionPurgeStatus" is already set.

  Without this fix, there is a DELETE marker created incorrectly on
  the source where the "DELETE" was triggered.

- The second fix is perhaps an older problem started since we inlined-data
  on the disk for small objects, CopyObject() incorrectly inline's
  a non-inlined data. This is due to the fact that we have code where
  we read the `part.1` under certain conditions where the size of the
  `part.1` is less than the specific "threshold".

  This eventually causes problems when we are "deleting" the data that
  is only inlined, which means dataDir is ignored leaving such
  dataDir on the disk, that looks like an inconsistent content on
  the namespace.

fixes #14767

cmd/erasure-object.go

@@ -112,9 +112,20 @@ func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, d
 		// preserve destination versionId if specified.
 		if versionID == "" {
 			versionID = mustGetUUID()
+			fi.IsLatest = true // we are creating a new version so this is latest.
 		}
 		modTime = UTCNow()
 	}
+
+	// If the data is not inlined, we may end up incorrectly
+	// inlining the data here, that leads to an inconsistent
+	// situation where some objects are were not inlined
+	// were now inlined, make sure to `nil` the Data such
+	// that xl.meta is written as expected.
+	if !fi.InlineData() {
+		fi.Data = nil
+	}
+
 	fi.VersionID = versionID // set any new versionID we might have created
 	fi.ModTime = modTime     // set modTime for the new versionID
 	if !dstOpts.MTime.IsZero() {
@@ -130,6 +141,14 @@ func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, d
 			metaArr[index].ModTime = modTime
 			metaArr[index].VersionID = versionID
 			metaArr[index].Metadata = srcInfo.UserDefined
+			if !metaArr[index].InlineData() {
+				// If the data is not inlined, we may end up incorrectly
+				// inlining the data here, that leads to an inconsistent
+				// situation where some objects are were not inlined
+				// were now inlined, make sure to `nil` the Data such
+				// that xl.meta is written as expected.
+				metaArr[index].Data = nil
+			}
 		}
 	}
 
@@ -138,8 +157,6 @@ func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, d
 		return oi, toObjectErr(err, srcBucket, srcObject)
 	}
 
-	// we are adding a new version to this object under the namespace lock, so this is the latest version.
-	fi.IsLatest = true
 	return fi.ToObjectInfo(srcBucket, srcObject), nil
 }
 
@@ -1319,10 +1336,21 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string
 		if opts.VersionPurgeStatus() == Complete {
 			markDelete = false
 		}
-		// determine if the version represents an object delete
-		// deleteMarker = true
-		if versionFound && !goi.DeleteMarker { // implies a versioned delete of object
-			deleteMarker = false
+
+		// Version is found but we do not wish to create more delete markers
+		// 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.
+		if versionFound {
+			if !goi.VersionPurgeStatus.Empty() {
+				deleteMarker = false
+			} else if !goi.DeleteMarker { // implies a versioned delete of object
+				deleteMarker = false
+			}
 		}
 	}
 

cmd/object-handlers.go

@@ -1118,10 +1118,10 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 	}
 
 	var chStorageClass bool
-	if dstSc != "" {
+	if dstSc != "" && dstSc != srcInfo.StorageClass {
 		chStorageClass = true
 		srcInfo.metadataOnly = false
-	}
+	} // no changes in storage-class expected so its a metadataonly operation.
 
 	var reader io.Reader = gr
 

cmd/xl-storage.go

@@ -901,19 +901,24 @@ func (s *xlStorage) deleteVersions(ctx context.Context, volume, path string, fis
 			if versionID == "" {
 				versionID = nullVersionID
 			}
+
 			// PR #11758 used DataDir, preserve it
 			// for users who might have used master
 			// branch
-			if !xlMeta.data.remove(versionID, dataDir) {
-				filePath := pathJoin(volumeDir, path, dataDir)
-				if err = checkPathLength(filePath); err != nil {
+			xlMeta.data.remove(versionID, dataDir)
+
+			// We need to attempt delete "dataDir" on the disk
+			// due to a CopyObject() bug where it might have
+			// inlined the data incorrectly, to avoid a situation
+			// where we potentially leave "DataDir"
+			filePath := pathJoin(volumeDir, path, dataDir)
+			if err = checkPathLength(filePath); err != nil {
+				return err
+			}
+			if err = s.moveToTrash(filePath, true); err != nil {
+				if err != errFileNotFound {
 					return err
 				}
-				if err = s.moveToTrash(filePath, true); err != nil {
-					if err != errFileNotFound {
-						return err
-					}
-				}
 			}
 		}
 	}
@@ -1026,16 +1031,20 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F
 		// PR #11758 used DataDir, preserve it
 		// for users who might have used master
 		// branch
-		if !xlMeta.data.remove(versionID, dataDir) {
-			filePath := pathJoin(volumeDir, path, dataDir)
-			if err = checkPathLength(filePath); err != nil {
+		xlMeta.data.remove(versionID, dataDir)
+
+		// We need to attempt delete "dataDir" on the disk
+		// due to a CopyObject() bug where it might have
+		// inlined the data incorrectly, to avoid a situation
+		// where we potentially leave "DataDir"
+		filePath := pathJoin(volumeDir, path, dataDir)
+		if err = checkPathLength(filePath); err != nil {
+			return err
+		}
+		if err = s.moveToTrash(filePath, true); err != nil {
+			if err != errFileNotFound {
 				return err
 			}
-			if err = s.moveToTrash(filePath, true); err != nil {
-				if err != errFileNotFound {
-					return err
-				}
-			}
 		}
 	}