diff --git a/weed/filer/reader_cache.go b/weed/filer/reader_cache.go index cd6049beea..ccf9512dfc 100644 --- a/weed/filer/reader_cache.go +++ b/weed/filer/reader_cache.go @@ -49,6 +49,9 @@ type SingleChunkCacher struct { wg sync.WaitGroup cacheStartedCh chan struct{} done chan struct{} // signals when download is complete + // ctx is cancelled once no reader can use the download, or when it ends + ctx context.Context + cancel context.CancelFunc } func NewReaderCache(limit int, chunkCache chunk_cache.ChunkCache, lookupFileIdFn wdclient.LookupFileIdFunctionType, cacheInvalidator CacheInvalidator, budgets ...*ReaderCacheBudget) *ReaderCache { @@ -392,12 +395,18 @@ func (rc *ReaderCache) removeConsumed(downloader *SingleChunkCacher) { } } +// removeConsumedLocked also cancels a running download, as no reader can attach to it any more, +// including one destroy already dropped from the map. func (rc *ReaderCache) removeConsumedLocked(downloader *SingleChunkCacher) (removed bool) { - if rc.downloaders[downloader.chunkFileId] == downloader && downloader.consumable() { - delete(rc.downloaders, downloader.chunkFileId) - return true + if !downloader.consumable() { + return false } - return false + downloader.cancel() + if rc.downloaders[downloader.chunkFileId] != downloader { + return false + } + delete(rc.downloaders, downloader.chunkFileId) + return true } func (rc *ReaderCache) destroy() { @@ -405,13 +414,24 @@ func (rc *ReaderCache) destroy() { downloaders := rc.downloaders rc.downloaders = make(map[string]*SingleChunkCacher) rc.Unlock() + // out of the map no stream can come back to a download, so it is cancelled + // now or once its last reader or stream leaves + for _, downloader := range downloaders { + atomic.StoreInt32(&downloader.left, 1) + if downloader.consumable() { + downloader.cancel() + } + } for _, downloader := range downloaders { downloader.destroy() } } func newSingleChunkCacher(parent *ReaderCache, fileId string, cipherKey []byte, isGzipped bool, chunkSize int, shouldCache bool) *SingleChunkCacher { + ctx, cancel := context.WithCancel(context.Background()) return &SingleChunkCacher{ + ctx: ctx, + cancel: cancel, parent: parent, chunkFileId: fileId, cipherKey: cipherKey, @@ -428,6 +448,7 @@ func newSingleChunkCacher(parent *ReaderCache, fileId string, cipherKey []byte, // downloaders map, so a removal can never observe the fetch as absent. func (s *SingleChunkCacher) startCaching() { defer func() { + s.cancel() close(s.done) s.wg.Done() if s.hasCompletedError() { @@ -443,13 +464,7 @@ func (s *SingleChunkCacher) startCaching() { return } - // Intentionally use context.Background(), not a request-specific context. - // The downloaded chunk is a shared resource: multiple concurrent readers may - // wait on this same download via s.done. A request-scoped context that got - // cancelled would abort the download and error every other waiting reader. - // The download always runs to completion once started; readers that cancel - // individually drop out via readChunkAt's select on ctx.Done(). - urlStrings, err := s.parent.lookupFileIdFn(context.Background(), s.chunkFileId) + urlStrings, err := s.lookupFileId(s.ctx, s.chunkFileId) if err != nil { s.setError(fmt.Errorf("operation LookupFileId %s failed, err: %v", s.chunkFileId, err)) return @@ -459,9 +474,16 @@ func (s *SingleChunkCacher) startCaching() { return } - data, fetchErr := s.fetchChunkData(context.Background(), urlStrings) + // not a request's context: other readers may wait on this download via s.done + // and it is cancelled only when removeConsumedLocked drops the cacher + data, fetchErr := s.fetchChunkData(s.ctx, urlStrings) if fetchErr != nil { - data, fetchErr = s.retryFetchAfterCacheInvalidation(context.Background(), urlStrings, fetchErr) + data, fetchErr = s.retryFetchAfterCacheInvalidation(s.ctx, urlStrings, fetchErr) + } + // a cancelled fetch can return a partly filled buffer without an error + if fetchErr == nil && s.ctx.Err() != nil { + mem.Free(data) + data, fetchErr = nil, s.ctx.Err() } // Now acquire lock to update state @@ -498,7 +520,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, refreshUrls(ctx, s.parent.cacheInvalidator, s.parent.lookupFileIdFn, s.chunkFileId)) + _, fetchErr := s.parent.fetchChunkDataFn(ctx, data, urlStrings, s.cipherKey, s.isGzipped, true, 0, s.chunkFileId, refreshUrls(ctx, s.parent.cacheInvalidator, s.lookupFileId, s.chunkFileId)) if fetchErr != nil { mem.Free(data) return nil, fetchErr @@ -508,7 +530,7 @@ func (s *SingleChunkCacher) fetchChunkData(ctx context.Context, urlStrings []str 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 { + err := retryFetchWithFreshLocations(ctx, s.parent.cacheInvalidator, s.lookupFileId, s.chunkFileId, oldUrlStrings, originalErr, func(newUrls []string) error { var fetchErr error data, fetchErr = s.fetchChunkData(ctx, newUrls) return fetchErr @@ -519,6 +541,12 @@ func (s *SingleChunkCacher) retryFetchAfterCacheInvalidation(ctx context.Context return data, nil } +// lookupFileId ignores the download's context: FilerClient coalesces concurrent +// lookups of a volume, so a cancelled download must not fail other readers' lookups. +func (s *SingleChunkCacher) lookupFileId(_ context.Context, fileId string) ([]string, error) { + return s.parent.lookupFileIdFn(context.Background(), fileId) +} + func (s *SingleChunkCacher) destroy() { // wait for all reads to finish before destroying the data s.wg.Wait() @@ -533,8 +561,8 @@ func (s *SingleChunkCacher) destroy() { // readChunkAt reads data from the cached chunk. // It waits for the download to complete if it's still in progress. -// The ctx parameter allows the reader to cancel its wait (but the download continues -// for other readers - see comment in startCaching about shared resource semantics). +// The ctx parameter allows the reader to cancel its wait; the download continues +// for the other readers and streams that still wait on it. // The caller must s.wg.Add(1) under the ReaderCache lock before calling; this only releases it. func (s *SingleChunkCacher) readChunkAt(ctx context.Context, buf []byte, offset int64) (n int, err error) { defer func() { diff --git a/weed/filer/reader_cache_budget.go b/weed/filer/reader_cache_budget.go index b048e26d99..17f4fd78c1 100644 --- a/weed/filer/reader_cache_budget.go +++ b/weed/filer/reader_cache_budget.go @@ -45,6 +45,11 @@ func (b *ReaderCacheBudget) reserve(s *SingleChunkCacher) error { return fmt.Errorf("chunk buffer needs %d bytes, exceeding reader cache budget %d; increase the readerCacheSizeMB budget", size, b.limit) } for { + // a cancelled download must not evict a chunk for itself + if err := s.ctx.Err(); err != nil { + return err + } + b.Lock() if size <= b.limit-b.used { b.used += size @@ -89,7 +94,11 @@ func (b *ReaderCacheBudget) reserve(s *SingleChunkCacher) error { } changed := b.changed b.Unlock() - <-changed + select { + case <-changed: + case <-s.ctx.Done(): + return s.ctx.Err() + } } } diff --git a/weed/filer/reader_cache_memory_test.go b/weed/filer/reader_cache_memory_test.go index da73c7a2ea..d9c11f9cb2 100644 --- a/weed/filer/reader_cache_memory_test.go +++ b/weed/filer/reader_cache_memory_test.go @@ -428,3 +428,60 @@ func TestReaderCacheReReadAfterEviction(t *testing.T) { t.Fatalf("fetchCount=%d, want 3 (a, b, a-re-read)", got) } } + +// A download still waiting for budget is cancelled when its last stream leaves. +func TestReaderCacheBudgetWaitStopsWithItsLastStream(t *testing.T) { + budget := NewReaderCacheBudget(4 << 10) + var lookups sync.Map + rc := NewReaderCache(256, newMockChunkCacheForReaderCache(), func(_ context.Context, fileId string) ([]string, error) { + lookups.Store(fileId, true) + return []string{"unused"}, nil + }, nil, budget) + defer rc.destroy() + + holding := make(chan struct{}) + var holdingOnce sync.Once + finish := make(chan struct{}) + rc.fetchChunkDataFn = func(ctx context.Context, buffer []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, _ string, _ util_http.RefreshUrlsFunc) (int, error) { + holdingOnce.Do(func() { close(holding) }) + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-finish: + return len(buffer), nil + } + } + defer close(finish) + + // "a" takes the whole budget and its download does not finish + go rc.ReadChunkAt(context.Background(), make([]byte, 1), "a", nil, false, 0, 3<<10, false) + <-holding + + var stream chunkStream + ctx, cancel := context.WithCancel(context.Background()) + readDone := make(chan error, 1) + go func() { + _, err := rc.readChunkAt(ctx, &stream, make([]byte, 1), "b", nil, true, 0, 3<<10, false) + readDone <- err + }() + time.Sleep(50 * time.Millisecond) + if _, looked := lookups.Load("b"); looked { + t.Fatal(`"b" did not wait for budget`) + } + + cancel() + if err := <-readDone; err != context.Canceled { + t.Fatalf("cancelled read: got %v, want context.Canceled", err) + } + + released := make(chan struct{}) + go func() { + rc.releaseStream(&stream) + close(released) + }() + select { + case <-released: + case <-time.After(2 * time.Second): + t.Fatal("releasing the stream waited for budget nobody needs") + } +} diff --git a/weed/filer/reader_cache_test.go b/weed/filer/reader_cache_test.go index faf184f878..844f3f8eb7 100644 --- a/weed/filer/reader_cache_test.go +++ b/weed/filer/reader_cache_test.go @@ -773,3 +773,406 @@ func TestReaderCacheBookkeepingOffCacheLock(t *testing.T) { } rc.Unlock() } + +// Releasing the last stream of an unfinished download cancels it without waiting for it. +func TestAbandonedChunkDownloadStopsWithItsLastStream(t *testing.T) { + lookupFn := func(ctx context.Context, fileId string) ([]string, error) { + return []string{"http://volume/" + fileId}, nil + } + rc := NewReaderCache(10, newMockChunkCacheForReaderCache(), lookupFn, nil) + defer rc.destroy() + + fetchStarted := make(chan struct{}) + fetchEnded := make(chan error, 1) + rc.fetchChunkDataFn = func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (int, error) { + close(fetchStarted) + select { + case <-ctx.Done(): + fetchEnded <- ctx.Err() + return 0, ctx.Err() + case <-time.After(10 * time.Second): + fetchEnded <- nil + return 0, fmt.Errorf("volume server never answered") + } + } + + // a gzipped chunk is read through the shared whole-chunk download + var stream chunkStream + ctx, cancel := context.WithCancel(context.Background()) + readDone := make(chan error, 1) + go func() { + _, err := rc.readChunkAt(ctx, &stream, make([]byte, 100), "3077,02c3475d67aaf564", nil, true, 0, 4<<20, false) + readDone <- err + }() + + <-fetchStarted + cancel() + if err := <-readDone; err != context.Canceled { + t.Fatalf("cancelled read: got %v, want context.Canceled", err) + } + + released := make(chan struct{}) + go func() { + rc.releaseStream(&stream) + close(released) + }() + + select { + case <-released: + case <-time.After(2 * time.Second): + t.Fatal("releasing the stream waited for a download nobody wants") + } + + select { + case err := <-fetchEnded: + if err != context.Canceled { + t.Fatalf("abandoned download: got %v, want context.Canceled", err) + } + case <-time.After(2 * time.Second): + t.Fatal("the abandoned download kept running") + } +} + +// One stream leaving a shared download must not cancel it for another stream. +func TestAbandonedChunkDownloadKeepsRunningForOtherStream(t *testing.T) { + content := []byte("gzipped chunk 01") + fileId := "3077,02c3475d67aaf564" + lookupFn := func(ctx context.Context, fileId string) ([]string, error) { + return []string{"http://volume/" + fileId}, nil + } + rc := NewReaderCache(10, newMockChunkCacheForReaderCache(), lookupFn, nil) + defer rc.destroy() + + fetchStarted := make(chan struct{}) + finishFetch := make(chan struct{}) + rc.fetchChunkDataFn = func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (int, error) { + close(fetchStarted) + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-finishFetch: + return copy(buffer, content), nil + } + } + + read := func(ctx context.Context, stream *chunkStream, done chan<- error, buffer []byte) { + _, err := rc.readChunkAt(ctx, stream, buffer, fileId, nil, true, 0, len(content), false) + done <- err + } + + var leaving, staying chunkStream + leavingCtx, cancel := context.WithCancel(context.Background()) + leavingDone, stayingDone := make(chan error, 1), make(chan error, 1) + stayingBuffer := make([]byte, len(content)) + go read(leavingCtx, &leaving, leavingDone, make([]byte, len(content))) + <-fetchStarted + go read(context.Background(), &staying, stayingDone, stayingBuffer) + + // wait until the second stream is attached to the same download + deadline := time.Now().Add(2 * time.Second) + for { + rc.RLock() + cacher := rc.downloaders[fileId] + rc.RUnlock() + if cacher != nil && atomic.LoadInt32(&cacher.pins) == 2 { + break + } + if time.Now().After(deadline) { + t.Fatal("second stream never attached to the download") + } + time.Sleep(time.Millisecond) + } + + cancel() + if err := <-leavingDone; err != context.Canceled { + t.Fatalf("cancelled read: got %v, want context.Canceled", err) + } + rc.releaseStream(&leaving) + + close(finishFetch) + select { + case err := <-stayingDone: + if err != nil { + t.Fatalf("staying stream: got %v, want the chunk", err) + } + if string(stayingBuffer) != string(content) { + t.Fatalf("staying stream read %q, want %q", stayingBuffer, content) + } + case <-time.After(2 * time.Second): + t.Fatal("staying stream never got its data") + } + rc.releaseStream(&staying) +} + +func TestAbandonedChunkDownloadKeepsRunningForWaitingReader(t *testing.T) { + content := []byte("gzipped chunk 01") + fileId := "3077,02c3475d67aaf564" + lookupFn := func(ctx context.Context, fileId string) ([]string, error) { + return []string{"http://volume/" + fileId}, nil + } + rc := NewReaderCache(10, newMockChunkCacheForReaderCache(), lookupFn, nil) + defer rc.destroy() + + fetchStarted := make(chan struct{}) + finishFetch := make(chan struct{}) + rc.fetchChunkDataFn = func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (int, error) { + close(fetchStarted) + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-finishFetch: + return copy(buffer, content), nil + } + } + + var leaving chunkStream + leavingCtx, cancel := context.WithCancel(context.Background()) + leavingDone, waitingDone := make(chan error, 1), make(chan error, 1) + go func() { + _, err := rc.readChunkAt(leavingCtx, &leaving, make([]byte, len(content)), fileId, nil, true, 0, len(content), false) + leavingDone <- err + }() + <-fetchStarted + waitingBuffer := make([]byte, len(content)) + go func() { + _, err := rc.ReadChunkAt(context.Background(), waitingBuffer, fileId, nil, true, 0, len(content), false) + waitingDone <- err + }() + + deadline := time.Now().Add(2 * time.Second) + for { + rc.RLock() + cacher := rc.downloaders[fileId] + rc.RUnlock() + if cacher != nil && atomic.LoadInt32(&cacher.readers) == 2 { + break + } + if time.Now().After(deadline) { + t.Fatal("waiting reader never attached to the download") + } + time.Sleep(time.Millisecond) + } + + cancel() + if err := <-leavingDone; err != context.Canceled { + t.Fatalf("cancelled read: got %v, want context.Canceled", err) + } + rc.releaseStream(&leaving) + + close(finishFetch) + select { + case err := <-waitingDone: + if err != nil { + t.Fatalf("waiting reader: got %v, want the chunk", err) + } + if string(waitingBuffer) != string(content) { + t.Fatalf("waiting reader read %q, want %q", waitingBuffer, content) + } + case <-time.After(2 * time.Second): + t.Fatal("waiting reader never got its data") + } +} + +// A download cancelled mid-copy must not cache the partly filled buffer. +func TestAbandonedChunkDownloadIsNotCached(t *testing.T) { + fileId := "3077,02c3475d67aaf564" + lookupFn := func(ctx context.Context, fileId string) ([]string, error) { + return []string{"http://volume/" + fileId}, nil + } + cache := newMockChunkCacheForReaderCache() + rc := NewReaderCache(10, cache, lookupFn, nil) + defer rc.destroy() + + fetchStarted := make(chan struct{}) + rc.fetchChunkDataFn = func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (int, error) { + close(fetchStarted) + <-ctx.Done() + // like readEncryptedUrl: the body was read, the copy was skipped, no error + return 0, nil + } + + var stream chunkStream + ctx, cancel := context.WithCancel(context.Background()) + readDone := make(chan error, 1) + go func() { + _, err := rc.readChunkAt(ctx, &stream, make([]byte, 100), fileId, nil, true, 0, 100, true) + readDone <- err + }() + + <-fetchStarted + cancel() + if err := <-readDone; err != context.Canceled { + t.Fatalf("cancelled read: got %v, want context.Canceled", err) + } + rc.releaseStream(&stream) + + cache.mu.Lock() + defer cache.mu.Unlock() + if _, cached := cache.data[fileId]; cached { + t.Fatal("the partly filled buffer of a cancelled download was cached") + } +} + +// Other readers can join a re-lookup through FilerClient, so cancelling the download must not cancel it. +func TestAbandonedChunkDownloadKeepsItsRelookupAlive(t *testing.T) { + fileId := "3077,02c3475d67aaf564" + var lookups int32 + relookupStarted := make(chan struct{}) + downloadCancelled := make(chan struct{}) + relookupErr := make(chan error, 1) + lookupFn := func(ctx context.Context, fileId string) ([]string, error) { + if atomic.AddInt32(&lookups, 1) > 1 { + close(relookupStarted) + <-downloadCancelled + relookupErr <- ctx.Err() + } + return []string{"http://volume/" + fileId}, nil + } + rc := NewReaderCache(10, newMockChunkCacheForReaderCache(), lookupFn, &mockCacheInvalidatorForReaderCache{}) + defer rc.destroy() + rc.fetchChunkDataFn = func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (int, error) { + return 0, fmt.Errorf("connection reset by peer") + } + + var stream chunkStream + ctx, cancel := context.WithCancel(context.Background()) + readDone := make(chan error, 1) + go func() { + _, err := rc.readChunkAt(ctx, &stream, make([]byte, 100), fileId, nil, true, 0, 100, false) + readDone <- err + }() + + <-relookupStarted + cancel() + if err := <-readDone; err != context.Canceled { + t.Fatalf("cancelled read: got %v, want context.Canceled", err) + } + released := make(chan struct{}) + go func() { + rc.releaseStream(&stream) + close(released) + }() + // the cacher leaves the map under the same lock that cancels it + deadline := time.Now().Add(2 * time.Second) + for { + rc.RLock() + _, found := rc.downloaders[fileId] + rc.RUnlock() + if !found { + break + } + if time.Now().After(deadline) { + t.Fatal("releasing the last stream never dropped the download") + } + time.Sleep(time.Millisecond) + } + close(downloadCancelled) + + if err := <-relookupErr; err != nil { + t.Fatalf("re-lookup context: got %v, want it independent of the download", err) + } + select { + case <-released: + case <-time.After(2 * time.Second): + t.Fatal("releasing the stream never returned") + } +} + +// Closing the reader cache cancels a stalled prefetch nobody reads instead of waiting for it. +func TestReaderCacheDestroyCancelsUnreadPrefetch(t *testing.T) { + lookupFn := func(ctx context.Context, fileId string) ([]string, error) { + return []string{"http://volume/" + fileId}, nil + } + rc := NewReaderCache(10, newMockChunkCacheForReaderCache(), lookupFn, nil) + fetchStarted := make(chan struct{}) + rc.fetchChunkDataFn = func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (int, error) { + close(fetchStarted) + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-time.After(10 * time.Second): + return 0, fmt.Errorf("volume server never answered") + } + } + rc.MaybeCache(&Interval[*ChunkView]{Value: &ChunkView{FileId: "3077,02c3475d67aaf564", ViewSize: 1024, ChunkSize: 1024}}, 1) + <-fetchStarted + + destroyed := make(chan struct{}) + go func() { + rc.destroy() + close(destroyed) + }() + select { + case <-destroyed: + case <-time.After(2 * time.Second): + t.Fatal("destroy waited for a prefetch nobody reads") + } +} + +// A download destroy dropped while a stream still held it is cancelled once that stream leaves. +func TestReaderCacheDestroyCancelsDownloadAfterItsLastStream(t *testing.T) { + lookupFn := func(ctx context.Context, fileId string) ([]string, error) { + return []string{"http://volume/" + fileId}, nil + } + rc := NewReaderCache(10, newMockChunkCacheForReaderCache(), lookupFn, nil) + fetchStarted := make(chan struct{}) + fetchEnded := make(chan error, 1) + rc.fetchChunkDataFn = func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (int, error) { + close(fetchStarted) + select { + case <-ctx.Done(): + fetchEnded <- ctx.Err() + return 0, ctx.Err() + case <-time.After(10 * time.Second): + fetchEnded <- nil + return 0, fmt.Errorf("volume server never answered") + } + } + + var stream chunkStream + ctx, cancel := context.WithCancel(context.Background()) + readDone := make(chan error, 1) + go func() { + _, err := rc.readChunkAt(ctx, &stream, make([]byte, 100), "3077,02c3475d67aaf564", nil, true, 0, 4<<20, false) + readDone <- err + }() + <-fetchStarted + cancel() + if err := <-readDone; err != context.Canceled { + t.Fatalf("cancelled read: got %v, want context.Canceled", err) + } + + destroyed := make(chan struct{}) + go func() { + rc.destroy() + close(destroyed) + }() + deadline := time.Now().Add(2 * time.Second) + for { + rc.RLock() + mapped := len(rc.downloaders) + rc.RUnlock() + if mapped == 0 { + break + } + if time.Now().After(deadline) { + t.Fatal("destroy never dropped the download from the map") + } + time.Sleep(time.Millisecond) + } + select { + case err := <-fetchEnded: + t.Fatalf("destroy cancelled a download a stream still holds: %v", err) + case <-time.After(50 * time.Millisecond): + } + + rc.releaseStream(&stream) + select { + case <-destroyed: + case <-time.After(2 * time.Second): + t.Fatal("destroy kept waiting for a download its last stream left") + } + if err := <-fetchEnded; err != context.Canceled { + t.Fatalf("orphaned download: got %v, want context.Canceled", err) + } +}