mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-30 17:00:48 +01:00
Chris Lu
495e24a476
A ciphered or compressed chunk larger than readerCacheSizeMB can never be read through the whole-chunk path — the budget rejects the buffer — so its partial views must still range-fetch or the GET fails outright.
479 lines
17 KiB
Go
479 lines
17 KiB
Go
package filer
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/util/chunk_cache"
|
|
util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
|
|
)
|
|
|
|
// Two streams (e.g. two S3 GETs) reading the same object through one shared
|
|
// ReaderCache, interleaved in small slices, must share each chunk download:
|
|
// one stream finishing a chunk, or moving on to the next one, must not drop a
|
|
// buffer the other stream is still reading.
|
|
func TestChunkReadAtConcurrentStreamsShareChunks(t *testing.T) {
|
|
const chunkSize = 64 << 10
|
|
const chunkCount = 3
|
|
const sliceSize = 16 << 10
|
|
|
|
var mu sync.Mutex
|
|
fetches := map[string]int{}
|
|
rc := NewReaderCache(64, (*chunk_cache.TieredChunkCache)(nil), func(context.Context, string) ([]string, error) {
|
|
return []string{"unused"}, nil
|
|
}, nil)
|
|
defer rc.destroy()
|
|
rc.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, fileId string, _ util_http.RefreshUrlsFunc) (int, error) {
|
|
mu.Lock()
|
|
fetches[fileId]++
|
|
mu.Unlock()
|
|
for i := range buffer {
|
|
buffer[i] = fileId[len(fileId)-1]
|
|
}
|
|
return len(buffer), nil
|
|
}
|
|
|
|
newStream := func() *ChunkReadAt {
|
|
views := NewIntervalList[*ChunkView]()
|
|
for i := 0; i < chunkCount; i++ {
|
|
views.AppendInterval(&Interval[*ChunkView]{
|
|
StartOffset: int64(i * chunkSize),
|
|
StopOffset: int64((i + 1) * chunkSize),
|
|
Value: &ChunkView{
|
|
FileId: fmt.Sprintf("chunk%d", i),
|
|
ViewSize: chunkSize,
|
|
ViewOffset: int64(i * chunkSize),
|
|
ChunkSize: chunkSize,
|
|
},
|
|
})
|
|
}
|
|
return NewChunkReaderAtFromClient(context.Background(), rc, views, chunkSize*chunkCount, 0)
|
|
}
|
|
streams := []*ChunkReadAt{newStream(), newStream()}
|
|
|
|
// The streams alternate slice by slice, the second one lagging one slice
|
|
// behind, so the leader always finishes a chunk while the follower is
|
|
// still inside it.
|
|
offsets := make([]int64, len(streams))
|
|
offsets[1] = -sliceSize
|
|
for offsets[1] < chunkSize*chunkCount {
|
|
for i, stream := range streams {
|
|
if offsets[i] < 0 || offsets[i] >= chunkSize*chunkCount {
|
|
offsets[i] += sliceSize
|
|
continue
|
|
}
|
|
buf := make([]byte, sliceSize)
|
|
n, err := stream.ReadAt(buf, offsets[i])
|
|
if (err != nil && err != io.EOF) || n != sliceSize {
|
|
t.Fatalf("stream %d at %d: n=%d err=%v", i, offsets[i], n, err)
|
|
}
|
|
if want := byte('0' + offsets[i]/chunkSize); buf[0] != want || buf[n-1] != want {
|
|
t.Fatalf("stream %d at %d: got %q, want %q", i, offsets[i], buf[0], want)
|
|
}
|
|
offsets[i] += sliceSize
|
|
}
|
|
}
|
|
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
for i := 0; i < chunkCount; i++ {
|
|
fileId := fmt.Sprintf("chunk%d", i)
|
|
if fetches[fileId] != 1 {
|
|
t.Errorf("%s fetched %d times for two concurrent streams, want 1", fileId, fetches[fileId])
|
|
}
|
|
}
|
|
}
|
|
|
|
func newStreamTestReaderCache(chunkCache chunk_cache.ChunkCache) *ReaderCache {
|
|
rc := NewReaderCache(64, chunkCache, func(context.Context, string) ([]string, error) {
|
|
return []string{"unused"}, nil
|
|
}, nil)
|
|
rc.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, _ string, _ util_http.RefreshUrlsFunc) (int, error) {
|
|
return len(buffer), nil
|
|
}
|
|
return rc
|
|
}
|
|
|
|
func isRetained(rc *ReaderCache, fileId string) bool {
|
|
rc.Lock()
|
|
defer rc.Unlock()
|
|
_, found := rc.downloaders[fileId]
|
|
return found
|
|
}
|
|
|
|
// A stream moving on to a chunk served from the chunk cache must still
|
|
// release the chunk it was positioned in before.
|
|
func TestChunkStreamReleasesPreviousChunkOnCacheHit(t *testing.T) {
|
|
cache := newMockChunkCacheForReaderCache()
|
|
cache.SetChunk("chunk1", make([]byte, 4<<10))
|
|
rc := newStreamTestReaderCache(cache)
|
|
defer rc.destroy()
|
|
|
|
stream := &chunkStream{}
|
|
if _, err := rc.readChunkAt(context.Background(), stream, make([]byte, 1<<10), "chunk0", nil, false, 0, 4<<10, false); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if n, err := rc.readChunkAt(context.Background(), stream, make([]byte, 1<<10), "chunk1", nil, false, 0, 4<<10, true); err != nil || n == 0 {
|
|
t.Fatalf("cache hit read: n=%d err=%v", n, err)
|
|
}
|
|
if isRetained(rc, "chunk0") {
|
|
t.Fatal("chunk0 still retained after the stream moved on to a cached chunk")
|
|
}
|
|
}
|
|
|
|
// A chunk the stream leaves while another read is in flight must be dropped
|
|
// once that read ends, even if it did not read the chunk to the end.
|
|
func TestChunkStreamDropsLeftChunkAfterInFlightRead(t *testing.T) {
|
|
rc := newStreamTestReaderCache(newMockChunkCacheForReaderCache())
|
|
defer rc.destroy()
|
|
|
|
stream := &chunkStream{}
|
|
if _, err := rc.readChunkAt(context.Background(), stream, make([]byte, 1<<10), "chunk0", nil, false, 0, 4<<10, false); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// Another reader is in the middle of a partial read of chunk0.
|
|
rc.Lock()
|
|
other := rc.downloaders["chunk0"]
|
|
other.wg.Add(1)
|
|
atomic.AddInt32(&other.readers, 1)
|
|
rc.Unlock()
|
|
|
|
if _, err := rc.readChunkAt(context.Background(), stream, make([]byte, 1<<10), "chunk1", nil, false, 0, 4<<10, false); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !isRetained(rc, "chunk0") {
|
|
t.Fatal("chunk0 dropped while a read was still in flight")
|
|
}
|
|
|
|
// The in-flight read ends without reaching the end of the chunk.
|
|
other.wg.Done()
|
|
atomic.AddInt32(&other.readers, -1)
|
|
rc.removeConsumed(other)
|
|
if isRetained(rc, "chunk0") {
|
|
t.Fatal("chunk0 retained after the stream left it and the last read ended")
|
|
}
|
|
}
|
|
|
|
// Concurrent ReadAt calls on one ChunkReadAt (as mount does) share its stream
|
|
// pin; they must neither race on it nor leak or double-release pins.
|
|
func TestChunkStreamConcurrentReadsOnOneReader(t *testing.T) {
|
|
const chunkSize = 16 << 10
|
|
const chunkCount = 4
|
|
const sliceSize = 4 << 10
|
|
rc := newStreamTestReaderCache((*chunk_cache.TieredChunkCache)(nil))
|
|
defer rc.destroy()
|
|
|
|
views := NewIntervalList[*ChunkView]()
|
|
for i := 0; i < chunkCount; i++ {
|
|
views.AppendInterval(&Interval[*ChunkView]{
|
|
StartOffset: int64(i * chunkSize),
|
|
StopOffset: int64((i + 1) * chunkSize),
|
|
Value: &ChunkView{
|
|
FileId: fmt.Sprintf("chunk%d", i),
|
|
ViewSize: chunkSize,
|
|
ViewOffset: int64(i * chunkSize),
|
|
ChunkSize: chunkSize,
|
|
},
|
|
})
|
|
}
|
|
reader := NewChunkReaderAtFromClient(context.Background(), rc, views, chunkSize*chunkCount, 0)
|
|
|
|
var wg sync.WaitGroup
|
|
for g := 0; g < 8; g++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
for offset := int64(0); offset < chunkSize*chunkCount; offset += sliceSize {
|
|
if _, err := reader.ReadAt(make([]byte, sliceSize), offset); err != nil && err != io.EOF {
|
|
t.Error(err)
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
|
|
// Finishing the last chunk releases the stream's final pin.
|
|
if _, err := reader.ReadAt(make([]byte, sliceSize), chunkSize*chunkCount-sliceSize); err != nil && err != io.EOF {
|
|
t.Fatal(err)
|
|
}
|
|
rc.Lock()
|
|
defer rc.Unlock()
|
|
for fileId, cacher := range rc.downloaders {
|
|
t.Errorf("%s retained after all reads finished: pins=%d readers=%d", fileId, atomic.LoadInt32(&cacher.pins), atomic.LoadInt32(&cacher.readers))
|
|
}
|
|
}
|
|
|
|
type recordedFetch struct {
|
|
fileId string
|
|
isFullChunk bool
|
|
offset int64
|
|
size int
|
|
}
|
|
|
|
// fetchRecorder stubs the volume fetch and records how each chunk was
|
|
// requested: isFullChunk=false is a range fetch of just the view's slice,
|
|
// isFullChunk=true is a whole-chunk download into the shared cache.
|
|
func fetchRecorder(rc *ReaderCache) (fetches *[]recordedFetch) {
|
|
var mu sync.Mutex
|
|
recorded := &[]recordedFetch{}
|
|
rc.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, isFullChunk bool, offset int64, fileId string, _ util_http.RefreshUrlsFunc) (int, error) {
|
|
mu.Lock()
|
|
*recorded = append(*recorded, recordedFetch{fileId, isFullChunk, offset, len(buffer)})
|
|
mu.Unlock()
|
|
for i := range buffer {
|
|
buffer[i] = fileId[len(fileId)-1]
|
|
}
|
|
return len(buffer), nil
|
|
}
|
|
return recorded
|
|
}
|
|
|
|
// A reader whose views are clipped to a request window — how the S3 gateway
|
|
// builds a ranged GET — must fetch only the covered part of each chunk:
|
|
// clipped edge views take range fetches, a fully covered chunk keeps the
|
|
// shared whole-chunk path. This is what keeps a ranged GET larger than a
|
|
// buffer from multiplying volume-server reads (issue #11564), without giving
|
|
// up whole-chunk caching where the whole chunk is actually wanted.
|
|
func TestChunkReadAtClippedViewsFetchOnlyCoveredParts(t *testing.T) {
|
|
const chunkSize = 64 << 10
|
|
|
|
rc := NewReaderCache(64, (*chunk_cache.TieredChunkCache)(nil), func(context.Context, string) ([]string, error) {
|
|
return []string{"unused"}, nil
|
|
}, nil)
|
|
defer rc.destroy()
|
|
fetches := fetchRecorder(rc)
|
|
|
|
// Window [56KiB, 152KiB): tail of chunk0, all of chunk1, head of
|
|
// chunk2, head of ciphered chunk3, head of compressed chunk4 (file
|
|
// chunks need not be aligned).
|
|
views := NewIntervalList[*ChunkView]()
|
|
views.AppendInterval(&Interval[*ChunkView]{
|
|
StartOffset: chunkSize - 8<<10,
|
|
StopOffset: chunkSize,
|
|
Value: &ChunkView{FileId: "chunk0", OffsetInChunk: chunkSize - 8<<10, ViewSize: 8 << 10, ViewOffset: chunkSize - 8<<10, ChunkSize: chunkSize},
|
|
})
|
|
views.AppendInterval(&Interval[*ChunkView]{
|
|
StartOffset: chunkSize,
|
|
StopOffset: 2 * chunkSize,
|
|
Value: &ChunkView{FileId: "chunk1", ViewSize: chunkSize, ViewOffset: chunkSize, ChunkSize: chunkSize},
|
|
})
|
|
views.AppendInterval(&Interval[*ChunkView]{
|
|
StartOffset: 2 * chunkSize,
|
|
StopOffset: 2*chunkSize + 8<<10,
|
|
Value: &ChunkView{FileId: "chunk2", ViewSize: 8 << 10, ViewOffset: 2 * chunkSize, ChunkSize: chunkSize},
|
|
})
|
|
views.AppendInterval(&Interval[*ChunkView]{
|
|
StartOffset: 2*chunkSize + 8<<10,
|
|
StopOffset: 2*chunkSize + 16<<10,
|
|
Value: &ChunkView{FileId: "chunk3", ViewSize: 8 << 10, ViewOffset: 2*chunkSize + 8<<10, ChunkSize: chunkSize, CipherKey: []byte("key")},
|
|
})
|
|
views.AppendInterval(&Interval[*ChunkView]{
|
|
StartOffset: 2*chunkSize + 16<<10,
|
|
StopOffset: 2*chunkSize + 24<<10,
|
|
Value: &ChunkView{FileId: "chunk4", ViewSize: 8 << 10, ViewOffset: 2*chunkSize + 16<<10, ChunkSize: chunkSize, IsGzipped: true},
|
|
})
|
|
|
|
reader := NewChunkReaderAtFromClient(context.Background(), rc, views, 4*chunkSize, 0)
|
|
buf := make([]byte, chunkSize+32<<10)
|
|
if n, err := reader.ReadAt(buf, chunkSize-8<<10); err != nil || n != len(buf) {
|
|
t.Fatalf("window read: n=%d err=%v", n, err)
|
|
}
|
|
// buf holds [56KiB, 152KiB): chunk0's tail, chunk1, and the heads of
|
|
// chunk2, chunk3 and chunk4.
|
|
for i, b := range buf {
|
|
want := byte('1')
|
|
if i < 8<<10 {
|
|
want = '0'
|
|
} else if i >= 24<<10+chunkSize {
|
|
want = '4'
|
|
} else if i >= 16<<10+chunkSize {
|
|
want = '3'
|
|
} else if i >= 8<<10+chunkSize {
|
|
want = '2'
|
|
}
|
|
if b != want {
|
|
t.Fatalf("buf[%d]=%q, want %q", i, b, want)
|
|
}
|
|
}
|
|
|
|
want := []recordedFetch{
|
|
{fileId: "chunk0", isFullChunk: false, offset: chunkSize - 8<<10, size: 8 << 10},
|
|
{fileId: "chunk1", isFullChunk: true, offset: 0, size: chunkSize},
|
|
{fileId: "chunk2", isFullChunk: false, offset: 0, size: 8 << 10},
|
|
// partial views, but ciphered and compressed chunks download whole
|
|
// either way and the shared path decrypts/decompresses once for
|
|
// every buffer
|
|
{fileId: "chunk3", isFullChunk: true, offset: 0, size: chunkSize},
|
|
{fileId: "chunk4", isFullChunk: true, offset: 0, size: chunkSize},
|
|
}
|
|
got := map[string]recordedFetch{}
|
|
for _, f := range *fetches {
|
|
if _, dup := got[f.fileId]; dup {
|
|
t.Fatalf("chunk %s fetched more than once: %+v", f.fileId, *fetches)
|
|
}
|
|
got[f.fileId] = f
|
|
}
|
|
for _, w := range want {
|
|
if g, ok := got[w.fileId]; !ok {
|
|
t.Fatalf("chunk %s never fetched: %+v", w.fileId, *fetches)
|
|
} else if g != w {
|
|
t.Fatalf("chunk %s fetched as %+v, want %+v", w.fileId, g, w)
|
|
}
|
|
}
|
|
}
|
|
|
|
// The regression from issue #11564: a ranged GET sitting inside one big
|
|
// chunk. Every buffer of the request must stay a range fetch — none may
|
|
// escalate into a whole-chunk download once the reads look sequential.
|
|
func TestChunkReadAtRangeInsideOneChunkStaysRangeFetch(t *testing.T) {
|
|
const chunkSize = 1 << 20
|
|
const sliceSize = 16 << 10
|
|
|
|
rc := NewReaderCache(64, (*chunk_cache.TieredChunkCache)(nil), func(context.Context, string) ([]string, error) {
|
|
return []string{"unused"}, nil
|
|
}, nil)
|
|
defer rc.destroy()
|
|
fetches := fetchRecorder(rc)
|
|
|
|
// Range [32KiB, 96KiB) inside one 1MiB chunk: a single clipped view.
|
|
views := NewIntervalList[*ChunkView]()
|
|
views.AppendInterval(&Interval[*ChunkView]{
|
|
StartOffset: 32 << 10,
|
|
StopOffset: 96 << 10,
|
|
Value: &ChunkView{FileId: "chunk0", OffsetInChunk: 32 << 10, ViewSize: 64 << 10, ViewOffset: 32 << 10, ChunkSize: chunkSize},
|
|
})
|
|
|
|
reader := NewChunkReaderAtFromClient(context.Background(), rc, views, chunkSize, 0)
|
|
for offset := int64(32 << 10); offset < 96<<10; offset += sliceSize {
|
|
buf := make([]byte, sliceSize)
|
|
if n, err := reader.ReadAt(buf, offset); err != nil || n != sliceSize {
|
|
t.Fatalf("read at %d: n=%d err=%v", offset, n, err)
|
|
}
|
|
}
|
|
|
|
if len(*fetches) != 4 {
|
|
t.Fatalf("got %d fetches, want 4 range fetches: %+v", len(*fetches), *fetches)
|
|
}
|
|
for i, f := range *fetches {
|
|
wantOffset := int64(32<<10) + int64(i)*sliceSize
|
|
if f.isFullChunk || f.offset != wantOffset || f.size != sliceSize {
|
|
t.Fatalf("fetch %d = %+v, want range fetch offset=%d size=%d", i, f, wantOffset, sliceSize)
|
|
}
|
|
}
|
|
}
|
|
|
|
// A compressed chunk larger than the reader cache budget can never be
|
|
// downloaded whole — the budget rejects the buffer — so its partial view
|
|
// must fall back to a range fetch even though each range costs a full
|
|
// decompress server-side. The alternative is a failed GET.
|
|
func TestChunkReadAtOversizedCompressedChunkFallsBackToRange(t *testing.T) {
|
|
const chunkSize = 1 << 20
|
|
const sliceSize = 16 << 10
|
|
|
|
budget := NewReaderCacheBudget(64 << 10) // smaller than the chunk
|
|
rc := NewReaderCache(64, (*chunk_cache.TieredChunkCache)(nil), func(context.Context, string) ([]string, error) {
|
|
return []string{"unused"}, nil
|
|
}, nil, budget)
|
|
defer rc.destroy()
|
|
fetches := fetchRecorder(rc)
|
|
|
|
views := NewIntervalList[*ChunkView]()
|
|
views.AppendInterval(&Interval[*ChunkView]{
|
|
StartOffset: 32 << 10,
|
|
StopOffset: 64 << 10,
|
|
Value: &ChunkView{FileId: "chunk0", OffsetInChunk: 32 << 10, ViewSize: 32 << 10, ViewOffset: 32 << 10, ChunkSize: chunkSize, IsGzipped: true},
|
|
})
|
|
|
|
reader := NewChunkReaderAtFromClient(context.Background(), rc, views, chunkSize, 0)
|
|
buf := make([]byte, 32<<10)
|
|
if n, err := reader.ReadAt(buf, 32<<10); err != nil || n != len(buf) {
|
|
t.Fatalf("read: n=%d err=%v", n, err)
|
|
}
|
|
|
|
if len(*fetches) != 1 {
|
|
t.Fatalf("got %d fetches, want 1 range fetch: %+v", len(*fetches), *fetches)
|
|
}
|
|
if f := (*fetches)[0]; f.isFullChunk || f.offset != 32<<10 || f.size != 32<<10 {
|
|
t.Fatalf("fetch = %+v, want range fetch offset=%d size=%d", f, 32<<10, 32<<10)
|
|
}
|
|
}
|
|
|
|
// A chunk a stream is positioned in must outlast downloader-limit eviction:
|
|
// otherwise a busy cache drops the buffer mid-stream and forces a refetch.
|
|
func TestChunkReadAtPinnedChunkSurvivesEviction(t *testing.T) {
|
|
const chunkSize = 64 << 10
|
|
|
|
var fetches int32
|
|
rc := NewReaderCache(2, newMockChunkCacheForReaderCache(), func(context.Context, string) ([]string, error) {
|
|
return []string{"unused"}, nil
|
|
}, nil)
|
|
defer rc.destroy()
|
|
rc.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, fileId string, _ util_http.RefreshUrlsFunc) (int, error) {
|
|
if fileId == "chunk0" {
|
|
atomic.AddInt32(&fetches, 1)
|
|
}
|
|
return len(buffer), nil
|
|
}
|
|
|
|
stream := &chunkStream{}
|
|
buf := make([]byte, 16<<10)
|
|
// One slice in: the stream is positioned in chunk0 but has not left it.
|
|
if _, err := rc.readChunkAt(context.Background(), stream, buf, "chunk0", nil, false, 0, chunkSize, false); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// Fill the downloader map past its limit with unpinned chunks.
|
|
for _, fileId := range []string{"chunk1", "chunk2", "chunk3"} {
|
|
if _, err := rc.ReadChunkAt(context.Background(), buf, fileId, nil, false, 0, chunkSize, true); err != nil {
|
|
t.Fatalf("read %s: %v", fileId, err)
|
|
}
|
|
}
|
|
|
|
// The stream's next slice must come from the still-pinned buffer.
|
|
if _, err := rc.readChunkAt(context.Background(), stream, buf, "chunk0", nil, false, 16<<10, chunkSize, false); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if got := atomic.LoadInt32(&fetches); got != 1 {
|
|
t.Errorf("chunk0 fetched %d times, want 1", got)
|
|
}
|
|
}
|
|
|
|
// When every downloader is pinned the limit still applies: the oldest pinned
|
|
// buffer is evicted so abandoned streams cannot grow memory past the limit.
|
|
func TestChunkReadAtPinnedEvictionFallsBackWhenAllPinned(t *testing.T) {
|
|
const chunkSize = 64 << 10
|
|
rc := NewReaderCache(2, newMockChunkCacheForReaderCache(), func(context.Context, string) ([]string, error) {
|
|
return []string{"unused"}, nil
|
|
}, nil)
|
|
defer rc.destroy()
|
|
rc.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, _ string, _ util_http.RefreshUrlsFunc) (int, error) {
|
|
return len(buffer), nil
|
|
}
|
|
|
|
buf := make([]byte, 16<<10)
|
|
streamA := &chunkStream{}
|
|
streamB := &chunkStream{}
|
|
if _, err := rc.readChunkAt(context.Background(), streamA, buf, "chunk0", nil, false, 0, chunkSize, false); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := rc.readChunkAt(context.Background(), streamB, buf, "chunk1", nil, false, 0, chunkSize, false); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// Every downloader is now pinned; the next chunk must still get in.
|
|
if _, err := rc.readChunkAt(context.Background(), &chunkStream{}, buf, "chunk2", nil, false, 0, chunkSize, false); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if isRetained(rc, "chunk0") {
|
|
t.Fatal("oldest pinned downloader was not evicted past the limit")
|
|
}
|
|
if !isRetained(rc, "chunk2") {
|
|
t.Fatal("new downloader missing after pinned fallback eviction")
|
|
}
|
|
}
|