Re-look-up a chunk's locations as soon as they all fail (#10800)

* mount: re-resolve volume locations after a failed chunk read

NewChunkGroup passed nil as the ReaderCache's CacheInvalidator, so
retryFetchAfterCacheInvalidation was dead code on the FUSE read path. A
mount that cached a volume's locations while one server was down kept
retrying that server after it died, then returned EIO, even though the
master and filer both resolved the live replica. The S3 gateway already
passes its filerClient; do the same for the mount.

* test: FUSE integration tests for volume server failover

One mount appends while a second tails, and a volume server is killed,
started or restarted mid-stream against a 001-replicated cluster of three
volume servers. Automates the scenario matrix reported for Docker Swarm
mounts, including the large-file variant and a no-chaos control.

* test: report the filer's own view when append content mismatches

A mismatch between what the writer wrote and what the reader sees can come
from either side's cache. Read the file back through the filer's HTTP
handler as well, and let the mount verbosity be raised from the
environment, so a failing run says which layer lost the data.

* test: wait for the reader mount to converge before comparing

A mount caches metadata for about a second, so reading the file the instant
the writer's last close returned can legitimately come back short. Poll the
reader until it matches or the timeout expires; content that is wrong rather
than merely late never converges and still fails, now with the writer's
mount and the filer's own view alongside it.

* test: detect a failover cluster child that exited at startup

Signal(0) succeeds for a zombie and nothing reaped these children until
shutdown, so a process that died on startup looked alive until the readiness
timeout expired. Reap each child as it is started and consult the result.

* test: read a file the killed volume server actually holds

Placement decides which two of three servers back each volume, so killing
volume N and reading readfile-N could pass without the victim ever holding a
replica of it. Resolve each file's volumes through the filer and the master,
and pick one the victim backs, preferring a file the reader has not cached.

* ci: stop persisting checkout credentials in the failover workflow

The job does not use the token after cloning. Also tag the README's command
block as bash and match the timeout the workflow actually uses.

* test: discard the ignored errors errcheck flags in the failover harness

* test: resolve manifests when mapping a file to its volumes

A manifest chunk's own fid names the volume holding the manifest, not the
volumes holding the data, so a large enough file would point the failover
victim at the wrong server.

* test: pin the stale-location recovery path with a primed reader

Reading a file for the first time after a server dies proves nothing: the
lookup is fresh and returns the survivor. Kill one holder and wait for the
master to drop it, read a file on that volume so the reader caches the lone
survivor, restart the first server, then kill the survivor. The reader's only
cached location is now dead while the data is live elsewhere, which is the
case the invalidator exists for: EIO without it, recovery with it.

* filer: re-look-up a chunk's locations as soon as they all fail

A read that fails against every location it was given is far more likely to be
holding a stale list than to be hitting a cluster that is briefly slow, but the
retry loops spent the whole backoff ladder, about 13 s, before the caller got a
chance to invalidate and look the chunk up again. Give the loops a refresh hook
and let the reader cache invalidate on the first fully failed pass, so recovery
starts in milliseconds. Clients without an invalidator keep the old behavior.

The filer's streaming read path has its own fetch loop and is not covered.

* filer: refresh locations on the random-read path too

readChunkSliceAt bypasses the chunk cacher in random-access mode and fetches
the range directly, which left it without the invalidation the cacher does:
a random reader parked on a stale location had no way back at all. Hoist the
refresh hook onto the reader cache so both paths share it.

* filer: compare chunk locations as a set, not in order

Lookups shuffle the locations they return, so comparing positionally reads a
reshuffle of the very same replicas as a fresh set and spends an immediate
retry on locations that just failed. weed/filer already had an
order-independent comparison for this; move it next to the retry loops so
both callers share one helper.

weed/filer/filechunk_manifest.go

@@ -117,13 +117,13 @@ func fetchWholeChunk(ctx context.Context, bytesBuffer *bytes.Buffer, lookupFileI
 	return nil
 }
 
