fix(filer): bound retained reader cache buffers by bytes

weed/filer/reader_cache.go

@@ -20,6 +20,8 @@ type CacheInvalidator interface {
 
 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)
 
+const readerCacheMemoryLimit = 64 << 20
+
 type ReaderCache struct {
 	chunkCache       chunk_cache.ChunkCache
 	lookupFileIdFn   wdclient.LookupFileIdFunctionType
@@ -177,6 +179,30 @@ func (rc *ReaderCache) UnCache(fileId string) {
 	}
 }
 
+func (rc *ReaderCache) trim() {
+	rc.Lock()
+	defer rc.Unlock()
+
+	for {
+		var retained int64
+		var oldest *SingleChunkCacher
+		for _, downloader := range rc.downloaders {
+			downloader.Lock()
+			size := cap(downloader.data)
+			downloader.Unlock()
+			retained += int64(size)
+			if size > 0 && (oldest == nil || atomic.LoadInt64(&downloader.completedTimeNew) < atomic.LoadInt64(&oldest.completedTimeNew)) {
+				oldest = downloader
+			}
+		}
+		if retained <= readerCacheMemoryLimit {
+			return
+		}
+		delete(rc.downloaders, oldest.chunkFileId)
+		oldest.destroy()
+	}
+}
+
 func (rc *ReaderCache) destroy() {
 	rc.Lock()
 	defer rc.Unlock()
@@ -200,27 +226,19 @@ func newSingleChunkCacher(parent *ReaderCache, fileId string, cipherKey []byte,
 	}
 }
 
-// startCaching downloads the chunk data in the background.
-// It does NOT hold the lock during the HTTP download to allow concurrent readers
-// to wait efficiently using the done channel.
-//
-// Concurrent downloads of the same chunk are already deduplicated by the
-// ReaderCache.downloaders map (guarded by the ReaderCache mutex). Each fileId
-// has at most one active SingleChunkCacher at any time.
+// startCaching downloads a chunk shared by concurrent readers.
 func (s *SingleChunkCacher) startCaching() {
 	s.wg.Add(1)
-	defer s.wg.Done()
-	defer close(s.done) // guarantee completion signal even on panic
+	defer func() {
+		close(s.done)
+		s.wg.Done()
+		// Release download waiters before eviction waits for active readers.
+		s.parent.trim()
+	}()
 
 	s.cacheStartedCh <- struct{}{} // signal that we've started
 
-	// Note: We intentionally use context.Background() here, NOT a request-specific context.
-	// The downloaded chunk is a shared resource - multiple concurrent readers may be waiting
-	// for this same download to complete. If we used a request context and that request was
-	// cancelled, it would abort the download and cause errors for all other waiting readers.
-	// The download should always complete once started to serve all potential consumers.
-
-	// Lookup file ID without holding the lock
+	// Request cancellation must not abort a download shared by other readers.
 	urlStrings, err := s.parent.lookupFileIdFn(context.Background(), s.chunkFileId)
 	if err != nil {
 		s.setError(fmt.Errorf("operation LookupFileId %s failed, err: %v", s.chunkFileId, err))

weed/filer/reader_cache_memory_test.go

@@ -0,0 +1,103 @@
+package filer
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
+)
+
+func TestChunkGroupReaderCacheMemory(t *testing.T) {
+	for _, tt := range []struct {
+		name      string
+		chunkSize int
+		chunks    int
+		mode      string
+	}{
+		{"default chunks", 2 << 20, 40, "sequential"},
+		{"pooled buffers", 3 << 20, 20, "sequential"},
+		{"oversized chunk", 65 << 20, 1, "sequential"},
+		{"concurrent readers", 2 << 20, 40, "concurrent"},
+		{"prefetch", 2 << 20, 40, "prefetch"},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			group, err := NewChunkGroup(func(context.Context, string) ([]string, error) {
+				return []string{"unused"}, nil
+			}, newMockChunkCacheForReaderCache(), nil, 128, nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			rc := group.readerCache
+			defer rc.destroy()
+			started := make(chan struct{}, tt.chunks)
+			gate := make(chan struct{})
+			if tt.mode == "sequential" {
+				close(gate)
+			}
+			rc.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, _ string, _ util_http.RefreshUrlsFunc) (int, error) {
+				started <- struct{}{}
+				<-gate
+				buffer[0] = 42
+				return len(buffer), nil
+			}
+			var readers sync.WaitGroup
+			var views *Interval[*ChunkView]
+			for i := 0; i < tt.chunks; i++ {
+				read := func() {
+					buffer := make([]byte, 1)
+					n, err := rc.ReadChunkAt(context.Background(), buffer, fmt.Sprint(i), nil, false, 0, tt.chunkSize, false)
+					if err != nil || n != 1 || buffer[0] != 42 {
+						t.Errorf("read %d: n=%d, data=%v, err=%v", i, n, buffer, err)
+					}
+				}
+				switch tt.mode {
+				case "sequential":
+					read()
+				case "concurrent":
+					readers.Add(1)
+					go func() { defer readers.Done(); read() }()
+				case "prefetch":
+					views = &Interval[*ChunkView]{Value: &ChunkView{FileId: fmt.Sprint(i), ChunkSize: uint64(tt.chunkSize)}, Next: views}
+				}
+			}
+			if tt.mode == "prefetch" {
+				rc.MaybeCache(views, tt.chunks)
+			}
+			if tt.mode != "sequential" {
+				for i := 0; i < tt.chunks; i++ {
+					<-started
+				}
+				close(gate)
+				readers.Wait()
+			}
+
+			deadline := time.Now().Add(time.Second)
+			for {
+				rc.Lock()
+				retained := 0
+				completed := true
+				for _, downloader := range rc.downloaders {
+					select {
+					case <-downloader.done:
+					default:
+						completed = false
+					}
+					downloader.Lock()
+					retained += cap(downloader.data)
+					downloader.Unlock()
+				}
+				rc.Unlock()
+				if completed && retained <= 64<<20 {
+					break
+				}
+				if time.Now().After(deadline) {
+					t.Fatalf("reader cache retained %d MiB, want at most 64 MiB", retained>>20)
+				}
+				time.Sleep(time.Millisecond)
+			}
+		})
+	}
+}