From 9bf5990ea939a0afb7fe4870d2c0889a6db04fb2 Mon Sep 17 00:00:00 2001
From: Klaus Post <klauspost@gmail.com>
Date: Fri, 6 Nov 2020 08:54:09 -0800
Subject: [PATCH] metadata: Invalidate cache if unreadable and not updating
 (#10844)

If a scanning server shuts down unexpectedly we may have "successful" caches that are incomplete on a set.

In this case mark the cache with an error so it will no longer be handed out.
---
 cmd/metacache-manager.go | 28 +++++++++++++++++-----------
 1 file changed, 17 insertions(+), 11 deletions(-)

diff --git a/cmd/metacache-manager.go b/cmd/metacache-manager.go
index a3061f128..099a9c6f9 100644
--- a/cmd/metacache-manager.go
+++ b/cmd/metacache-manager.go
@@ -233,17 +233,14 @@ func (o listPathOptions) checkMetacacheState(ctx context.Context, rpc *peerRESTC
 	// We operate on a copy...
 	o.Create = false
 	var cache metacache
-	if !o.Transient {
-		if rpc == nil || o.Transient {
-			// Local
-			cache = localMetacacheMgr.findCache(ctx, o)
-		} else {
-			c, err := rpc.GetMetacacheListing(ctx, o)
-			if err != nil {
-				return err
-			}
-			cache = *c
+	if rpc == nil || o.Transient {
+		cache = localMetacacheMgr.findCache(ctx, o)
+	} else {
+		c, err := rpc.GetMetacacheListing(ctx, o)
+		if err != nil {
+			return err
 		}
+		cache = *c
 	}
 
 	if cache.status == scanStateNone || cache.fileNotFound {
@@ -251,7 +248,16 @@ func (o listPathOptions) checkMetacacheState(ctx context.Context, rpc *peerRESTC
 	}
 	if cache.status == scanStateSuccess || cache.status == scanStateStarted {
 		if time.Since(cache.lastUpdate) > metacacheMaxRunningAge {
-			return fmt.Errorf("timeout: list %s not updated for 1 minute", cache.id)
+			// We got a stale entry, mark error on handling server.
+			err := fmt.Errorf("timeout: list %s not updated", cache.id)
+			cache.error = err.Error()
+			cache.status = scanStateError
+			if rpc == nil || o.Transient {
+				localMetacacheMgr.updateCacheEntry(cache)
+			} else {
+				rpc.UpdateMetacacheListing(ctx, cache)
+			}
+			return err
 		}
 		return nil
 	}