From 769f0b1e24c42d6ef1a370a5c657c9f6d5484fe7 Mon Sep 17 00:00:00 2001
From: Harshavardhana <harsha@minio.io>
Date: Thu, 23 Sep 2021 17:24:24 -0700
Subject: [PATCH] fix: fallback listing on drives that are unformatted,
 disconnected (#13249)

---
 cmd/metacache-server-pool.go |  4 ++++
 cmd/metacache-set.go         | 33 +++++++++++++++++++++------------
 2 files changed, 25 insertions(+), 12 deletions(-)

diff --git a/cmd/metacache-server-pool.go b/cmd/metacache-server-pool.go
index 7ef1cca16..02d2fc1bd 100644
--- a/cmd/metacache-server-pool.go
+++ b/cmd/metacache-server-pool.go
@@ -209,6 +209,10 @@ func (z *erasureServerPools) listPath(ctx context.Context, o *listPathOptions) (
 			}
 		}()
 		o.ID = ""
+
+		if err != nil {
+			logger.LogIf(ctx, fmt.Errorf("Resuming listing from drives failed %w, proceeding to do raw listing", err))
+		}
 	}
 
 	// Do listing in-place.
diff --git a/cmd/metacache-set.go b/cmd/metacache-set.go
index 700933c1e..9fdceb287 100644
--- a/cmd/metacache-set.go
+++ b/cmd/metacache-set.go
@@ -365,7 +365,6 @@ func (er *erasureObjects) streamMetadataParts(ctx context.Context, o listPathOpt
 		}
 
 		const retryDelay = 500 * time.Millisecond
-		// Load first part metadata...
 		// All operations are performed without locks, so we must be careful and allow for failures.
 		// Read metadata associated with the object from a disk.
 		if retries > 0 {
@@ -373,6 +372,9 @@ func (er *erasureObjects) streamMetadataParts(ctx context.Context, o listPathOpt
 				if disk == nil {
 					continue
 				}
+				if !disk.IsOnline() {
+					continue
+				}
 				_, err := disk.ReadVersion(ctx, minioMetaBucket,
 					o.objectPath(0), "", false)
 				if err != nil {
@@ -384,6 +386,7 @@ func (er *erasureObjects) streamMetadataParts(ctx context.Context, o listPathOpt
 			}
 		}
 
+		// Load first part metadata...
 		// Read metadata associated with the object from all disks.
 		fi, metaArr, onlineDisks, err := er.getObjectFileInfo(ctx, minioMetaBucket, o.objectPath(0), ObjectOptions{}, true)
 		if err != nil {
@@ -441,6 +444,9 @@ func (er *erasureObjects) streamMetadataParts(ctx context.Context, o listPathOpt
 						if disk == nil {
 							continue
 						}
+						if !disk.IsOnline() {
+							continue
+						}
 						_, err := disk.ReadVersion(ctx, minioMetaBucket,
 							o.objectPath(partN), "", false)
 						if err != nil {
@@ -452,7 +458,7 @@ func (er *erasureObjects) streamMetadataParts(ctx context.Context, o listPathOpt
 					}
 				}
 
-				// Load first part metadata...
+				// Load partN metadata...
 				fi, metaArr, onlineDisks, err = er.getObjectFileInfo(ctx, minioMetaBucket, o.objectPath(partN), ObjectOptions{}, true)
 				if err != nil {
 					time.Sleep(retryDelay)
@@ -785,11 +791,13 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) {
 	defer cancel()
 
 	fallback := func(err error) bool {
-		if err == nil {
-			return false
+		switch err.(type) {
+		case StorageErr:
+			// all supported disk errors
+			// attempt a fallback.
+			return true
 		}
-		return err.Error() == errUnformattedDisk.Error() ||
-			err.Error() == errVolumeNotFound.Error()
+		return false
 	}
 	askDisks := len(disks)
 	readers := make([]*metacacheReader, askDisks)
@@ -844,6 +852,8 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) {
 			if werr != io.EOF && werr != nil &&
 				werr.Error() != errFileNotFound.Error() &&
 				werr.Error() != errVolumeNotFound.Error() &&
+				werr.Error() != errDiskNotFound.Error() &&
+				werr.Error() != errUnformattedDisk.Error() &&
 				!errors.Is(werr, context.Canceled) {
 				logger.LogIf(ctx, werr)
 			}
@@ -874,12 +884,11 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) {
 				continue
 			case nil:
 			default:
-				if err.Error() == errFileNotFound.Error() {
-					atEOF++
-					fnf++
-					continue
-				}
-				if err.Error() == errVolumeNotFound.Error() {
+				switch err.Error() {
+				case errFileNotFound.Error(),
+					errVolumeNotFound.Error(),
+					errUnformattedDisk.Error(),
+					errDiskNotFound.Error():
 					atEOF++
 					fnf++
 					continue