From 4a412223100ce85912105357f5ecaa9690e3b816 Mon Sep 17 00:00:00 2001
From: Harshavardhana <harsha@minio.io>
Date: Wed, 21 Apr 2021 10:57:36 -0700
Subject: [PATCH] fix: newMultipartUpload should go to same pool (#12106)

avoid potential for duplicates under multi-pool
setup, additionally also make sure CompleteMultipart
is using a more optimal API for uploadID lookup
and never delete the object there is a potential
to create a delete marker during complete multipart.

Signed-off-by: Harshavardhana <harsha@minio.io>
---
 cmd/erasure-server-pool.go | 34 ++++++++++++++++++++--------------
 1 file changed, 20 insertions(+), 14 deletions(-)

diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go
index 85c8a784c..db8363ff4 100644
--- a/cmd/erasure-server-pool.go
+++ b/cmd/erasure-server-pool.go
@@ -1035,10 +1035,23 @@ func (z *erasureServerPools) NewMultipartUpload(ctx context.Context, bucket, obj
 		return z.serverPools[0].NewMultipartUpload(ctx, bucket, object, opts)
 	}
 
-	// We don't know the exact size, so we ask for at least 1GiB file.
-	idx, err := z.getPoolIdx(ctx, bucket, object, 1<<30)
-	if err != nil {
-		return "", err
+	for idx, pool := range z.serverPools {
+		result, err := pool.ListMultipartUploads(ctx, bucket, object, "", "", "", maxUploadsList)
+		if err != nil {
+			return "", err
+		}
+		// If there is a multipart upload with the same bucket/object name,
+		// create the new multipart in the same pool, this will avoid
+		// creating two multiparts uploads in two different pools
+		if len(result.Uploads) != 0 {
+			return z.serverPools[idx].NewMultipartUpload(ctx, bucket, object, opts)
+		}
+	}
+
+	// We multiply the size by 2 to account for erasure coding.
+	idx := z.getAvailablePoolIdx(ctx, (1<<30)*2)
+	if idx < 0 {
+		return "", toObjectErr(errDiskFull)
 	}
 
 	return z.serverPools[idx].NewMultipartUpload(ctx, bucket, object, opts)
@@ -1180,20 +1193,13 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
 		return z.serverPools[0].CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts)
 	}
 
-	// Purge any existing object.
 	for _, pool := range z.serverPools {
-		pool.DeleteObject(ctx, bucket, object, opts)
-	}
-
-	for _, pool := range z.serverPools {
-		result, err := pool.ListMultipartUploads(ctx, bucket, object, "", "", "", maxUploadsList)
-		if err != nil {
-			return objInfo, err
-		}
-		if result.Lookup(uploadID) {
+		_, err := pool.GetMultipartInfo(ctx, bucket, object, uploadID, opts)
+		if err == nil {
 			return pool.CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts)
 		}
 	}
+
 	return objInfo, InvalidUploadID{
 		Bucket:   bucket,
 		Object:   object,