fix: do not return an error for successfully deleted dangling objects

dangling objects when removed `mc admin heal -r` or crawler
auto heal would incorrectly return error - this can interfere
with usage calculation as the entry size for this would be
returned as `0`, instead upon success use the resultant
object size to calculate the final size for the object
and avoid reporting this in the log messages

Also do not set ObjectSize in healResultItem to be '-1'
this has an effect on crawler metrics calculating 1 byte
less for objects which seem to be missing their `xl.meta`

cmd/erasure-healing.go

@@ -219,10 +219,12 @@ func shouldHealObjectOnDisk(erErr, dataErr error, meta FileInfo, quorumModTime t
 
 // Heals an object by re-writing corrupt/missing erasure blocks.
 func (er erasureObjects) healObject(ctx context.Context, bucket string, object string,
-	partsMetadata []FileInfo, errs []error, latestFileInfo FileInfo,
-	dryRun bool, remove bool, scanMode madmin.HealScanMode) (result madmin.HealResultItem, err error) {
+	versionID string, partsMetadata []FileInfo, errs []error, lfi FileInfo, opts madmin.HealOpts) (result madmin.HealResultItem, err error) {
 
-	dataBlocks := latestFileInfo.Erasure.DataBlocks
+	dryRun := opts.DryRun
+	scanMode := opts.ScanMode
+
+	dataBlocks := lfi.Erasure.DataBlocks
 
 	storageDisks := er.getDisks()
 	storageEndpoints := er.getEndpoints()
@@ -242,10 +244,6 @@ func (er erasureObjects) healObject(ctx context.Context, bucket string, object s
 		DiskCount:    len(storageDisks),
 		ParityBlocks: len(storageDisks) / 2,
 		DataBlocks:   len(storageDisks) / 2,
-
-		// Initialize object size to -1, so we can detect if we are
-		// unable to reliably find the object size.
-		ObjectSize: -1,
 	}
 
 	// Loop to find number of disks with valid data, per-drive
@@ -306,28 +304,14 @@ func (er erasureObjects) healObject(ctx context.Context, bucket string, object s
 	}
 
 	if isAllNotFound(errs) {
-		return defaultHealResult(latestFileInfo, storageDisks, storageEndpoints, errs, bucket, object), nil
+		// File is fully gone, fileInfo is empty.
+		return defaultHealResult(FileInfo{}, storageDisks, storageEndpoints, errs, bucket, object, versionID), nil
 	}
 
 	// If less than read quorum number of disks have all the parts
 	// of the data, we can't reconstruct the erasure-coded data.
 	if numAvailableDisks < dataBlocks {
-		// Check if er.meta, and corresponding parts are also missing.
-		if m, ok := isObjectDangling(partsMetadata, errs, dataErrs); ok {
-			writeQuorum := m.Erasure.DataBlocks + 1
-			if m.Erasure.DataBlocks == 0 {
-				writeQuorum = getWriteQuorum(len(storageDisks))
-			}
-			if !dryRun && remove {
-				if latestFileInfo.VersionID == "" {
-					err = er.deleteObject(ctx, bucket, object, writeQuorum)
-				} else {
-					err = er.deleteObjectVersion(ctx, bucket, object, writeQuorum, FileInfo{VersionID: latestFileInfo.VersionID})
-				}
-			}
-			return defaultHealResult(latestFileInfo, storageDisks, storageEndpoints, errs, bucket, object), err
-		}
-		return result, toObjectErr(errErasureReadQuorum, bucket, object)
+		return er.purgeObjectDangling(ctx, bucket, object, versionID, partsMetadata, errs, dataErrs, opts)
 	}
 
 	if disksToHealCount == 0 {
@@ -343,9 +327,9 @@ func (er erasureObjects) healObject(ctx context.Context, bucket string, object s
 
 	// Latest FileInfo for reference. If a valid metadata is not
 	// present, it is as good as object not found.
-	latestMeta, pErr := pickValidFileInfo(ctx, partsMetadata, modTime, dataBlocks)
-	if pErr != nil {
-		return result, toObjectErr(pErr, bucket, object)
+	latestMeta, err := pickValidFileInfo(ctx, partsMetadata, modTime, dataBlocks)
+	if err != nil {
+		return result, toObjectErr(err, bucket, object, versionID)
 	}
 	defer ObjectPathUpdated(pathJoin(bucket, object))
 
@@ -461,7 +445,7 @@ func (er erasureObjects) healObject(ctx context.Context, bucket string, object s
 	outDatedDisks, err = writeUniqueFileInfo(ctx, outDatedDisks, minioMetaTmpBucket, tmpID,
 		partsMetadata, diskCount(outDatedDisks))
 	if err != nil {
-		return result, toObjectErr(err, bucket, object)
+		return result, toObjectErr(err, bucket, object, versionID)
 	}
 
 	// Rename from tmp location to the actual location.
@@ -574,20 +558,17 @@ func (er erasureObjects) healObjectDir(ctx context.Context, bucket, object strin
 
 // Populates default heal result item entries with possible values when we are returning prematurely.
 // This is to ensure that in any circumstance we are not returning empty arrays with wrong values.
-func defaultHealResult(latestFileInfo FileInfo, storageDisks []StorageAPI, storageEndpoints []string, errs []error, bucket, object string) madmin.HealResultItem {
+func defaultHealResult(lfi FileInfo, storageDisks []StorageAPI, storageEndpoints []string, errs []error, bucket, object, versionID string) madmin.HealResultItem {
 	// Initialize heal result object
 	result := madmin.HealResultItem{
 		Type:      madmin.HealItemObject,
 		Bucket:    bucket,
 		Object:    object,
+		VersionID: versionID,
 		DiskCount: len(storageDisks),
-
-		// Initialize object size to -1, so we can detect if we are
-		// unable to reliably find the object size.
-		ObjectSize: -1,
 	}
-	if latestFileInfo.IsValid() {
-		result.ObjectSize = latestFileInfo.Size
+	if lfi.IsValid() {
+		result.ObjectSize = lfi.Size
 	}
 
 	for index, disk := range storageDisks {
@@ -621,13 +602,13 @@ func defaultHealResult(latestFileInfo FileInfo, storageDisks []StorageAPI, stora
 		})
 	}
 
-	if !latestFileInfo.IsValid() {
+	if !lfi.IsValid() {
 		// Default to most common configuration for erasure blocks.
 		result.ParityBlocks = getDefaultParityBlocks(len(storageDisks))
 		result.DataBlocks = getDefaultDataBlocks(len(storageDisks))
 	} else {
-		result.ParityBlocks = latestFileInfo.Erasure.ParityBlocks
-		result.DataBlocks = latestFileInfo.Erasure.DataBlocks
+		result.ParityBlocks = lfi.Erasure.ParityBlocks
+		result.DataBlocks = lfi.Erasure.DataBlocks
 	}
 
 	return result
@@ -693,6 +674,52 @@ func isObjectDirDangling(errs []error) (ok bool) {
 	return found < notFound && found > 0
 }
 
+func (er erasureObjects) purgeObjectDangling(ctx context.Context, bucket, object, versionID string,
+	metaArr []FileInfo, errs []error, dataErrs []error, opts madmin.HealOpts) (madmin.HealResultItem, error) {
+
+	storageDisks := er.getDisks()
+	storageEndpoints := er.getEndpoints()
+
+	// Check if the object is dangling, if yes and user requested
+	// remove we simply delete it from namespace.
+	m, ok := isObjectDangling(metaArr, errs, dataErrs)
+	if ok {
+		writeQuorum := m.Erasure.DataBlocks
+		if m.Erasure.DataBlocks == 0 || m.Erasure.DataBlocks == m.Erasure.ParityBlocks {
+			writeQuorum = getWriteQuorum(len(storageDisks))
+		}
+		var err error
+		if !opts.DryRun && opts.Remove {
+			if versionID == "" {
+				err = er.deleteObject(ctx, bucket, object, writeQuorum)
+			} else {
+				err = er.deleteObjectVersion(ctx, bucket, object, writeQuorum, FileInfo{VersionID: versionID})
+			}
+		}
+		if !opts.DryRun && opts.Remove {
+			// Delete was successful, make sure to return the appropriate error
+			// and heal result appropriate with delete's error messages
+			for i := range errs {
+				errs[i] = err
+			}
+		}
+		if err == nil {
+			// dangling object successfully purged, size is '0'
+			m.Size = 0
+			return defaultHealResult(m, storageDisks, storageEndpoints, errs, bucket, object, versionID), toObjectErr(err, bucket, object, versionID)
+		}
+		return defaultHealResult(m, storageDisks, storageEndpoints, errs, bucket, object, versionID), toObjectErr(err, bucket, object, versionID)
+	}
+
+	readQuorum := m.Erasure.DataBlocks
+	if m.Erasure.DataBlocks == 0 || m.Erasure.DataBlocks == m.Erasure.ParityBlocks {
+		readQuorum = getReadQuorum(len(storageDisks))
+	}
+
+	err := toObjectErr(reduceReadQuorumErrs(ctx, errs, objectOpIgnoredErrs, readQuorum), bucket, object, versionID)
+	return defaultHealResult(m, storageDisks, storageEndpoints, errs, bucket, object, versionID), err
+}
+
 // Object is considered dangling/corrupted if any only
 // if total disks - a combination of corrupted and missing
 // files is lesser than number of data blocks.
@@ -771,61 +798,16 @@ func (er erasureObjects) HealObject(ctx context.Context, bucket, object, version
 	partsMetadata, errs := readAllFileInfo(healCtx, storageDisks, bucket, object, versionID)
 
 	if isAllNotFound(errs) {
-		// Nothing to do
-		return defaultHealResult(FileInfo{}, storageDisks, storageEndpoints, errs, bucket, object), nil
-	}
-	// Check if the object is dangling, if yes and user requested
-	// remove we simply delete it from namespace.
-	if m, ok := isObjectDangling(partsMetadata, errs, []error{}); ok {
-		writeQuorum := m.Erasure.DataBlocks + 1
-		if m.Erasure.DataBlocks == 0 {
-			writeQuorum = getWriteQuorum(len(storageDisks))
-		}
-		if !opts.DryRun && opts.Remove {
-			if versionID == "" {
-				er.deleteObject(healCtx, bucket, object, writeQuorum)
-			} else {
-				er.deleteObjectVersion(healCtx, bucket, object, writeQuorum, FileInfo{VersionID: versionID})
-			}
-		}
-		err = reduceReadQuorumErrs(ctx, errs, nil, writeQuorum-1)
-		return defaultHealResult(FileInfo{}, storageDisks, storageEndpoints, errs, bucket, object), toObjectErr(err, bucket, object)
+		// Nothing to do, file is already gone.
+		return defaultHealResult(FileInfo{}, storageDisks, storageEndpoints, errs, bucket, object, versionID), nil
 	}
 
-	latestFileInfo, err := getLatestFileInfo(healCtx, partsMetadata, errs)
+	fi, err := getLatestFileInfo(healCtx, partsMetadata, errs)
 	if err != nil {
-		return defaultHealResult(FileInfo{}, storageDisks, storageEndpoints, errs, bucket, object), toObjectErr(err, bucket, object)
-	}
-
-	errCount := 0
-	for _, err := range errs {
-		if err != nil {
-			errCount++
-		}
-	}
-
-	if errCount == len(errs) {
-		// Only if we get errors from all the disks we return error. Else we need to
-		// continue to return filled madmin.HealResultItem struct which includes info
-		// on what disks the file is available etc.
-		if err = reduceReadQuorumErrs(ctx, errs, nil, latestFileInfo.Erasure.DataBlocks); err != nil {
-			if m, ok := isObjectDangling(partsMetadata, errs, []error{}); ok {
-				writeQuorum := m.Erasure.DataBlocks + 1
-				if m.Erasure.DataBlocks == 0 {
-					writeQuorum = getWriteQuorum(len(storageDisks))
-				}
-				if !opts.DryRun && opts.Remove {
-					if versionID == "" {
-						er.deleteObject(ctx, bucket, object, writeQuorum)
-					} else {
-						er.deleteObjectVersion(ctx, bucket, object, writeQuorum, FileInfo{VersionID: versionID})
-					}
-				}
-			}
-			return defaultHealResult(latestFileInfo, storageDisks, storageEndpoints, errs, bucket, object), toObjectErr(err, bucket, object)
-		}
+		fmt.Println(err)
+		return er.purgeObjectDangling(healCtx, bucket, object, versionID, partsMetadata, errs, []error{}, opts)
 	}
 
 	// Heal the object.
-	return er.healObject(healCtx, bucket, object, partsMetadata, errs, latestFileInfo, opts.DryRun, opts.Remove, opts.ScanMode)
+	return er.healObject(healCtx, bucket, object, versionID, partsMetadata, errs, fi, opts)
 }

cmd/erasure-object_test.go

@@ -19,6 +19,7 @@ package cmd
 import (
 	"bytes"
 	"context"
+	"errors"
 	"io/ioutil"
 	"os"
 	"testing"
@@ -257,7 +258,7 @@ func TestErasureDeleteObjectDiskNotFound(t *testing.T) {
 	z.serverSets[0].erasureDisksMu.Unlock()
 	_, err = obj.DeleteObject(ctx, bucket, object, ObjectOptions{})
 	// since majority of disks are not available, metaquorum is not achieved and hence errErasureWriteQuorum error
-	if err != toObjectErr(errErasureWriteQuorum, bucket, object) {
+	if !errors.Is(err, errErasureWriteQuorum) {
 		t.Errorf("Expected deleteObject to fail with %v, but failed with %v", toObjectErr(errErasureWriteQuorum, bucket, object), err)
 	}
 }
@@ -381,7 +382,7 @@ func TestPutObjectNoQuorum(t *testing.T) {
 		z.serverSets[0].erasureDisksMu.Unlock()
 		// Upload new content to same object "object"
 		_, err = obj.PutObject(ctx, bucket, object, mustGetPutObjReader(t, bytes.NewReader([]byte("abcd")), int64(len("abcd")), "", ""), opts)
-		if err != toObjectErr(errErasureWriteQuorum, bucket, object) {
+		if !errors.Is(err, errErasureWriteQuorum) {
 			t.Errorf("Expected putObject to fail with %v, but failed with %v", toObjectErr(errErasureWriteQuorum, bucket, object), err)
 		}
 	}

cmd/object-api-errors.go

@@ -130,9 +130,27 @@ func toObjectErr(err error, params ...string) error {
 			}
 		}
 	case errErasureReadQuorum:
-		err = InsufficientReadQuorum{}
+		if len(params) == 1 {
+			err = InsufficientReadQuorum{
+				Bucket: params[0],
+			}
+		} else if len(params) >= 2 {
+			err = InsufficientReadQuorum{
+				Bucket: params[0],
+				Object: params[1],
+			}
+		}
 	case errErasureWriteQuorum:
-		err = InsufficientWriteQuorum{}
+		if len(params) == 1 {
+			err = InsufficientWriteQuorum{
+				Bucket: params[0],
+			}
+		} else if len(params) >= 2 {
+			err = InsufficientWriteQuorum{
+				Bucket: params[0],
+				Object: params[1],
+			}
+		}
 	case io.ErrUnexpectedEOF, io.ErrShortWrite:
 		err = IncompleteBody{}
 	case context.Canceled, context.DeadlineExceeded:
@@ -163,10 +181,10 @@ func (e SlowDown) Error() string {
 }
 
 // InsufficientReadQuorum storage cannot satisfy quorum for read operation.
-type InsufficientReadQuorum struct{}
+type InsufficientReadQuorum GenericError
 
 func (e InsufficientReadQuorum) Error() string {
-	return "Storage resources are insufficient for the read operation."
+	return "Storage resources are insufficient for the read operation " + e.Bucket + "/" + e.Object
 }
 
 // Unwrap the error.
@@ -175,10 +193,10 @@ func (e InsufficientReadQuorum) Unwrap() error {
 }
 
 // InsufficientWriteQuorum storage cannot satisfy quorum for write operation.
-type InsufficientWriteQuorum struct{}
+type InsufficientWriteQuorum GenericError
 
 func (e InsufficientWriteQuorum) Error() string {
-	return "Storage resources are insufficient for the write operation."
+	return "Storage resources are insufficient for the write operation " + e.Bucket + "/" + e.Object
 }
 
 // Unwrap the error.
@@ -194,6 +212,11 @@ type GenericError struct {
 	Err       error
 }
 
+// Unwrap the error to its underlying error.
+func (e GenericError) Unwrap() error {
+	return e.Err
+}
+
 // InvalidArgument incorrect input argument
 type InvalidArgument GenericError
 

cmd/object-api-putobject_test.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"crypto/md5"
 	"encoding/hex"
+	"errors"
 	"io/ioutil"
 	"os"
 	"path"
@@ -296,7 +297,7 @@ func testObjectAPIPutObjectDiskNotFound(obj ObjectLayer, instanceType string, di
 		int64(len("mnop")),
 		false,
 		"",
-		InsufficientWriteQuorum{},
+		errErasureWriteQuorum,
 	}
 
 	_, actualErr := obj.PutObject(context.Background(), testCase.bucketName, testCase.objName, mustGetPutObjReader(t, bytes.NewReader(testCase.inputData), testCase.intputDataSize, testCase.inputMeta["etag"], sha256sum), ObjectOptions{UserDefined: testCase.inputMeta})
@@ -305,7 +306,7 @@ func testObjectAPIPutObjectDiskNotFound(obj ObjectLayer, instanceType string, di
 	}
 	// Failed as expected, but does it fail for the expected reason.
 	if actualErr != nil && !testCase.shouldPass {
-		if testCase.expectedError.Error() != actualErr.Error() {
+		if !errors.Is(actualErr, testCase.expectedError) {
 			t.Errorf("Test %d: %s: Expected to fail with error \"%s\", but instead failed with error \"%s\" instead.", len(testCases)+1, instanceType, testCase.expectedError.Error(), actualErr.Error())
 		}
 	}

pkg/madmin/heal-commands.go

@@ -118,6 +118,7 @@ type HealResultItem struct {
 	Type         HealItemType `json:"type"`
 	Bucket       string       `json:"bucket"`
 	Object       string       `json:"object"`
+	VersionID    string       `json:"versionId"`
 	Detail       string       `json:"detail"`
 	ParityBlocks int          `json:"parityBlocks,omitempty"`
 	DataBlocks   int          `json:"dataBlocks,omitempty"`