From 417ec0df56175e74ec2c442f382abd1528459564 Mon Sep 17 00:00:00 2001
From: Krishnan Parthasarathi <krisis@users.noreply.github.com>
Date: Wed, 22 Mar 2017 22:45:16 +0530
Subject: [PATCH] HealObject should succeed when only N/2 disks have data
 (#3952)

---
 cmd/xl-v1-healing-common.go |  6 +++---
 cmd/xl-v1-healing_test.go   | 27 ++++++++++++++++++++++++---
 cmd/xl-v1-utils.go          |  9 ++++++++-
 cmd/xl-v1-utils_test.go     |  3 +++
 4 files changed, 38 insertions(+), 7 deletions(-)

diff --git a/cmd/xl-v1-healing-common.go b/cmd/xl-v1-healing-common.go
index 38f79ba6a..d1ef3335e 100644
--- a/cmd/xl-v1-healing-common.go
+++ b/cmd/xl-v1-healing-common.go
@@ -225,10 +225,10 @@ func disksWithAllParts(onlineDisks []StorageAPI, partsMetadata []xlMetaV1, errs
 		// disk has a valid xl.json but may not have all the
 		// parts. This is considered an outdated disk, since
 		// it needs healing too.
-		for pIndex, part := range partsMetadata[index].Parts {
+		for _, part := range partsMetadata[index].Parts {
 			// compute blake2b sum of part.
 			partPath := filepath.Join(object, part.Name)
-			hash := newHash(blake2bAlgo)
+			hash := newHash(partsMetadata[index].Erasure.Algorithm)
 			blakeBytes, hErr := hashSum(onlineDisk, bucket, partPath, hash)
 			if hErr == errFileNotFound {
 				errs[index] = errFileNotFound
@@ -239,7 +239,7 @@ func disksWithAllParts(onlineDisks []StorageAPI, partsMetadata []xlMetaV1, errs
 				return nil, nil, traceError(hErr)
 			}
 
-			partChecksum := partsMetadata[index].Erasure.Checksum[pIndex].Hash
+			partChecksum := partsMetadata[index].Erasure.GetCheckSumInfo(part.Name).Hash
 			blakeSum := hex.EncodeToString(blakeBytes)
 			// if blake2b sum doesn't match for a part
 			// then this disk is outdated and needs
diff --git a/cmd/xl-v1-healing_test.go b/cmd/xl-v1-healing_test.go
index e84b0af3e..7c97b538e 100644
--- a/cmd/xl-v1-healing_test.go
+++ b/cmd/xl-v1-healing_test.go
@@ -518,15 +518,36 @@ func TestHealObjectXL(t *testing.T) {
 
 	bucket := "bucket"
 	object := "object"
-	data := []byte("hello")
+	data := bytes.Repeat([]byte("a"), 5*1024*1024)
+
 	err = obj.MakeBucket(bucket)
 	if err != nil {
 		t.Fatalf("Failed to make a bucket - %v", err)
 	}
 
-	_, err = obj.PutObject(bucket, object, int64(len(data)), bytes.NewReader(data), nil, "")
+	// Create an object with multiple parts uploaded in decreasing
+	// part number.
+	uploadID, err := obj.NewMultipartUpload(bucket, object, nil)
+	if err != nil {
+		t.Fatalf("Failed to create a multipart upload - %v", err)
+	}
+
+	var uploadedParts []completePart
+	for _, partID := range []int{2, 1} {
+		pInfo, err := obj.PutObjectPart(bucket, object, uploadID, partID,
+			int64(len(data)), bytes.NewReader(data), "", "")
+		if err != nil {
+			t.Fatalf("Failed to upload a part - %v", err)
+		}
+		uploadedParts = append(uploadedParts, completePart{
+			PartNumber: pInfo.PartNumber,
+			ETag:       pInfo.ETag,
+		})
+	}
+
+	_, err = obj.CompleteMultipartUpload(bucket, object, uploadID, uploadedParts)
 	if err != nil {
-		t.Fatalf("Failed to put an object - %v", err)
+		t.Fatalf("Failed to complete multipart upload - %v", err)
 	}
 
 	// Remove the object backend files from the first disk.
diff --git a/cmd/xl-v1-utils.go b/cmd/xl-v1-utils.go
index 4faf67d74..283b690ad 100644
--- a/cmd/xl-v1-utils.go
+++ b/cmd/xl-v1-utils.go
@@ -41,11 +41,18 @@ func reduceErrs(errs []error, ignoredErrs []error) (maxCount int, maxErr error)
 		}
 		errorCounts[err]++
 	}
+
 	max := 0
 	for err, count := range errorCounts {
-		if max < count {
+		switch {
+		case max < count:
 			max = count
 			maxErr = err
+
+		// Prefer `nil` over other error values with the same
+		// number of occurrences.
+		case max == count && err == nil:
+			maxErr = err
 		}
 	}
 	return max, maxErr
diff --git a/cmd/xl-v1-utils_test.go b/cmd/xl-v1-utils_test.go
index e167f226d..d992918a0 100644
--- a/cmd/xl-v1-utils_test.go
+++ b/cmd/xl-v1-utils_test.go
@@ -82,6 +82,9 @@ func TestReduceErrs(t *testing.T) {
 			errDiskNotFound,
 		}, []error{errDiskNotFound}, errVolumeNotFound},
 		{[]error{}, []error{}, errXLReadQuorum},
+		{[]error{errFileNotFound, errFileNotFound, errFileNotFound,
+			errFileNotFound, errFileNotFound, nil, nil, nil, nil, nil},
+			nil, nil},
 	}
 	// Validates list of all the testcases for returning valid errors.
 	for i, testCase := range testCases {