diff --git a/weed/filer/filechunk_group.go b/weed/filer/filechunk_group.go index ed92e78a9..55a4ab4eb 100644 --- a/weed/filer/filechunk_group.go +++ b/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, } diff --git a/weed/filer/reader_at_test.go b/weed/filer/reader_at_test.go index 6c9041cd9..782e4b342 100644 --- a/weed/filer/reader_at_test.go +++ b/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(), } diff --git a/weed/filer/reader_cache.go b/weed/filer/reader_cache.go index c76ab4e1a..224ce8d8a 100644 --- a/weed/filer/reader_cache.go +++ b/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() diff --git a/weed/filer/reader_cache_test.go b/weed/filer/reader_cache_test.go index 8f7f6ffc5..a6771246c 100644 --- a/weed/filer/reader_cache_test.go +++ b/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) diff --git a/weed/filer/stream.go b/weed/filer/stream.go index 4acf93862..ec855be0b 100644 --- a/weed/filer/stream.go +++ b/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) diff --git a/weed/filer/stream_prefetch.go b/weed/filer/stream_prefetch.go index 511f2b90c..33d01be25 100644 --- a/weed/filer/stream_prefetch.go +++ b/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 + }) } diff --git a/weed/mq/logstore/read_parquet_to_log.go b/weed/mq/logstore/read_parquet_to_log.go index 7214965f5..87366d464 100644 --- a/weed/mq/logstore/read_parquet_to_log.go +++ b/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 diff --git a/weed/query/engine/hybrid_message_scanner.go b/weed/query/engine/hybrid_message_scanner.go index e263f781c..2b45e9cdc 100644 --- a/weed/query/engine/hybrid_message_scanner.go +++ b/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 diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index ed3431ad8..77f650b5a 100644 --- a/weed/s3api/s3api_server.go +++ b/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, diff --git a/weed/server/webdav_server.go b/weed/server/webdav_server.go index 56650ed91..3000ac8f9 100644 --- a/weed/server/webdav_server.go +++ b/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 }