-func fetchChunkRange(ctx context.Context, buffer []byte, lookupFileIdFn wdclient.LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool, offset int64) (int, error) {
+func fetchChunkRange(ctx context.Context, buffer []byte, lookupFileIdFn wdclient.LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool, offset int64, refreshUrls util_http.RefreshUrlsFunc) (int, error) {
 	urlStrings, err := lookupFileIdFn(ctx, fileId)
 	if err != nil {
 		glog.ErrorfCtx(ctx, "operation LookupFileId %s failed, err: %v", fileId, err)
 		return 0, err
 	}
-	return util_http.RetriedFetchChunkData(ctx, buffer, urlStrings, cipherKey, isGzipped, false, offset, fileId)
+	return util_http.RetriedFetchChunkData(ctx, buffer, urlStrings, cipherKey, isGzipped, false, offset, fileId, refreshUrls)
 }
 
 func retriedStreamFetchChunkData(ctx context.Context, writer io.Writer, urlStrings []string, jwt string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, size int) (written int64, err error) {

weed/filer/reader_at.go

@@ -324,7 +324,8 @@ func (c *ChunkReadAt) readChunkSliceAt(ctx context.Context, buffer []byte, chunk
 		if n > 0 {
 			return n, err
 		}
-		return fetchChunkRange(ctx, buffer, c.readerCache.lookupFileIdFn, chunkView.FileId, chunkView.CipherKey, chunkView.IsGzipped, int64(offset))
+		return fetchChunkRange(ctx, buffer, c.readerCache.lookupFileIdFn, chunkView.FileId, chunkView.CipherKey, chunkView.IsGzipped, int64(offset),
+			c.readerCache.refreshUrls(ctx, chunkView.FileId))
 	}
 
 	shouldCache := (uint64(chunkView.ViewOffset) + chunkView.ChunkSize) <= c.readerCache.chunkCache.GetMaxFilePartSizeInCache()

weed/filer/reader_cache.go

@@ -18,7 +18,7 @@ type CacheInvalidator interface {
 	InvalidateCache(fileId string)
 }
 
-type fetchChunkDataFnType func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string) (n int, err error)
+type fetchChunkDataFnType func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (n int, err error)
 
 type ReaderCache struct {
 	chunkCache       chunk_cache.ChunkCache
@@ -104,6 +104,25 @@ func (rc *ReaderCache) MaybeCache(chunkViews *Interval[*ChunkView], count int) {
 	return
 }
 
+// refreshUrls lets a fetch loop recover inside a single read: when every cached
+// location for a chunk has failed, drop the cached entry and look it up again
+// rather than spending the whole backoff ladder on locations that are gone.
+// Nil when there is nothing to invalidate against.
+func (rc *ReaderCache) refreshUrls(ctx context.Context, fileId string) util_http.RefreshUrlsFunc {
+	if rc.cacheInvalidator == nil || rc.lookupFileIdFn == nil {
+		return nil
+	}
+	return func() []string {
+		rc.cacheInvalidator.InvalidateCache(fileId)
+		urls, err := rc.lookupFileIdFn(ctx, fileId)
+		if err != nil {
+			glog.V(0).InfofCtx(ctx, "re-lookup chunk %s: %v", fileId, err)
+			return nil
+		}
+		return urls
+	}
+}
+
 func (rc *ReaderCache) ReadChunkAt(ctx context.Context, buffer []byte, fileId string, cipherKey []byte, isGzipped bool, offset int64, chunkSize int, shouldCache bool) (int, error) {
 	rc.Lock()
 
@@ -270,7 +289,7 @@ func (s *SingleChunkCacher) fetchChunkData(ctx context.Context, urlStrings []str
 	// Allocate buffer and download without holding the lock.
 	// This allows multiple downloads to proceed in parallel.
 	data := mem.Allocate(s.chunkSize)
-	_, fetchErr := s.parent.fetchChunkDataFn(ctx, data, urlStrings, s.cipherKey, s.isGzipped, true, 0, s.chunkFileId)
+	_, fetchErr := s.parent.fetchChunkDataFn(ctx, data, urlStrings, s.cipherKey, s.isGzipped, true, 0, s.chunkFileId, s.parent.refreshUrls(ctx, s.chunkFileId))
 	if fetchErr != nil {
 		mem.Free(data)
 		return nil, fetchErr

weed/filer/reader_cache_test.go

@@ -3,6 +3,7 @@ package filer
 import (
 	"context"
 	"fmt"
+	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -100,7 +101,7 @@ func TestReaderCacheRetryAfterCacheInvalidation(t *testing.T) {
 	}
 
 	var fetchCount int32
-	fetchFn := func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, requestedFileId string) (int, error) {
+	fetchFn := func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, requestedFileId string, _ util_http.RefreshUrlsFunc) (int, error) {
 		if requestedFileId != fileId {
 			return 0, fmt.Errorf("unexpected fetch file id %s", requestedFileId)
 		}
@@ -161,7 +162,7 @@ func TestReaderCacheRemovesFailedDownloader(t *testing.T) {
 	}
 
 	var fetchCount int32
-	fetchFn := func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, requestedFileId string) (int, error) {
+	fetchFn := func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, requestedFileId string, _ util_http.RefreshUrlsFunc) (int, error) {
 		atomic.AddInt32(&fetchCount, 1)
 		return 0, fmt.Errorf("fetch failed")
 	}
@@ -575,7 +576,7 @@ func TestReaderCacheDownloaderDedup(t *testing.T) {
 		return []string{"http://volume/" + fileId}, nil
 	}
 
-	fetchFn := func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string) (int, error) {
+	fetchFn := func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, _ util_http.RefreshUrlsFunc) (int, error) {
 		atomic.AddInt32(&fetchCount, 1)
 		<-fetchGate
 		return copy(buffer, testData), nil

weed/filer/stream.go

@@ -102,26 +102,6 @@ func PrepareStreamContent(masterClient wdclient.HasLookupFileIdFunction, jwtFunc
 
 type VolumeServerJwtFunction func(fileId string) string
 
-// urlSlicesEqual checks if two URL slices contain the same URLs (order-independent)
-func urlSlicesEqual(a, b []string) bool {
-	if len(a) != len(b) {
-		return false
-	}
-	// Create a map to count occurrences in first slice
-	counts := make(map[string]int)
-	for _, url := range a {
-		counts[url]++
-	}
-	// Verify all URLs in second slice match
-	for _, url := range b {
-		if counts[url] == 0 {
-			return false
-		}
-		counts[url]--
-	}
-	return true
-}
-
 // retryFetchWithFreshLocations is the shared self-heal for the read paths: when a chunk fetch
 // fails, invalidate the cached volume locations, re-lookup, and call refetch only when the
 // resolved locations actually changed (so we never retry against the same servers). originalErr
@@ -143,7 +123,7 @@ func retryFetchWithFreshLocations(ctx context.Context, invalidator CacheInvalida
 		glog.WarningfCtx(ctx, "re-lookup for chunk %s returned no locations, skipping retry", fileId)
 		return fmt.Errorf("re-lookup chunk %s returned no locations", fileId)
 	}
-	if urlSlicesEqual(oldUrls, newUrls) {
+	if util_http.SameUrls(oldUrls, newUrls) {
 		glog.V(0).InfofCtx(ctx, "re-lookup returned same locations for chunk %s, skipping retry", fileId)
 		return originalErr
 	}

weed/util/http/http_global_client_util.go

@@ -544,7 +544,52 @@ func (r *CountingReader) Read(p []byte) (n int, err error) {
 	return n, err
 }
 
-func RetriedFetchChunkData(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string) (n int, err error) {
+// refreshedUrls asks for a fresh location list and reports whether it is worth
+// retrying on: a list that comes back empty, or identical to the one that just
+// failed everywhere, says the locations were never the problem.
+func refreshedUrls(ctx context.Context, refreshUrls RefreshUrlsFunc, current []string, fileId string) ([]string, bool) {
+	if refreshUrls == nil {
+		return nil, false
+	}
+	fresh := refreshUrls()
+	if len(fresh) == 0 || SameUrls(current, fresh) {
+		return nil, false
+	}
+	glog.V(0).InfofCtx(ctx, "chunk %s failed on every known location, retrying on %d fresh ones", fileId, len(fresh))
+	return fresh, true
+}
+
+// SameUrls reports whether two location lists hold the same URLs, regardless of
+// order: lookups shuffle the locations they return, so comparing positionally
+// would read a reshuffle of the very same replicas as a fresh set.
+func SameUrls(a, b []string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	counts := make(map[string]int, len(a))
+	for _, url := range a {
+		counts[url]++
+	}
+	for _, url := range b {
+		if counts[url] == 0 {
+			return false
+		}
+		counts[url]--
+	}
+	return true
+}
+
+// RefreshUrlsFunc supplies a fresh location list for a chunk. The retry loops
+// call it once, after every location in the list they were given has failed,
+// which is the point at which the list itself is the likely problem. Returning
+// nil or the same list leaves the caller on the original locations.
+type RefreshUrlsFunc func() []string
+
+// RetriedFetchChunkData reads a chunk, trying every location before backing off
+// and trying them again. refreshUrls may be nil; when it is not, a pass in which
+// every location failed is treated as a stale list rather than a slow cluster,
+// and the fresh list is tried immediately instead of after the next backoff.
+func RetriedFetchChunkData(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls RefreshUrlsFunc) (n int, err error) {
 
 	loadJwtConfigOnce.Do(loadJwtConfig)
 	var jwt security.EncodedJwt
@@ -555,7 +600,7 @@ func RetriedFetchChunkData(ctx context.Context, buffer []byte, urlStrings []stri
 	// For unencrypted, non-gzipped full chunks, use direct buffer read
 	// This avoids the 64KB intermediate buffer and callback overhead
 	if cipherKey == nil && !isGzipped && isFullChunk {
-		return retriedFetchChunkDataDirect(ctx, buffer, urlStrings, string(jwt))
+		return retriedFetchChunkDataDirect(ctx, buffer, urlStrings, string(jwt), fileId, refreshUrls)
 	}
 
 	var shouldRetry bool
@@ -604,6 +649,11 @@ func RetriedFetchChunkData(ctx context.Context, buffer []byte, urlStrings []stri
 			}
 		}
 		if err != nil && shouldRetry {
+			if fresh, ok := refreshedUrls(ctx, refreshUrls, urlStrings, fileId); ok {
+				urlStrings, refreshUrls = fresh, nil
+				continue
+			}
+			refreshUrls = nil
 			glog.V(0).InfofCtx(ctx, "retry reading in %v", waitTime)
 			// Sleep with proper context cancellation and timer cleanup
 			timer := time.NewTimer(waitTime)
@@ -626,7 +676,7 @@ func RetriedFetchChunkData(ctx context.Context, buffer []byte, urlStrings []stri
 // retriedFetchChunkDataDirect reads chunk data directly into the buffer without
 // intermediate buffering. This reduces memory copies and improves throughput
 // for large chunk reads.
-func retriedFetchChunkDataDirect(ctx context.Context, buffer []byte, urlStrings []string, jwt string) (n int, err error) {
+func retriedFetchChunkDataDirect(ctx context.Context, buffer []byte, urlStrings []string, jwt, fileId string, refreshUrls RefreshUrlsFunc) (n int, err error) {
 	var shouldRetry bool
 
 	for waitTime := time.Second; waitTime < util.RetryWaitTime; waitTime += waitTime / 2 {
@@ -654,6 +704,11 @@ func retriedFetchChunkDataDirect(ctx context.Context, buffer []byte, urlStrings
 		}
 
 		if err != nil && shouldRetry {
+			if fresh, ok := refreshedUrls(ctx, refreshUrls, urlStrings, fileId); ok {
+				urlStrings, refreshUrls = fresh, nil
+				continue
+			}
+			refreshUrls = nil
 			glog.V(0).InfofCtx(ctx, "retry reading in %v", waitTime)
 			timer := time.NewTimer(waitTime)
 			select {

weed/util/http/http_global_client_util_test.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"testing"
+	"time"
 )
 
 func TestAppendQueryParameter(t *testing.T) {
@@ -132,3 +133,83 @@ func TestDeleteProxiedReturnsInvalidRequestErrorBeforeAddingAuth(t *testing.T) {
 		t.Fatal("expected invalid request error")
 	}
 }
+
+// TestRetriedFetchChunkDataRetriesFreshUrlsImmediately covers the case the
+// refresh hook exists for: every location the caller knew about is gone, and
+// the data is live somewhere the caller has not heard of yet. The read must
+// land on the fresh location without first sitting through the backoff ladder.
+func TestRetriedFetchChunkDataRetriesFreshUrlsImmediately(t *testing.T) {
+	payload := []byte("chunk contents")
+	live := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Write(payload)
+	}))
+	defer live.Close()
+
+	// A port nothing listens on: the address is well formed, the dial fails.
+	dead := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
+	deadURL := dead.URL
+	dead.Close()
+
+	refreshed := 0
+	buffer := make([]byte, len(payload))
+	start := time.Now()
+	n, err := RetriedFetchChunkData(context.Background(), buffer, []string{deadURL + "/3,abc"}, nil, false, true, 0, "3,abc",
+		func() []string {
+			refreshed++
+			return []string{live.URL + "/3,abc"}
+		})
+	elapsed := time.Since(start)
+
+	if err != nil {
+		t.Fatalf("fetch with a refreshed location: %v", err)
+	}
+	if n != len(payload) || string(buffer[:n]) != string(payload) {
+		t.Fatalf("got %q, want %q", buffer[:n], payload)
+	}
+	if refreshed != 1 {
+		t.Fatalf("refresh called %d times, want exactly 1", refreshed)
+	}
+	// The first backoff is a full second; landing well under it is the point.
+	if elapsed > 500*time.Millisecond {
+		t.Fatalf("took %v, expected the retry to skip the backoff", elapsed)
+	}
+}
+
+// TestRetriedFetchChunkDataKeepsBackoffWhenLocationsAreUnchanged makes sure a
+// refresh that returns the same list is treated as "the locations were never
+// the problem" rather than as a reason to spin.
+func TestRetriedFetchChunkDataKeepsBackoffWhenLocationsAreUnchanged(t *testing.T) {
+	dead := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
+	deadURL := dead.URL
+	dead.Close()
+
+	urls := []string{deadURL + "/3,abc"}
+	refreshed := 0
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	buffer := make([]byte, 4)
+	_, err := RetriedFetchChunkData(ctx, buffer, urls, nil, false, true, 0, "3,abc", func() []string {
+		refreshed++
+		return urls
+	})
+	if err == nil {
+		t.Fatal("expected the fetch to fail against a dead location")
+	}
+	if refreshed != 1 {
+		t.Fatalf("refresh called %d times, want exactly 1", refreshed)
+	}
+}
+
+func TestSameUrlsIgnoresOrder(t *testing.T) {
+	a := []string{"http://a:8080/3,x", "http://b:8080/3,x"}
+	if !SameUrls(a, []string{a[1], a[0]}) {
+		t.Fatal("a reshuffle of the same locations should not count as fresh")
+	}
+	if SameUrls(a, []string{a[0], "http://c:8080/3,x"}) {
+		t.Fatal("a different location should count as fresh")
+	}
+	if SameUrls(a, a[:1]) {
+		t.Fatal("a shorter list should count as fresh")
+	}
+}