fix: make sure to lock CopyObject for in-place updates (#10492)

cmd/bucket-replication.go

@@ -18,6 +18,7 @@ package cmd
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"strings"
 	"time"
@@ -179,20 +180,24 @@ func replicateObject(ctx context.Context, bucket, object, versionID string, obje
 	if tgt == nil {
 		return
 	}
-	gr, err := objectAPI.GetObjectNInfo(ctx, bucket, object, nil, http.Header{}, readLock, ObjectOptions{})
+	gr, err := objectAPI.GetObjectNInfo(ctx, bucket, object, nil, http.Header{}, readLock, ObjectOptions{
+		VersionID: versionID,
+	})
 	if err != nil {
 		return
 	}
-	defer gr.Close()
+
 	objInfo := gr.ObjInfo
 	size, err := objInfo.GetActualSize()
 	if err != nil {
 		logger.LogIf(ctx, err)
+		gr.Close()
 		return
 	}
 
 	dest := cfg.GetDestination()
 	if dest.Bucket == "" {
+		gr.Close()
 		return
 	}
 	// In the rare event that replication is in pending state either due to
@@ -201,6 +206,7 @@ func replicateObject(ctx context.Context, bucket, object, versionID string, obje
 	if healPending {
 		_, err := tgt.StatObject(ctx, dest.Bucket, object, miniogo.StatObjectOptions{VersionID: objInfo.VersionID})
 		if err == nil {
+			gr.Close()
 			// object with same VersionID already exists, replication kicked off by
 			// PutObject might have completed.
 			return
@@ -210,6 +216,7 @@ func replicateObject(ctx context.Context, bucket, object, versionID string, obje
 
 	replicationStatus := replication.Complete
 	_, err = tgt.PutObject(ctx, dest.Bucket, object, gr, size, "", "", putOpts)
+	gr.Close()
 	if err != nil {
 		replicationStatus = replication.Failed
 		// Notify replication failure  event.
@@ -231,8 +238,10 @@ func replicateObject(ctx context.Context, bucket, object, versionID string, obje
 	objInfo.metadataOnly = true // Perform only metadata updates.
 	if _, err = objectAPI.CopyObject(ctx, bucket, object, bucket, object, objInfo, ObjectOptions{
 		VersionID: objInfo.VersionID,
-	}, ObjectOptions{VersionID: objInfo.VersionID}); err != nil {
-		logger.LogIf(ctx, err)
+	}, ObjectOptions{
+		VersionID: objInfo.VersionID,
+	}); err != nil {
+		logger.LogIf(ctx, fmt.Errorf("Unable to update replication metadata for %s: %s", objInfo.VersionID, err))
 	}
 }
 

cmd/data-update-tracker.go

@@ -426,6 +426,8 @@ func (d *dataUpdateTracker) deserialize(src io.Reader, newerThan time.Time) erro
 	}
 	// Ignore what remains on the stream.
 	// Update d:
+	d.mu.Lock()
+	defer d.mu.Unlock()
 	d.Current = dst.Current
 	d.History = dst.History
 	d.Saved = dst.Saved

cmd/erasure-object.go

@@ -72,6 +72,11 @@ func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, d
 		return oi, NotImplemented{}
 	}
 	defer ObjectPathUpdated(path.Join(dstBucket, dstObject))
+	lk := er.NewNSLock(ctx, dstBucket, dstObject)
+	if err := lk.GetLock(globalOperationTimeout); err != nil {
+		return oi, err
+	}
+	defer lk.Unlock()
 
 	// Read metadata associated with the object from all disks.
 	storageDisks := er.getDisks()
@@ -112,12 +117,13 @@ func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, d
 	fi.VersionID = versionID // set any new versionID we might have created
 	fi.ModTime = modTime     // set modTime for the new versionID
 
+	srcInfo.UserDefined["etag"] = srcInfo.ETag
+
 	// Update `xl.meta` content on each disks.
 	for index := range metaArr {
 		metaArr[index].ModTime = modTime
 		metaArr[index].VersionID = versionID
 		metaArr[index].Metadata = srcInfo.UserDefined
-		metaArr[index].Metadata["etag"] = srcInfo.ETag
 	}
 
 	tempObj := mustGetUUID()
@@ -741,6 +747,12 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 		return ObjectInfo{}, IncompleteBody{Bucket: bucket, Object: object}
 	}
 
+	lk := er.NewNSLock(ctx, bucket, object)
+	if err := lk.GetLock(globalOperationTimeout); err != nil {
+		return ObjectInfo{}, err
+	}
+	defer lk.Unlock()
+
 	for i, w := range writers {
 		if w == nil {
 			onlineDisks[i] = nil
@@ -780,12 +792,6 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 		return ObjectInfo{}, toObjectErr(err, bucket, object)
 	}
 
-	lk := er.NewNSLock(ctx, bucket, object)
-	if err := lk.GetLock(globalOperationTimeout); err != nil {
-		return ObjectInfo{}, err
-	}
-	defer lk.Unlock()
-
 	// Rename the successfully written temporary object to final location.
 	if onlineDisks, err = renameData(ctx, onlineDisks, minioMetaTmpBucket, tempObj, fi.DataDir, bucket, object, writeQuorum, nil); err != nil {
 		return ObjectInfo{}, toObjectErr(err, bucket, object)