s3: invalidate stale reader cache locations on chunk read failure (#10156)

* s3: invalidate stale reader cache locations on chunk read failure

* filer: share the chunk-read self-heal across reader cache and streaming paths

The reader cache retry added a third copy of the invalidate-relookup-compare-retry
dance already inlined in PrepareStreamContentWithThrottler and duplicated in
retryWithCacheInvalidation. Extract retryFetchWithFreshLocations and route all
three through it, parameterized by the refetch primitive.

* filer: drop redundant completedTimeNew store in reader cache success path

startCaching already stamps completedTimeNew unconditionally before the
fetchErr branch; the second store inside the success branch is dead.

* filer: make NewReaderCache cache invalidator an explicit parameter

The variadic ...CacheInvalidator only ever read the first element, so a caller
could pass two and silently get one. Take a single explicit argument and have
the non-S3 callers pass nil.

* filer: inject reader cache chunk fetch as a struct field

Replace the process-global readerCacheFetchChunkData test seam with a
per-instance fetchChunkDataFn field defaulted in NewReaderCache, matching how
lookupFileIdFn is already wired. Tests set the field on the cache instead of
swapping a shared global.

* filer: log the location count, not full URLs, on self-heal retry

---------

Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/filer/filechunk_group.go

@@ -41,7 +41,7 @@ func NewChunkGroup(lookupFn wdclient.LookupFileIdFunctionType, chunkCache chunk_
 	group := &ChunkGroup{
 		lookupFn:          lookupFn,
 		sections:          make(map[SectionIndex]*FileChunkSection),
-		readerCache:       NewReaderCache(readerCacheLimit, chunkCache, lookupFn),
+		readerCache:       NewReaderCache(readerCacheLimit, chunkCache, lookupFn, nil),
 		concurrentReaders: concurrentReaders,
 	}
 

weed/filer/reader_at_test.go

@@ -77,7 +77,7 @@ func TestReaderAt(t *testing.T) {
 	readerAt := &ChunkReadAt{
 		chunkViews:    ViewFromVisibleIntervals(visibles, 0, math.MaxInt64),
 		fileSize:      10,
-		readerCache:   NewReaderCache(3, &mockChunkCache{}, nil),
+		readerCache:   NewReaderCache(3, &mockChunkCache{}, nil, nil),
 		readerPattern: NewReaderPattern(),
 	}
 
@@ -124,7 +124,7 @@ func TestReaderAt0(t *testing.T) {
 	readerAt := &ChunkReadAt{
 		chunkViews:    ViewFromVisibleIntervals(visibles, 0, math.MaxInt64),
 		fileSize:      10,
-		readerCache:   NewReaderCache(3, &mockChunkCache{}, nil),
+		readerCache:   NewReaderCache(3, &mockChunkCache{}, nil, nil),
 		readerPattern: NewReaderPattern(),
 	}
 
@@ -150,7 +150,7 @@ func TestReaderAt1(t *testing.T) {
 	readerAt := &ChunkReadAt{
 		chunkViews:    ViewFromVisibleIntervals(visibles, 0, math.MaxInt64),
 		fileSize:      20,
-		readerCache:   NewReaderCache(3, &mockChunkCache{}, nil),
+		readerCache:   NewReaderCache(3, &mockChunkCache{}, nil, nil),
 		readerPattern: NewReaderPattern(),
 	}
 
@@ -183,7 +183,7 @@ func TestReaderAtGappedChunksDoNotLeak(t *testing.T) {
 	readerAt := &ChunkReadAt{
 		chunkViews:    ViewFromVisibleIntervals(visibles, 0, math.MaxInt64),
 		fileSize:      9,
-		readerCache:   NewReaderCache(3, &mockChunkCache{}, nil),
+		readerCache:   NewReaderCache(3, &mockChunkCache{}, nil, nil),
 		readerPattern: NewReaderPattern(),
 	}
 
@@ -195,7 +195,7 @@ func TestReaderAtSparseFileDoesNotLeak(t *testing.T) {
 	readerAt := &ChunkReadAt{
 		chunkViews:    ViewFromVisibleIntervals(NewIntervalList[*VisibleInterval](), 0, math.MaxInt64),
 		fileSize:      3,
-		readerCache:   NewReaderCache(3, &mockChunkCache{}, nil),
+		readerCache:   NewReaderCache(3, &mockChunkCache{}, nil, nil),
 		readerPattern: NewReaderPattern(),
 	}
 

weed/filer/reader_cache.go

@@ -14,9 +14,17 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
 )
 
+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 ReaderCache struct {
-	chunkCache     chunk_cache.ChunkCache
-	lookupFileIdFn wdclient.LookupFileIdFunctionType
+	chunkCache       chunk_cache.ChunkCache
+	lookupFileIdFn   wdclient.LookupFileIdFunctionType
+	cacheInvalidator CacheInvalidator
+	fetchChunkDataFn fetchChunkDataFnType
 	sync.Mutex
 	downloaders map[string]*SingleChunkCacher
 	limit       int
@@ -38,12 +46,14 @@ type SingleChunkCacher struct {
 	done           chan struct{} // signals when download is complete
 }
 
-func NewReaderCache(limit int, chunkCache chunk_cache.ChunkCache, lookupFileIdFn wdclient.LookupFileIdFunctionType) *ReaderCache {
+func NewReaderCache(limit int, chunkCache chunk_cache.ChunkCache, lookupFileIdFn wdclient.LookupFileIdFunctionType, cacheInvalidator CacheInvalidator) *ReaderCache {
 	return &ReaderCache{
-		limit:          limit,
-		chunkCache:     chunkCache,
-		lookupFileIdFn: lookupFileIdFn,
-		downloaders:    make(map[string]*SingleChunkCacher),
+		limit:            limit,
+		chunkCache:       chunkCache,
+		lookupFileIdFn:   lookupFileIdFn,
+		cacheInvalidator: cacheInvalidator,
+		fetchChunkDataFn: util_http.RetriedFetchChunkData,
+		downloaders:      make(map[string]*SingleChunkCacher),
 	}
 }
 
@@ -97,15 +107,25 @@ func (rc *ReaderCache) MaybeCache(chunkViews *Interval[*ChunkView], count int) {
 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()
 
-	if cacher, found := rc.downloaders[fileId]; found {
-		rc.Unlock()
-		n, err := cacher.readChunkAt(ctx, buffer, offset)
-		if n > 0 || err != nil {
-			return n, err
+	for {
+		if cacher, found := rc.downloaders[fileId]; found {
+			if cacher.hasCompletedError() {
+				delete(rc.downloaders, fileId)
+				rc.Unlock()
+				cacher.destroy()
+				rc.Lock()
+				continue
+			}
+			rc.Unlock()
+			n, err := cacher.readChunkAt(ctx, buffer, offset)
+			if n > 0 || err != nil {
+				return n, err
+			}
+			// If n=0 and err=nil, the cacher couldn't provide data for this offset.
+			// Fall through to try chunkCache.
+			rc.Lock()
 		}
-		// If n=0 and err=nil, the cacher couldn't provide data for this offset.
-		// Fall through to try chunkCache.
-		rc.Lock()
+		break
 	}
 	if shouldCache || rc.lookupFileIdFn == nil {
 		n, err := rc.chunkCache.ReadChunkAt(buffer, fileId, uint64(offset))
@@ -198,32 +218,74 @@ func (s *SingleChunkCacher) startCaching() {
 	// Lookup file ID without holding the lock
 	urlStrings, err := s.parent.lookupFileIdFn(context.Background(), s.chunkFileId)
 	if err != nil {
-		s.Lock()
-		s.err = fmt.Errorf("operation LookupFileId %s failed, err: %v", s.chunkFileId, err)
-		s.Unlock()
+		s.setError(fmt.Errorf("operation LookupFileId %s failed, err: %v", s.chunkFileId, err))
+		return
+	}
+	if len(urlStrings) == 0 {
+		s.setError(fmt.Errorf("operation LookupFileId %s failed, err: urls not found", s.chunkFileId))
 		return
 	}
 
-	// Allocate buffer and download without holding the lock
-	// This allows multiple downloads to proceed in parallel
-	data := mem.Allocate(s.chunkSize)
-	_, fetchErr := util_http.RetriedFetchChunkData(context.Background(), data, urlStrings, s.cipherKey, s.isGzipped, true, 0, s.chunkFileId)
+	data, fetchErr := s.fetchChunkData(context.Background(), urlStrings)
+	if fetchErr != nil {
+		data, fetchErr = s.retryFetchAfterCacheInvalidation(context.Background(), urlStrings, fetchErr)
+	}
 
 	// Now acquire lock to update state
 	s.Lock()
+	atomic.StoreInt64(&s.completedTimeNew, time.Now().UnixNano())
 	if fetchErr != nil {
-		mem.Free(data)
 		s.err = fetchErr
 	} else {
 		s.data = data
 		if s.shouldCache {
 			s.parent.chunkCache.SetChunk(s.chunkFileId, s.data)
 		}
-		atomic.StoreInt64(&s.completedTimeNew, time.Now().UnixNano())
 	}
 	s.Unlock()
 }
 
+func (s *SingleChunkCacher) setError(err error) {
+	s.Lock()
+	defer s.Unlock()
+	s.err = err
+	atomic.StoreInt64(&s.completedTimeNew, time.Now().UnixNano())
+}
+
+func (s *SingleChunkCacher) hasCompletedError() bool {
+	if atomic.LoadInt64(&s.completedTimeNew) == 0 {
+		return false
+	}
+	s.Lock()
+	defer s.Unlock()
+	return s.err != nil
+}
+
+func (s *SingleChunkCacher) fetchChunkData(ctx context.Context, urlStrings []string) ([]byte, error) {
+	// 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)
+	if fetchErr != nil {
+		mem.Free(data)
+		return nil, fetchErr
+	}
+	return data, nil
+}
+
+func (s *SingleChunkCacher) retryFetchAfterCacheInvalidation(ctx context.Context, oldUrlStrings []string, originalErr error) ([]byte, error) {
+	var data []byte
+	err := retryFetchWithFreshLocations(ctx, s.parent.cacheInvalidator, s.parent.lookupFileIdFn, s.chunkFileId, oldUrlStrings, originalErr, func(newUrls []string) error {
+		var fetchErr error
+		data, fetchErr = s.fetchChunkData(ctx, newUrls)
+		return fetchErr
+	})
+	if err != nil {
+		return nil, err
+	}
+	return data, nil
+}
+
 func (s *SingleChunkCacher) destroy() {
 	// wait for all reads to finish before destroying the data
 	s.wg.Wait()

weed/filer/reader_cache_test.go

@@ -16,6 +16,25 @@ type mockChunkCacheForReaderCache struct {
 	mu       sync.Mutex
 }
 
+type mockCacheInvalidatorForReaderCache struct {
+	calls  int32
+	mu     sync.Mutex
+	fileId string
+}
+
+func (m *mockCacheInvalidatorForReaderCache) InvalidateCache(fileId string) {
+	atomic.AddInt32(&m.calls, 1)
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.fileId = fileId
+}
+
+func (m *mockCacheInvalidatorForReaderCache) lastFileId() string {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.fileId
+}
+
 func newMockChunkCacheForReaderCache() *mockChunkCacheForReaderCache {
 	return &mockChunkCacheForReaderCache{
 		data: make(map[string][]byte),
@@ -60,6 +79,113 @@ func (m *mockChunkCacheForReaderCache) IsInCache(fileId string, lockNeeded bool)
 	return ok
 }
 
+func TestReaderCacheRetryAfterCacheInvalidation(t *testing.T) {
+	cache := newMockChunkCacheForReaderCache()
+	invalidator := &mockCacheInvalidatorForReaderCache{}
+	fileId := "425141,ef8914e9bdbbe8cb6838191f"
+	staleUrl := "http://fast-volume-1/" + fileId
+	freshUrl := "http://fast-volume-9/" + fileId
+	testData := []byte("fresh chunk data after cache invalidation")
+
+	var lookupCount int32
+	lookupFn := func(ctx context.Context, requestedFileId string) ([]string, error) {
+		if requestedFileId != fileId {
+			return nil, fmt.Errorf("unexpected lookup file id %s", requestedFileId)
+		}
+		atomic.AddInt32(&lookupCount, 1)
+		if atomic.LoadInt32(&invalidator.calls) == 0 {
+			return []string{staleUrl}, nil
+		}
+		return []string{freshUrl}, nil
+	}
+
+	var fetchCount int32
+	fetchFn := func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, requestedFileId string) (int, error) {
+		if requestedFileId != fileId {
+			return 0, fmt.Errorf("unexpected fetch file id %s", requestedFileId)
+		}
+		switch atomic.AddInt32(&fetchCount, 1) {
+		case 1:
+			if len(urlStrings) != 1 || urlStrings[0] != staleUrl {
+				return 0, fmt.Errorf("first fetch should use stale url %v", urlStrings)
+			}
+			return 0, fmt.Errorf("404 Not Found: not found")
+		case 2:
+			if len(urlStrings) != 1 || urlStrings[0] != freshUrl {
+				return 0, fmt.Errorf("retry fetch should use fresh url %v", urlStrings)
+			}
+			return copy(buffer, testData), nil
+		default:
+			return 0, fmt.Errorf("unexpected extra fetch with urls %v", urlStrings)
+		}
+	}
+
+	rc := NewReaderCache(10, cache, lookupFn, invalidator)
+	rc.fetchChunkDataFn = fetchFn
+	defer rc.destroy()
+
+	buffer := make([]byte, len(testData))
+	n, err := rc.ReadChunkAt(context.Background(), buffer, fileId, nil, false, 0, len(testData), true)
+	if err != nil {
+		t.Fatalf("expected successful retry, got %v", err)
+	}
+	if got := string(buffer[:n]); got != string(testData) {
+		t.Fatalf("expected %q, got %q", testData, got)
+	}
+	if got := atomic.LoadInt32(&invalidator.calls); got != 1 {
+		t.Fatalf("expected one cache invalidation, got %d", got)
+	}
+	if got := invalidator.lastFileId(); got != fileId {
+		t.Fatalf("expected invalidated file id %s, got %s", fileId, got)
+	}
+	if got := atomic.LoadInt32(&lookupCount); got != 2 {
+		t.Fatalf("expected lookup before fetch and after invalidation, got %d", got)
+	}
+	if got := atomic.LoadInt32(&fetchCount); got != 2 {
+		t.Fatalf("expected stale fetch and retry fetch, got %d", got)
+	}
+}
+
+func TestReaderCacheRemovesFailedDownloader(t *testing.T) {
+	cache := newMockChunkCacheForReaderCache()
+	fileId := "425141,failed"
+	url := "http://fast-volume-1/" + fileId
+
+	var lookupCount int32
+	lookupFn := func(ctx context.Context, requestedFileId string) ([]string, error) {
+		if requestedFileId != fileId {
+			return nil, fmt.Errorf("unexpected lookup file id %s", requestedFileId)
+		}
+		atomic.AddInt32(&lookupCount, 1)
+		return []string{url}, nil
+	}
+
+	var fetchCount int32
+	fetchFn := func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, requestedFileId string) (int, error) {
+		atomic.AddInt32(&fetchCount, 1)
+		return 0, fmt.Errorf("fetch failed")
+	}
+
+	rc := NewReaderCache(10, cache, lookupFn, nil)
+	rc.fetchChunkDataFn = fetchFn
+	defer rc.destroy()
+
+	for i := 0; i < 2; i++ {
+		buffer := make([]byte, 8)
+		_, err := rc.ReadChunkAt(context.Background(), buffer, fileId, nil, false, 0, len(buffer), true)
+		if err == nil {
+			t.Fatalf("read %d should fail", i+1)
+		}
+	}
+
+	if got := atomic.LoadInt32(&lookupCount); got != 2 {
+		t.Fatalf("failed downloader should be removed so the second read re-lookups, got %d lookups", got)
+	}
+	if got := atomic.LoadInt32(&fetchCount); got != 2 {
+		t.Fatalf("failed downloader should be removed so the second read refetches, got %d fetches", got)
+	}
+}
+
 // TestReaderCacheContextCancellation tests that a reader can cancel its wait
 // while the download continues for other readers
 func TestReaderCacheContextCancellation(t *testing.T) {
@@ -67,7 +193,7 @@ func TestReaderCacheContextCancellation(t *testing.T) {
 
 	// Create a ReaderCache - we can't easily test the full flow without mocking HTTP,
 	// but we can test the context cancellation in readChunkAt
-	rc := NewReaderCache(10, cache, nil)
+	rc := NewReaderCache(10, cache, nil, nil)
 	defer rc.destroy()
 
 	// Pre-populate cache to avoid HTTP calls
@@ -106,7 +232,7 @@ func TestReaderCacheFallbackToChunkCache(t *testing.T) {
 	testData := []byte("fallback test data that should be found in chunk cache")
 	cache.SetChunk("fallback-file", testData)
 
-	rc := NewReaderCache(10, cache, nil)
+	rc := NewReaderCache(10, cache, nil, nil)
 	defer rc.destroy()
 
 	// Read should hit the chunk cache
@@ -138,7 +264,7 @@ func TestReaderCacheMultipleReadersWaitForSameChunk(t *testing.T) {
 	}
 	cache.SetChunk("shared-chunk", testData)
 
-	rc := NewReaderCache(10, cache, nil)
+	rc := NewReaderCache(10, cache, nil, nil)
 	defer rc.destroy()
 
 	// Launch multiple concurrent readers for the same chunk
@@ -184,7 +310,7 @@ func TestReaderCachePartialRead(t *testing.T) {
 	testData := []byte("0123456789ABCDEFGHIJ")
 	cache.SetChunk("partial-read-file", testData)
 
-	rc := NewReaderCache(10, cache, nil)
+	rc := NewReaderCache(10, cache, nil, nil)
 	defer rc.destroy()
 
 	tests := []struct {
@@ -222,7 +348,7 @@ func TestReaderCacheCleanup(t *testing.T) {
 	cache := newMockChunkCacheForReaderCache()
 
 	// Create cache with limit of 3
-	rc := NewReaderCache(3, cache, nil)
+	rc := NewReaderCache(3, cache, nil, nil)
 	defer rc.destroy()
 
 	// Add data for multiple files
@@ -259,7 +385,7 @@ func TestReaderCacheCleanup(t *testing.T) {
 // TestSingleChunkCacherDoneSignal tests that done channel is always closed
 func TestSingleChunkCacherDoneSignal(t *testing.T) {
 	cache := newMockChunkCacheForReaderCache()
-	rc := NewReaderCache(10, cache, nil)
+	rc := NewReaderCache(10, cache, nil, nil)
 	defer rc.destroy()
 
 	// Test that we can read even when data is in cache (done channel should work)
@@ -317,7 +443,7 @@ func TestSingleChunkCacherLookupError(t *testing.T) {
 		return nil, fmt.Errorf("lookup failed for %s", fileId)
 	}
 
-	rc := NewReaderCache(10, cache, lookupFn)
+	rc := NewReaderCache(10, cache, lookupFn, nil)
 	defer rc.destroy()
 
 	buffer := make([]byte, 100)
@@ -343,7 +469,7 @@ func TestSingleChunkCacherContextCancellationDuringLookup(t *testing.T) {
 		return nil, fmt.Errorf("lookup completed but reader should have cancelled")
 	}
 
-	rc := NewReaderCache(10, cache, lookupFn)
+	rc := NewReaderCache(10, cache, lookupFn, nil)
 	defer rc.destroy()
 	defer close(lookupCanFinish) // Ensure cleanup
 
@@ -391,7 +517,7 @@ func TestSingleChunkCacherMultipleReadersWaitForDownload(t *testing.T) {
 		return nil, fmt.Errorf("simulated lookup error")
 	}
 
-	rc := NewReaderCache(10, cache, lookupFn)
+	rc := NewReaderCache(10, cache, lookupFn, nil)
 	defer rc.destroy()
 
 	numReaders := 5
@@ -438,17 +564,25 @@ func TestSingleChunkCacherMultipleReadersWaitForDownload(t *testing.T) {
 // the downloaders map deduplicates in-flight downloads.
 func TestReaderCacheDownloaderDedup(t *testing.T) {
 	cache := newMockChunkCacheForReaderCache()
+
 	var lookupCount int32
-	lookupGate := make(chan struct{})
+	var fetchCount int32
+	fetchGate := make(chan struct{})
+	testData := []byte("deduplicated data")
 
 	lookupFn := func(ctx context.Context, fileId string) ([]string, error) {
 		atomic.AddInt32(&lookupCount, 1)
-		<-lookupGate
-		// Return an error so we don't need to mock the HTTP fetch.
-		return nil, fmt.Errorf("simulated lookup for %s", fileId)
+		return []string{"http://volume/" + fileId}, nil
 	}
 
-	rc := NewReaderCache(10, cache, lookupFn)
+	fetchFn := func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string) (int, error) {
+		atomic.AddInt32(&fetchCount, 1)
+		<-fetchGate
+		return copy(buffer, testData), nil
+	}
+
+	rc := NewReaderCache(10, cache, lookupFn, nil)
+	rc.fetchChunkDataFn = fetchFn
 	defer rc.destroy()
 
 	const numReaders = 10
@@ -464,13 +598,15 @@ func TestReaderCacheDownloaderDedup(t *testing.T) {
 	}
 
 	// Allow downloads to proceed.
-	close(lookupGate)
+	close(fetchGate)
 	wg.Wait()
 
-	count := atomic.LoadInt32(&lookupCount)
-	if count != 1 {
+	if count := atomic.LoadInt32(&lookupCount); count != 1 {
 		t.Errorf("expected exactly 1 lookup call, got %d", count)
 	}
+	if count := atomic.LoadInt32(&fetchCount); count != 1 {
+		t.Errorf("expected exactly 1 fetch call, got %d", count)
+	}
 }
 
 // TestSingleChunkCacherOneReaderCancelsOthersContinue tests that when one reader
@@ -487,7 +623,7 @@ func TestSingleChunkCacherOneReaderCancelsOthersContinue(t *testing.T) {
 		return nil, fmt.Errorf("simulated error after delay")
 	}
 
-	rc := NewReaderCache(10, cache, lookupFn)
+	rc := NewReaderCache(10, cache, lookupFn, nil)
 	defer rc.destroy()
 
 	cancelledReaderDone := make(chan error, 1)

weed/filer/stream.go

@@ -102,10 +102,6 @@ func PrepareStreamContent(masterClient wdclient.HasLookupFileIdFunction, jwtFunc
 
 type VolumeServerJwtFunction func(fileId string) string
 
-type CacheInvalidator interface {
-	InvalidateCache(fileId string)
-}
-
 // urlSlicesEqual checks if two URL slices contain the same URLs (order-independent)
 func urlSlicesEqual(a, b []string) bool {
 	if len(a) != len(b) {
@@ -126,6 +122,36 @@ func urlSlicesEqual(a, b []string) bool {
 	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
+// is returned unchanged when no retry is attempted, so callers surface the real fetch failure.
+func retryFetchWithFreshLocations(ctx context.Context, invalidator CacheInvalidator, lookupFn wdclient.LookupFileIdFunctionType, fileId string, oldUrls []string, originalErr error, refetch func(newUrls []string) error) error {
+	if invalidator == nil {
+		return originalErr
+	}
+
+	glog.V(0).InfofCtx(ctx, "read chunk %s failed, invalidating cache and retrying: %v", fileId, originalErr)
+	invalidator.InvalidateCache(fileId)
+
+	newUrls, lookupErr := lookupFn(ctx, fileId)
+	if lookupErr != nil {
+		glog.WarningfCtx(ctx, "failed to re-lookup chunk %s after cache invalidation: %v", fileId, lookupErr)
+		return fmt.Errorf("re-lookup chunk %s after cache invalidation: %w", fileId, lookupErr)
+	}
+	if len(newUrls) == 0 {
+		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) {
+		glog.V(0).InfofCtx(ctx, "re-lookup returned same locations for chunk %s, skipping retry", fileId)
+		return originalErr
+	}
+
+	glog.V(0).InfofCtx(ctx, "retrying read chunk %s with %d new locations", fileId, len(newUrls))
+	return refetch(newUrls)
+}
+
 func PrepareStreamContentWithThrottler(ctx context.Context, masterClient wdclient.HasLookupFileIdFunction, jwtFunc VolumeServerJwtFunction, chunks []*filer_pb.FileChunk, offset int64, size int64, downloadMaxBytesPs int64) (DoStreamContent, error) {
 	glog.V(4).InfofCtx(ctx, "prepare to stream content for chunks: %d", len(chunks))
 	chunkViews := ViewFromChunks(ctx, masterClient.GetLookupFileIdFunction(), chunks, offset, size)
@@ -195,32 +221,15 @@ func PrepareStreamContentWithThrottler(ctx context.Context, masterClient wdclien
 
 			// If read failed, try to invalidate cache and re-lookup
 			if err != nil && written == 0 {
-				if invalidator, ok := masterClient.(CacheInvalidator); ok {
-					glog.V(0).InfofCtx(ctx, "read chunk %s failed, invalidating cache and retrying", chunkView.FileId)
-					invalidator.InvalidateCache(chunkView.FileId)
-
-					// Re-lookup
-					newUrlStrings, lookupErr := masterClient.GetLookupFileIdFunction()(ctx, chunkView.FileId)
-					if lookupErr == nil && len(newUrlStrings) > 0 {
-						// Check if new URLs are different from old ones to avoid infinite retry
-						if !urlSlicesEqual(urlStrings, newUrlStrings) {
-							glog.V(0).InfofCtx(ctx, "retrying read chunk %s with new locations: %v", chunkView.FileId, newUrlStrings)
-							_, err = retriedStreamFetchChunkData(ctx, writer, newUrlStrings, jwt, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.OffsetInChunk, int(chunkView.ViewSize))
-							// Update the map so subsequent references use fresh URLs
-							if err == nil {
-								fileId2Url[chunkView.FileId] = newUrlStrings
-							}
-						} else {
-							glog.V(0).InfofCtx(ctx, "re-lookup returned same locations for chunk %s, skipping retry", chunkView.FileId)
-						}
-					} else {
-						if lookupErr != nil {
-							glog.WarningfCtx(ctx, "failed to re-lookup chunk %s after cache invalidation: %v", chunkView.FileId, lookupErr)
-						} else {
-							glog.WarningfCtx(ctx, "re-lookup for chunk %s returned no locations, skipping retry", chunkView.FileId)
-						}
+				invalidator, _ := masterClient.(CacheInvalidator)
+				err = retryFetchWithFreshLocations(ctx, invalidator, masterClient.GetLookupFileIdFunction(), chunkView.FileId, urlStrings, err, func(newUrls []string) error {
+					_, refetchErr := retriedStreamFetchChunkData(ctx, writer, newUrls, jwt, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.OffsetInChunk, int(chunkView.ViewSize))
+					if refetchErr == nil {
+						// Update the map so subsequent references use fresh URLs
+						fileId2Url[chunkView.FileId] = newUrls
 					}
-				}
+					return refetchErr
+				})
 			}
 
 			offset += int64(chunkView.ViewSize)

weed/filer/stream_prefetch.go

@@ -177,7 +177,7 @@ func streamChunksPrefetched(
 				consumeErr = err
 				break
 			}
-			retryErr := retryWithCacheInvalidation(localCtx, writer, chunkView, result.urlStrings, jwtFunc, masterClient)
+			retryErr := retryWithCacheInvalidation(localCtx, writer, chunkView, result.urlStrings, result.fetchErr, jwtFunc, masterClient)
 			if retryErr != nil {
 				stats.FilerHandlerCounter.WithLabelValues("chunkDownloadError").Inc()
 				consumeErr = fmt.Errorf("read chunk: %w", retryErr)
@@ -230,45 +230,25 @@ func streamChunksPrefetched(
 	return nil
 }
 
-// retryWithCacheInvalidation attempts to re-fetch a chunk after invalidating the URL cache.
-// This mirrors the retry logic in PrepareStreamContentWithThrottler's sequential path.
+// retryWithCacheInvalidation re-fetches a chunk via the shared location self-heal after the
+// initial fetch failed with originalErr.
 func retryWithCacheInvalidation(
 	ctx context.Context,
 	writer io.Writer,
 	chunkView *ChunkView,
 	oldUrlStrings []string,
+	originalErr error,
 	jwtFunc VolumeServerJwtFunction,
 	masterClient wdclient.HasLookupFileIdFunction,
 ) error {
-	invalidator, ok := masterClient.(CacheInvalidator)
-	if !ok {
-		return fmt.Errorf("read chunk %s failed and no cache invalidator available", chunkView.FileId)
-	}
-
-	glog.V(0).InfofCtx(ctx, "prefetch read chunk %s failed, invalidating cache and retrying", chunkView.FileId)
-	invalidator.InvalidateCache(chunkView.FileId)
-
-	newUrlStrings, lookupErr := masterClient.GetLookupFileIdFunction()(ctx, chunkView.FileId)
-	if lookupErr != nil {
-		glog.WarningfCtx(ctx, "failed to re-lookup chunk %s after cache invalidation: %v", chunkView.FileId, lookupErr)
-		return fmt.Errorf("re-lookup chunk %s: %w", chunkView.FileId, lookupErr)
-	}
-	if len(newUrlStrings) == 0 {
-		glog.WarningfCtx(ctx, "re-lookup for chunk %s returned no locations, skipping retry", chunkView.FileId)
-		return fmt.Errorf("re-lookup chunk %s: no locations", chunkView.FileId)
-	}
-
-	if urlSlicesEqual(oldUrlStrings, newUrlStrings) {
-		glog.V(0).InfofCtx(ctx, "re-lookup returned same locations for chunk %s, skipping retry", chunkView.FileId)
-		return fmt.Errorf("read chunk %s failed, same locations after cache invalidation", chunkView.FileId)
-	}
-
-	glog.V(0).InfofCtx(ctx, "retrying read chunk %s with new locations: %v", chunkView.FileId, newUrlStrings)
-	jwt := jwtFunc(chunkView.FileId)
-	_, err := retriedStreamFetchChunkData(
-		ctx, writer, newUrlStrings, jwt,
-		chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(),
-		chunkView.OffsetInChunk, int(chunkView.ViewSize),
-	)
-	return err
+	invalidator, _ := masterClient.(CacheInvalidator)
+	return retryFetchWithFreshLocations(ctx, invalidator, masterClient.GetLookupFileIdFunction(), chunkView.FileId, oldUrlStrings, originalErr, func(newUrls []string) error {
+		jwt := jwtFunc(chunkView.FileId)
+		_, err := retriedStreamFetchChunkData(
+			ctx, writer, newUrls, jwt,
+			chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(),
+			chunkView.OffsetInChunk, int(chunkView.ViewSize),
+		)
+		return err
+	})
 }

weed/mq/logstore/read_parquet_to_log.go

@@ -100,7 +100,7 @@ func GenParquetReadFunc(filerClient filer_pb.FilerClient, t topic.Topic, p topic
 		fileSize := filer.FileSize(entry)
 		visibleIntervals, _ := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, entry.Chunks, 0, int64(fileSize))
 		chunkViews := filer.ViewFromVisibleIntervals(visibleIntervals, 0, int64(fileSize))
-		readerCache := filer.NewReaderCache(32, chunkCache, lookupFileIdFn)
+		readerCache := filer.NewReaderCache(32, chunkCache, lookupFileIdFn, nil)
 		readerAt := filer.NewChunkReaderAtFromClient(context.Background(), readerCache, chunkViews, int64(fileSize), filer.DefaultPrefetchCount)
 
 		// create parquet reader

weed/query/engine/hybrid_message_scanner.go

@@ -1166,7 +1166,7 @@ func (h *HybridMessageScanner) extractParquetFileStats(entry *filer_pb.Entry, lo
 	fileSize := filer.FileSize(entry)
 	visibleIntervals, _ := filer.NonOverlappingVisibleIntervals(context.Background(), lookupFileIdFn, entry.Chunks, 0, int64(fileSize))
 	chunkViews := filer.ViewFromVisibleIntervals(visibleIntervals, 0, int64(fileSize))
-	readerCache := filer.NewReaderCache(32, chunkCache, lookupFileIdFn)
+	readerCache := filer.NewReaderCache(32, chunkCache, lookupFileIdFn, nil)
 	readerAt := filer.NewChunkReaderAtFromClient(context.Background(), readerCache, chunkViews, int64(fileSize), filer.DefaultPrefetchCount)
 
 	// Create parquet reader - this only reads metadata, not data

weed/s3api/s3api_server.go

@@ -277,7 +277,7 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl
 	} else {
 		chunkCache = (*chunk_cache.TieredChunkCache)(nil)
 	}
-	readerCache := filer.NewReaderCache(s3ReaderCacheDownloaderLimit, chunkCache, filerClient.GetLookupFileIdFunction())
+	readerCache := filer.NewReaderCache(s3ReaderCacheDownloaderLimit, chunkCache, filerClient.GetLookupFileIdFunction(), filerClient)
 
 	s3ApiServer = &S3ApiServer{
 		option:                option,

weed/server/webdav_server.go

@@ -137,7 +137,7 @@ func NewWebDavFileSystem(option *WebDavOption) (webdav.FileSystem, error) {
 		chunkCache: chunkCache,
 		signature:  util.RandomInt32(),
 	}
-	t.readerCache = filer.NewReaderCache(32, chunkCache, filer.LookupFn(t))
+	t.readerCache = filer.NewReaderCache(32, chunkCache, filer.LookupFn(t), nil)
 	return t, nil
 }