Save listing error async (#10922)

Since the RPC call may have to time out save an error state async 
to not hold up the listing returning.

Fixes #10919

cmd/metacache-server-sets.go

@@ -22,6 +22,7 @@ import (
 	"io"
 	"path"
 	"sync"
+	"time"
 
 	"github.com/minio/minio/cmd/logger"
 )
@@ -103,14 +104,18 @@ func (z *erasureServerSets) listPath(ctx context.Context, o listPathOptions) (en
 			// Local
 			cache = localMetacacheMgr.findCache(ctx, o)
 		} else {
+			ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+			defer cancel()
 			c, err := rpc.GetMetacacheListing(ctx, o)
 			if err != nil {
-				if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+				if errors.Is(err, context.Canceled) {
 					// Context is canceled, return at once.
 					// request canceled, no entries to return
 					return entries, io.EOF
 				}
+				if !errors.Is(err, context.DeadlineExceeded) {
 					logger.LogIf(ctx, err)
+				}
 				o.Transient = true
 				cache = localMetacacheMgr.findCache(ctx, o)
 			} else {
@@ -176,11 +181,12 @@ func (z *erasureServerSets) listPath(ctx context.Context, o listPathOptions) (en
 
 	if isAllNotFound(errs) {
 		// All sets returned not found.
+		go func() {
 			// Update master cache with that information.
 			cache.status = scanStateSuccess
 			cache.fileNotFound = true
-		_, err := o.updateMetacacheListing(cache, globalNotificationSys.restClientFromHash(o.Bucket))
+			o.updateMetacacheListing(cache, globalNotificationSys.restClientFromHash(o.Bucket))
-		logger.LogIf(ctx, err)
+		}()
 		// cache returned not found, entries truncated.
 		return entries, io.EOF
 	}

cmd/metacache-set.go

@@ -560,13 +560,15 @@ func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions) (entr
 			console.Println("listPath returning:", entries.len(), "err:", err)
 		}
 		if err != nil && err != io.EOF {
+			go func(err string) {
 				metaMu.Lock()
 				if meta.status != scanStateError {
-				meta.error = err.Error()
+					meta.error = err
 					meta.status = scanStateError
 				}
 				meta, _ = o.updateMetacacheListing(meta, rpc)
 				metaMu.Unlock()
+			}(err.Error())
 			cancel()
 		}
 	}()