From 0cde17ae5d0b2c867245cd008cb02921f26ce046 Mon Sep 17 00:00:00 2001
From: Klaus Post <klauspost@gmail.com>
Date: Wed, 1 May 2024 10:59:08 -0700
Subject: [PATCH] Return listing when exceeding min disk errors (#19644)

When listing, with drives returning `errFileNotFound,` `errVolumeNotFound`, or `errUnformattedDisk,`,
we could get below `minDisks` drives being left.

This would result in a quorum never being reachable for any object. Therefore, the listing
would continue, but no results would ever be produced.

Include `fnf` in the mindisk check since it is incremented on these errors. This will stop
listing when minDisks are left.

Allow `opts.minDisks` to not return errVolumeNotFound or errFileNotFound and return that.
That will allow for good results even if disks return something else.

We switch `errUnformattedDisk` to a regular error. If we have enough of those, we should just fail.
---
 cmd/metacache-set.go | 38 ++++++++++++++++++--------------------
 1 file changed, 18 insertions(+), 20 deletions(-)

diff --git a/cmd/metacache-set.go b/cmd/metacache-set.go
index 9c073e1c9..1c9a6d085 100644
--- a/cmd/metacache-set.go
+++ b/cmd/metacache-set.go
@@ -1010,8 +1010,7 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) {
 		// not a storage error.
 		return nil
 	}
-	askDisks := len(disks)
-	readers := make([]*metacacheReader, askDisks)
+	readers := make([]*metacacheReader, len(disks))
 	defer func() {
 		for _, r := range readers {
 			r.Close()
@@ -1093,16 +1092,14 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) {
 			case nil:
 			default:
 				switch err.Error() {
-				case errFileNotFound.Error(),
-					errVolumeNotFound.Error(),
-					errUnformattedDisk.Error():
+				case errFileNotFound.Error():
 					atEOF++
 					fnf++
-					// This is a special case, to handle bucket does
-					// not exist situations.
-					if errors.Is(err, errVolumeNotFound) {
-						vnf++
-					}
+					continue
+				case errVolumeNotFound.Error():
+					atEOF++
+					fnf++
+					vnf++
 					continue
 				}
 				hasErr++
@@ -1141,8 +1138,17 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) {
 			topEntries[i] = entry
 		}
 
-		// Stop if we exceed number of bad disks
-		if hasErr > len(disks)-opts.minDisks && hasErr > 0 {
+		// Since minDisks is set to quorum, we return if we have enough.
+		if vnf > 0 && vnf >= len(readers)-opts.minDisks {
+			return errVolumeNotFound
+		}
+		// Since minDisks is set to quorum, we return if we have enough.
+		if fnf > 0 && fnf >= len(readers)-opts.minDisks {
+			return errFileNotFound
+		}
+
+		// Stop if we exceed number of bad disks.
+		if hasErr > 0 && hasErr+fnf > len(disks)-opts.minDisks {
 			if opts.finished != nil {
 				opts.finished(errs)
 			}
@@ -1160,10 +1166,6 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) {
 			return errors.New(strings.Join(combinedErr, ", "))
 		}
 
-		if vnf == len(readers) {
-			return errVolumeNotFound
-		}
-
 		// Break if all at EOF or error.
 		if atEOF+hasErr == len(readers) {
 			if hasErr > 0 && opts.finished != nil {
@@ -1172,10 +1174,6 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) {
 			break
 		}
 
-		if fnf == len(readers) {
-			return errFileNotFound
-		}
-
 		if agree == len(readers) {
 			// Everybody agreed
 			for _, r := range readers {