XL: erasure Index should have its corresponding distribution order. (#2300)

object-api-multipart_test.go

@@ -40,7 +40,7 @@ func testObjectNewMultipartUpload(obj ObjectLayer, instanceType string, t TestEr
 	// opearation expected to fail since the bucket on which NewMultipartUpload is being initiated doesn't exist.
 	uploadID, err := obj.NewMultipartUpload(bucket, object, nil)
 	if err == nil {
-		t.Fatalf("%s: Expected to fail since the NewMultipartUpload is intialized on a non-existant bucket.", instanceType)
+		t.Fatalf("%s: Expected to fail since the NewMultipartUpload is intialized on a non-existent bucket.", instanceType)
 	}
 	if errMsg != err.Error() {
 		t.Errorf("%s, Expected to fail with Error \"%s\", but instead found \"%s\".", instanceType, errMsg, err.Error())

object-handlers.go

@@ -226,7 +226,7 @@ func (api objectAPIHandlers) HeadObjectHandler(w http.ResponseWriter, r *http.Re
 	// Set standard object headers.
 	setObjectHeaders(w, objInfo, nil)
 
-	// Successfull response.
+	// Successful response.
 	w.WriteHeader(http.StatusOK)
 }
 

server_test.go

@@ -780,7 +780,7 @@ func (s *TestSuiteCommon) TestListBuckets(c *C) {
 	// parse the list bucket response.
 	decoder := xml.NewDecoder(response.Body)
 	err = decoder.Decode(&results)
-	// validating that the xml-decoding/parsing was successfull.
+	// validating that the xml-decoding/parsing was successful.
 	c.Assert(err, IsNil)
 }
 
@@ -1560,7 +1560,7 @@ func (s *TestSuiteCommon) TestGetPartialObjectLarge10MiB(c *C) {
 	// execute the HTTP request to upload the object.
 	response, err = client.Do(request)
 	c.Assert(err, IsNil)
-	// verify whether upload was successfull.
+	// verify whether upload was successful.
 	c.Assert(response.StatusCode, Equals, http.StatusOK)
 
 	// HTTP request to download the object.
@@ -1653,7 +1653,7 @@ func (s *TestSuiteCommon) TestGetObjectRangeErrors(c *C) {
 	// execute the HTTP request to upload the object.
 	response, err = client.Do(request)
 	c.Assert(err, IsNil)
-	// verify whether upload was successfull.
+	// verify whether upload was successful.
 	c.Assert(response.StatusCode, Equals, http.StatusOK)
 
 	// HTTP request to download the object.
@@ -2140,7 +2140,7 @@ func (s *TestSuiteCommon) TestObjectMultipart(c *C) {
 	// Execute the complete multipart request.
 	response, err = client.Do(request)
 	c.Assert(err, IsNil)
-	// verify whether complete multipart was successfull.
+	// verify whether complete multipart was successful.
 	c.Assert(response.StatusCode, Equals, http.StatusOK)
 
 }

xl-v1-multipart.go

@@ -617,7 +617,7 @@ func (xl xlObjects) CompleteMultipartUpload(bucket string, object string, upload
 		return "", toObjectErr(errXLWriteQuorum, bucket, object)
 	}
 
-	_, modTime := listOnlineDisks(xl.storageDisks, partsMetadata, errs)
+	onlineDisks, modTime := listOnlineDisks(xl.storageDisks, partsMetadata, errs)
 
 	// Calculate full object size.
 	var objectSize int64
@@ -625,6 +625,12 @@ func (xl xlObjects) CompleteMultipartUpload(bucket string, object string, upload
 	// Pick one from the first valid metadata.
 	xlMeta := pickValidXLMeta(partsMetadata, modTime)
 
+	// Order online disks in accordance with distribution order.
+	onlineDisks = getOrderedDisks(xlMeta.Erasure.Distribution, onlineDisks)
+
+	// Order parts metadata in accordance with distribution order.
+	partsMetadata = getOrderedPartsMetadata(xlMeta.Erasure.Distribution, partsMetadata)
+
 	// Save current xl meta for validation.
 	var currentXLMeta = xlMeta
 
@@ -695,10 +701,10 @@ func (xl xlObjects) CompleteMultipartUpload(bucket string, object string, upload
 	}
 
 	// Write unique `xl.json` for each disk.
-	if err = writeUniqueXLMetadata(xl.storageDisks, minioMetaBucket, tempUploadIDPath, partsMetadata, xl.writeQuorum); err != nil {
+	if err = writeUniqueXLMetadata(onlineDisks, minioMetaBucket, tempUploadIDPath, partsMetadata, xl.writeQuorum); err != nil {
 		return "", toObjectErr(err, minioMetaBucket, tempUploadIDPath)
 	}
-	rErr := commitXLMetadata(xl.storageDisks, tempUploadIDPath, uploadIDPath, xl.writeQuorum)
+	rErr := commitXLMetadata(onlineDisks, tempUploadIDPath, uploadIDPath, xl.writeQuorum)
 	if rErr != nil {
 		return "", toObjectErr(rErr, minioMetaBucket, uploadIDPath)
 	}
@@ -721,6 +727,9 @@ func (xl xlObjects) CompleteMultipartUpload(bucket string, object string, upload
 	// Rename if an object already exists to temporary location.
 	uniqueID := getUUID()
 	if xl.isObject(bucket, object) {
+		// NOTE: Do not use online disks slice here.
+		// The reason is that existing object should be purged
+		// regardless of `xl.json` status and rolled back in case of errors.
 		err = renameObject(xl.storageDisks, bucket, object, minioMetaBucket, path.Join(tmpMetaPrefix, uniqueID), xl.writeQuorum)
 		if err != nil {
 			return "", toObjectErr(err, bucket, object)
@@ -741,7 +750,7 @@ func (xl xlObjects) CompleteMultipartUpload(bucket string, object string, upload
 	}
 
 	// Rename the multipart object to final location.
-	if err = renameObject(xl.storageDisks, minioMetaBucket, uploadIDPath, bucket, object, xl.writeQuorum); err != nil {
+	if err = renameObject(onlineDisks, minioMetaBucket, uploadIDPath, bucket, object, xl.writeQuorum); err != nil {
 		return "", toObjectErr(err, bucket, object)
 	}
 

xl-v1-object.go

@@ -487,6 +487,9 @@ func (xl xlObjects) PutObject(bucket string, object string, size int64, data io.
 	// Rename if an object already exists to temporary location.
 	newUniqueID := getUUID()
 	if xl.isObject(bucket, object) {
+		// NOTE: Do not use online disks slice here.
+		// The reason is that existing object should be purged
+		// regardless of `xl.json` status and rolled back in case of errors.
 		err = renameObject(xl.storageDisks, bucket, object, minioMetaTmpBucket, newUniqueID, xl.writeQuorum)
 		if err != nil {
 			return "", toObjectErr(err, bucket, object)