fix: delete-markers without quorum were unreadable (#13351)

DeleteMarkers were unreadable if they had quorum based
guarantees, this PR tries to fix this behavior appropriately.

DeleteMarkers with sufficient should be allowed and the
return error should be accordingly with or without version-id.

This also allows for overwrites which may not be possible
in a multi-pool setup.

fixes #12787

cmd/erasure-healing-common.go

@@ -40,7 +40,11 @@ func commonTime(modTimes []time.Time, dataDirs []string) (modTime time.Time, dat
 	}
 
 	for _, dataDir := range dataDirs {
-		if dataDir == "" {
+		if dataDir == errorDir {
+			continue
+		}
+		if dataDir == delMarkerDir {
+			dataDirOccurenceMap[delMarkerDir]++
 			continue
 		}
 		dataDirOccurenceMap[dataDir]++
@@ -97,6 +101,11 @@ func listObjectModtimes(partsMetadata []FileInfo, errs []error) (modTimes []time
 	return modTimes
 }
 
+const (
+	errorDir     = "error-dir"
+	delMarkerDir = ""
+)
+
 // Notes:
 // There are 5 possible states a disk could be in,
 // 1. __online__             - has the latest copy of xl.meta - returned by listOnlineDisks
@@ -130,6 +139,7 @@ func listOnlineDisks(disks []StorageAPI, partsMetadata []FileInfo, errs []error)
 	dataDirs := make([]string, len(partsMetadata))
 	for idx, fi := range partsMetadata {
 		if errs[idx] != nil {
+			dataDirs[idx] = errorDir
 			continue
 		}
 		dataDirs[idx] = fi.DataDir
@@ -151,9 +161,9 @@ func listOnlineDisks(disks []StorageAPI, partsMetadata []FileInfo, errs []error)
 }
 
 // Returns the latest updated FileInfo files and error in case of failure.
-func getLatestFileInfo(ctx context.Context, partsMetadata []FileInfo, errs []error) (FileInfo, error) {
+func getLatestFileInfo(ctx context.Context, partsMetadata []FileInfo, errs []error, quorum int) (FileInfo, error) {
 	// There should be atleast half correct entries, if not return failure
-	if reducedErr := reduceReadQuorumErrs(ctx, errs, objectOpIgnoredErrs, len(partsMetadata)/2); reducedErr != nil {
+	if reducedErr := reduceReadQuorumErrs(ctx, errs, objectOpIgnoredErrs, quorum); reducedErr != nil {
 		return FileInfo{}, reducedErr
 	}
 
@@ -182,7 +192,8 @@ func getLatestFileInfo(ctx context.Context, partsMetadata []FileInfo, errs []err
 			count++
 		}
 	}
-	if count < len(partsMetadata)/2 {
+
+	if count < quorum {
 		return FileInfo{}, errErasureReadQuorum
 	}
 

cmd/erasure-healing-common_test.go

@@ -190,7 +190,7 @@ func TestListOnlineDisks(t *testing.T) {
 			}
 
 			partsMetadata, errs := readAllFileInfo(ctx, erasureDisks, bucket, object, "", false)
-			fi, err := getLatestFileInfo(ctx, partsMetadata, errs)
+			fi, err := getLatestFileInfo(ctx, partsMetadata, errs, getReadQuorum(len(disks)))
 			if err != nil {
 				t.Fatalf("Failed to getLatestFileInfo %v", err)
 			}
@@ -365,7 +365,7 @@ func TestListOnlineDisksSmallObjects(t *testing.T) {
 			}
 
 			partsMetadata, errs := readAllFileInfo(ctx, erasureDisks, bucket, object, "", true)
-			fi, err := getLatestFileInfo(ctx, partsMetadata, errs)
+			fi, err := getLatestFileInfo(ctx, partsMetadata, errs, getReadQuorum(len(disks)))
 			if err != nil {
 				t.Fatalf("Failed to getLatestFileInfo %v", err)
 			}
@@ -421,7 +421,7 @@ func TestListOnlineDisksSmallObjects(t *testing.T) {
 
 			}
 			partsMetadata, errs = readAllFileInfo(ctx, erasureDisks, bucket, object, "", true)
-			_, err = getLatestFileInfo(ctx, partsMetadata, errs)
+			_, err = getLatestFileInfo(ctx, partsMetadata, errs, len(disks)/2)
 			if err != nil {
 				t.Fatalf("Failed to getLatestFileInfo %v", err)
 			}

cmd/erasure-healing.go

@@ -275,7 +275,7 @@ func (er erasureObjects) healObject(ctx context.Context, bucket string, object s
 	// Re-read when we have lock...
 	partsMetadata, errs := readAllFileInfo(ctx, storageDisks, bucket, object, versionID, true)
 
-	if _, err = getLatestFileInfo(ctx, partsMetadata, errs); err != nil {
+	if _, err = getLatestFileInfo(ctx, partsMetadata, errs, er.defaultParityCount); err != nil {
 		return er.purgeObjectDangling(ctx, bucket, object, versionID, partsMetadata, errs, []error{}, opts)
 	}
 

cmd/erasure-healing_test.go

@@ -416,7 +416,7 @@ func TestHealObjectCorrupted(t *testing.T) {
 	}
 
 	fileInfos, errs := readAllFileInfo(ctx, erasureDisks, bucket, object, "", false)
-	fi, err := getLatestFileInfo(ctx, fileInfos, errs)
+	fi, err := getLatestFileInfo(ctx, fileInfos, errs, er.defaultParityCount)
 	if err != nil {
 		t.Fatalf("Failed to getLatestFileInfo - %v", err)
 	}
@@ -441,7 +441,7 @@ func TestHealObjectCorrupted(t *testing.T) {
 	}
 
 	fileInfos, errs = readAllFileInfo(ctx, erasureDisks, bucket, object, "", false)
-	nfi, err := getLatestFileInfo(ctx, fileInfos, errs)
+	nfi, err := getLatestFileInfo(ctx, fileInfos, errs, er.defaultParityCount)
 	if err != nil {
 		t.Fatalf("Failed to getLatestFileInfo - %v", err)
 	}
@@ -467,7 +467,7 @@ func TestHealObjectCorrupted(t *testing.T) {
 	}
 
 	fileInfos, errs = readAllFileInfo(ctx, erasureDisks, bucket, object, "", false)
-	nfi, err = getLatestFileInfo(ctx, fileInfos, errs)
+	nfi, err = getLatestFileInfo(ctx, fileInfos, errs, er.defaultParityCount)
 	if err != nil {
 		t.Fatalf("Failed to getLatestFileInfo - %v", err)
 	}

cmd/erasure-metadata.go

@@ -388,7 +388,7 @@ func writeUniqueFileInfo(ctx context.Context, disks []StorageAPI, bucket, prefix
 // writeQuorum is the min required disks to write data.
 func objectQuorumFromMeta(ctx context.Context, partsMetaData []FileInfo, errs []error, defaultParityCount int) (objectReadQuorum, objectWriteQuorum int, err error) {
 	// get the latest updated Metadata and a count of all the latest updated FileInfo(s)
-	latestFileInfo, err := getLatestFileInfo(ctx, partsMetaData, errs)
+	latestFileInfo, err := getLatestFileInfo(ctx, partsMetaData, errs, defaultParityCount)
 	if err != nil {
 		return 0, 0, err
 	}