From 57c64e386a647b619afd2431ba35aad4b261b7e2 Mon Sep 17 00:00:00 2001 From: Deluan Date: Wed, 22 Jul 2026 10:01:47 -0400 Subject: [PATCH] feat(artwork): add acquisition worker service --- cmd/root.go | 10 ++ cmd/wire_gen.go | 15 ++ cmd/wire_injectors.go | 6 + conf/configuration.go | 4 + core/artwork/image_store.go | 8 + core/artwork/prune.go | 4 +- core/artwork/wire_providers.go | 2 + core/artwork/worker.go | 250 +++++++++++++++++++++++++++++ core/artwork/worker_test.go | 188 ++++++++++++++++++++++ core/artwork/worker_timing_test.go | 95 +++++++++++ 10 files changed, 580 insertions(+), 2 deletions(-) create mode 100644 core/artwork/worker.go create mode 100644 core/artwork/worker_test.go create mode 100644 core/artwork/worker_timing_test.go diff --git a/cmd/root.go b/cmd/root.go index 9e2b38cd8..2b64ab850 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -88,6 +88,7 @@ func runNavidrome(ctx context.Context) { g.Go(startInsightsCollector(ctx)) g.Go(scheduleDBAnalyzer(ctx)) g.Go(startPluginManager(ctx)) + g.Go(startArtworkWorker(ctx)) g.Go(runInitialScan(ctx)) if conf.Server.Scanner.Enabled { g.Go(startScanWatcher(ctx)) @@ -344,6 +345,15 @@ func startPlaybackServer(ctx context.Context) func() error { } } +// startArtworkWorker starts the background artwork acquisition worker. It always +// runs; the queue is simply empty until something enqueues work into it. +func startArtworkWorker(ctx context.Context) func() error { + return func() error { + log.Info(ctx, "Starting artwork worker") + return CreateArtworkWorker().Run(ctx) + } +} + // startPluginManager starts the plugin manager, if configured. func startPluginManager(ctx context.Context) func() error { return func() error { diff --git a/cmd/wire_gen.go b/cmd/wire_gen.go index c4a797ef4..44c4cb46d 100644 --- a/cmd/wire_gen.go +++ b/cmd/wire_gen.go @@ -236,6 +236,21 @@ func GetPlaybackServer() playback.PlaybackServer { return playbackServer } +func CreateArtworkWorker() *artwork.Worker { + sqlDB := db.Db() + dataStore := persistence.New(sqlDB) + imageStore := artwork.ProvideImageStore() + broker := events.GetBroker() + metricsMetrics := metrics.GetPrometheusInstance(dataStore) + manager := plugins.GetManager(dataStore, broker, metricsMetrics) + agentsAgents := agents.GetAgents(dataStore, manager) + matcherMatcher := matcher.New(dataStore) + provider := external.NewProvider(dataStore, agentsAgents, matcherMatcher) + fFmpeg := ffmpeg.New() + worker := artwork.NewWorker(dataStore, imageStore, provider, fFmpeg) + return worker +} + func getPluginManager() *plugins.Manager { sqlDB := db.Db() dataStore := persistence.New(sqlDB) diff --git a/cmd/wire_injectors.go b/cmd/wire_injectors.go index 94faa5af3..a835e5c4c 100644 --- a/cmd/wire_injectors.go +++ b/cmd/wire_injectors.go @@ -136,6 +136,12 @@ func GetPlaybackServer() playback.PlaybackServer { )) } +func CreateArtworkWorker() *artwork.Worker { + panic(wire.Build( + allProviders, + )) +} + func getPluginManager() *plugins.Manager { panic(wire.Build( allProviders, diff --git a/conf/configuration.go b/conf/configuration.go index 83793bd43..57ee4e4bc 100644 --- a/conf/configuration.go +++ b/conf/configuration.go @@ -139,6 +139,8 @@ type configOptions struct { DevArtworkThrottleBacklogLimit int DevArtworkThrottleBacklogTimeout time.Duration DevArtworkThrottleBuffered bool + DevArtworkWorkerConcurrency int + DevArtworkExternalRPS int DevArtistInfoTimeToLive time.Duration DevAlbumInfoTimeToLive time.Duration DevExternalScanner bool @@ -900,6 +902,8 @@ func setViperDefaults() { viper.SetDefault("devartworkthrottlebackloglimit", consts.RequestThrottleBacklogLimit) viper.SetDefault("devartworkthrottlebacklogtimeout", consts.RequestThrottleBacklogTimeout) viper.SetDefault("devartworkthrottlebuffered", true) + viper.SetDefault("devartworkworkerconcurrency", 2) + viper.SetDefault("devartworkexternalrps", 2) viper.SetDefault("devartistinfotimetolive", consts.ArtistInfoTimeToLive) viper.SetDefault("devalbuminfotimetolive", consts.AlbumInfoTimeToLive) viper.SetDefault("devexternalscanner", true) diff --git a/core/artwork/image_store.go b/core/artwork/image_store.go index 4b4e482c9..53662f686 100644 --- a/core/artwork/image_store.go +++ b/core/artwork/image_store.go @@ -10,6 +10,8 @@ import ( "strings" "time" + "github.com/navidrome/navidrome/conf" + "github.com/navidrome/navidrome/consts" "github.com/zeebo/xxh3" ) @@ -31,6 +33,12 @@ func NewImageStore(rootDir string) *ImageStore { return &ImageStore{root: rootDir} } +// ProvideImageStore roots the store in its own subtree under the data folder, so +// Prune's recursive sweep never reaches the per-entity upload folders next to it. +func ProvideImageStore() *ImageStore { + return NewImageStore(filepath.Join(conf.Server.DataFolder.String(), consts.ArtworkFolder, "store")) +} + // extForMime is deliberately NOT mime.ExtensionsByType: extensions are baked into // content-addressed paths and re-derived on Open, so they must be stable across OSes. func extForMime(m string) string { diff --git a/core/artwork/prune.go b/core/artwork/prune.go index db1e4fdc1..a40194b8a 100644 --- a/core/artwork/prune.go +++ b/core/artwork/prune.go @@ -39,8 +39,8 @@ func Prune(ctx context.Context, ds model.DataStore, store *ImageStore) error { if _, ok := survivors[h]; ok { continue } - // A spared fresh file is at worst a stray a later sweep reclaims; full - // worker/prune mutual exclusion is Phase 2's concern. + // A spared fresh file is at worst a stray a later sweep reclaims; + // Worker.RunPrune serializes prune against in-flight acquisitions. if err := store.Remove(h, arts[h].Mime, cutoff); err != nil { log.Warn(ctx, "Prune: could not remove artwork file", "hash", h, err) } diff --git a/core/artwork/wire_providers.go b/core/artwork/wire_providers.go index 63231b54a..868ea8f56 100644 --- a/core/artwork/wire_providers.go +++ b/core/artwork/wire_providers.go @@ -8,4 +8,6 @@ var Set = wire.NewSet( NewArtwork, GetImageCache, NewCacheWarmer, + NewWorker, + ProvideImageStore, ) diff --git a/core/artwork/worker.go b/core/artwork/worker.go new file mode 100644 index 000000000..1ebe15527 --- /dev/null +++ b/core/artwork/worker.go @@ -0,0 +1,250 @@ +package artwork + +import ( + "context" + "errors" + "io" + "math" + "math/rand/v2" + "sync" + "time" + + "github.com/navidrome/navidrome/conf" + "github.com/navidrome/navidrome/core/external" + "github.com/navidrome/navidrome/core/ffmpeg" + "github.com/navidrome/navidrome/log" + "github.com/navidrome/navidrome/model" + "golang.org/x/time/rate" +) + +const ( + workerPollInterval = 5 * time.Second + backoffBase = 5 * time.Minute + backoffCap = 48 * time.Hour + breakerThreshold = 5 + breakerProbeAfter = time.Minute +) + +var errBreakerOpen = errors.New("artwork: external circuit breaker open") + +// Worker drains the artwork queue and runs each item through processItem. The +// external step is rate-limited and circuit-broken; prune is serialized against +// in-flight acquisitions via pruneMu (acquisitions RLock, prune Lock). +type Worker struct { + deps workerDeps + limiter *rate.Limiter + breaker *breaker + pruneMu sync.RWMutex + wake chan struct{} + runCtx context.Context + + mu sync.Mutex + inFlight map[string]struct{} +} + +func NewWorker(ds model.DataStore, store *ImageStore, prov external.Provider, ffmpeg ffmpeg.FFmpeg) *Worker { + rps := conf.Server.DevArtworkExternalRPS + limit := rate.Inf + if rps > 0 { + limit = rate.Limit(rps) + } + w := &Worker{ + deps: workerDeps{ds: ds, store: store, prov: prov, ffmpeg: ffmpeg}, + limiter: rate.NewLimiter(limit, max(1, rps)), + breaker: newBreaker(), + wake: make(chan struct{}, 1), + runCtx: context.Background(), + inFlight: map[string]struct{}{}, + } + w.deps.extGate = w.gate + return w +} + +// Run blocks draining the queue until ctx is cancelled. It exits cleanly with no +// leaked goroutines: each drain waits for its batch before the loop can return. +func (w *Worker) Run(ctx context.Context) error { + w.runCtx = ctx + concurrency := max(1, conf.Server.DevArtworkWorkerConcurrency) + ticker := time.NewTicker(workerPollInterval) + defer ticker.Stop() + for { + n, err := w.drain(ctx, concurrency) + if err != nil && ctx.Err() == nil { + log.Warn(ctx, "artwork: worker drain failed", err) + } + if ctx.Err() != nil { + return nil + } + if n > 0 { + continue // keep draining while the queue has ready work + } + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + case <-w.wake: + } + } +} + +// Bump enqueues an item at the highest priority and wakes the drain loop. It is +// non-blocking: a wake already pending is enough. +func (w *Worker) Bump(kind, id string) { + item := model.ArtworkQueueItem{ + ItemKind: kind, + ItemID: id, + ImageType: model.ImageTypePrimary, + Priority: model.ArtworkPriorityBump, + } + if err := w.deps.ds.ArtworkQueue(context.Background()).Enqueue(item); err != nil { + log.Warn("artwork: could not bump queue item", "kind", kind, "id", id, err) + return + } + select { + case w.wake <- struct{}{}: + default: + } +} + +// RunPrune runs Prune under the worker's write lock, so no acquisition can place +// a file while orphans are being reclaimed. This is the only sanctioned prune path. +func (w *Worker) RunPrune(ctx context.Context) error { + w.pruneMu.Lock() + defer w.pruneMu.Unlock() + return Prune(ctx, w.deps.ds, w.deps.store) +} + +func (w *Worker) drain(ctx context.Context, concurrency int) (int, error) { + batch, err := w.deps.ds.ArtworkQueue(ctx).DequeueBatch(2 * concurrency) + if err != nil { + return 0, err + } + items := w.claim(batch) + if len(items) == 0 { + return 0, nil + } + sem := make(chan struct{}, concurrency) + var wg sync.WaitGroup + for _, item := range items { + sem <- struct{}{} + wg.Add(1) + go func(it model.ArtworkQueueItem) { + defer wg.Done() + defer func() { <-sem }() + defer w.release(it) + w.process(ctx, it) + }(item) + } + wg.Wait() + return len(items), nil +} + +func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) { + if item.ImageType == "" { + item.ImageType = model.ImageTypePrimary + } + w.pruneMu.RLock() + out := processItem(ctx, &w.deps, item) + w.pruneMu.RUnlock() + + queue := w.deps.ds.ArtworkQueue(ctx) + switch out { + case outcomeFound, outcomeAbsent: + if err := queue.Delete(item.ItemKind, item.ItemID, item.ImageType); err != nil { + log.Warn(ctx, "artwork: could not delete processed queue item", "kind", item.ItemKind, "id", item.ItemID, err) + } + case outcomeFailed: + retryAt := time.Now().Add(backoff(item.Attempts)) + if err := queue.MarkFailed(item.ItemKind, item.ItemID, item.ImageType, retryAt); err != nil { + log.Warn(ctx, "artwork: could not reschedule failed queue item", "kind", item.ItemKind, "id", item.ItemID, err) + } + } +} + +// claim reserves items not already in flight, so a wake-triggered re-drain never +// double-processes an item still running from a previous cycle. +func (w *Worker) claim(batch []model.ArtworkQueueItem) []model.ArtworkQueueItem { + w.mu.Lock() + defer w.mu.Unlock() + var out []model.ArtworkQueueItem + for _, it := range batch { + k := queueKey(it) + if _, busy := w.inFlight[k]; busy { + continue + } + w.inFlight[k] = struct{}{} + out = append(out, it) + } + return out +} + +func (w *Worker) release(it model.ArtworkQueueItem) { + w.mu.Lock() + delete(w.inFlight, queueKey(it)) + w.mu.Unlock() +} + +func queueKey(it model.ArtworkQueueItem) string { + return it.ItemKind + "|" + it.ItemID + "|" + it.ImageType +} + +// gate wraps the external step with the rate limiter and circuit breaker, matching +// extGateFunc so it can be injected via workerDeps.extGate. +func (w *Worker) gate(f func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) { + if !w.breaker.allow() { + return nil, "", errBreakerOpen + } + if err := w.limiter.Wait(w.runCtx); err != nil { + return nil, "", err + } + r, path, err := f() + w.breaker.record(err) + return r, path, err +} + +// backoffFor returns min(5m×4^n, 48h) scaled by (1+jitter), with jitter in [-0.2, 0.2]. +func backoffFor(attempts int, jitter float64) time.Duration { + d := math.Min(float64(backoffBase)*math.Pow(4, float64(attempts)), float64(backoffCap)) + return time.Duration(d * (1 + jitter)) +} + +func backoff(attempts int) time.Duration { + return backoffFor(attempts, rand.Float64()*0.4-0.2) //nolint:gosec // retry jitter, not security-sensitive +} + +// breaker opens after breakerThreshold consecutive external errors and admits a +// single probe once breakerProbeAfter has elapsed; a success re-closes it. +type breaker struct { + mu sync.Mutex + failures int + openedAt time.Time +} + +func newBreaker() *breaker { return &breaker{} } + +func (b *breaker) allow() bool { + b.mu.Lock() + defer b.mu.Unlock() + if b.failures < breakerThreshold { + return true + } + if time.Since(b.openedAt) >= breakerProbeAfter { + b.openedAt = time.Now() // start a fresh probe window so only one caller passes + return true + } + return false +} + +func (b *breaker) record(err error) { + b.mu.Lock() + defer b.mu.Unlock() + // A not-found is a definitive answer, not a fault; only real errors trip the breaker. + if err == nil || errors.Is(err, model.ErrNotFound) { + b.failures = 0 + return + } + b.failures++ + if b.failures == breakerThreshold { + b.openedAt = time.Now() + } +} diff --git a/core/artwork/worker_test.go b/core/artwork/worker_test.go new file mode 100644 index 000000000..d41249fa2 --- /dev/null +++ b/core/artwork/worker_test.go @@ -0,0 +1,188 @@ +package artwork + +import ( + "context" + "errors" + "io" + "net/url" + "os" + "time" + + "github.com/navidrome/navidrome/conf" + "github.com/navidrome/navidrome/conf/configtest" + "github.com/navidrome/navidrome/model" + "github.com/navidrome/navidrome/tests" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func findQueued(q *tests.MockArtworkQueueRepo, kind, id string) *model.ArtworkQueueItem { + for _, it := range q.Data { + if it.ItemKind == kind && it.ItemID == id { + return &it + } + } + return nil +} + +var _ = Describe("Worker", func() { + var ( + ctx context.Context + ds *tests.MockDataStore + folderRepo *fakeFolderRepo + libRepo *tests.MockLibraryRepo + ffm *tests.MockFFmpeg + prov *fakeExternalProvider + store *ImageStore + artRepo *tests.MockArtworkRepo + queueRepo *tests.MockArtworkQueueRepo + repoRoot string + w *Worker + ) + + BeforeEach(func() { + DeferCleanup(configtest.SetupConfig()) + ctx = context.Background() + var err error + repoRoot, err = os.Getwd() + Expect(err).ToNot(HaveOccurred()) + + folderRepo = &fakeFolderRepo{} + libRepo = &tests.MockLibraryRepo{} + libRepo.SetData(model.Libraries{{ID: 0, Path: testFileLibPath(repoRoot)}}) + ffm = tests.NewMockFFmpeg("") + prov = &fakeExternalProvider{} + artRepo = tests.CreateMockArtworkRepo() + queueRepo = tests.CreateMockArtworkQueueRepo() + ds = &tests.MockDataStore{ + MockedFolder: folderRepo, + MockedLibrary: libRepo, + MockedArtwork: artRepo, + MockedArtworkQueue: queueRepo, + } + ds.MockedAlbum = tests.CreateMockAlbumRepo() + store = NewImageStore(GinkgoT().TempDir()) + conf.Server.CoverArtPriority = "cover.jpg, embedded" + conf.Server.DevArtworkExternalRPS = 1000 // keep the limiter out of the way of behavior tests + w = NewWorker(ds, store, prov, ffm) + }) + + Describe("drain", func() { + It("processes a seeded queue item and removes it from the queue", func() { + folderRepo.result = []model.Folder{{ + Path: "tests/fixtures/artist/an-album", + ImageFiles: []string{"cover.jpg"}, + }} + ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{ + {ID: "al1", Name: "Album", FolderIDs: []string{"f1"}}, + }) + Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ + ItemKind: "al", ItemID: "al1", Priority: model.ArtworkPriorityScan, + })).To(Succeed()) + + n, err := w.drain(ctx, 2) + Expect(err).ToNot(HaveOccurred()) + Expect(n).To(Equal(1)) + + ia, err := artRepo.GetItemArtwork("al", "al1", model.ImageTypePrimary) + Expect(err).ToNot(HaveOccurred()) + Expect(ia.Source).To(Equal("folder")) + + count, err := queueRepo.Count() + Expect(err).ToNot(HaveOccurred()) + Expect(count).To(BeZero(), "a found item must be deleted from the queue") + }) + + It("reschedules a failed item via MarkFailed with a backed-off retry_at", func() { + conf.Server.CoverArtPriority = "external" + ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{{ID: "al4", Name: "Album"}}) + prov.albumImage = func(context.Context, string) (*url.URL, error) { + return nil, errors.New("agent timed out") + } + Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "al", ItemID: "al4"})).To(Succeed()) + + n, err := w.drain(ctx, 2) + Expect(err).ToNot(HaveOccurred()) + Expect(n).To(Equal(1)) + + it := findQueued(queueRepo, "al", "al4") + Expect(it).ToNot(BeNil()) + Expect(it.Attempts).To(Equal(1)) + Expect(it.RetryAt).To(BeTemporally(">", time.Now())) + + _, err = artRepo.GetItemArtwork("al", "al4", model.ImageTypePrimary) + Expect(err).To(MatchError(model.ErrNotFound), "a timeout must never settle on absent") + }) + + It("returns zero when the queue is empty", func() { + n, err := w.drain(ctx, 2) + Expect(err).ToNot(HaveOccurred()) + Expect(n).To(BeZero()) + }) + }) + + Describe("Bump", func() { + It("enqueues at Bump priority and wakes the loop", func() { + w.Bump("al", "al9") + it := findQueued(queueRepo, "al", "al9") + Expect(it).ToNot(BeNil()) + Expect(it.Priority).To(Equal(model.ArtworkPriorityBump)) + }) + }) + + Describe("gate/breaker", func() { + It("opens after 5 consecutive external errors and short-circuits the step", func() { + var calls int + failing := func() (io.ReadCloser, string, error) { + calls++ + return nil, "", errors.New("boom") + } + for range 5 { + _, _, err := w.gate(failing) + Expect(err).To(HaveOccurred()) + } + Expect(calls).To(Equal(5)) + + _, _, err := w.gate(failing) + Expect(err).To(HaveOccurred()) + Expect(calls).To(Equal(5), "an open breaker must not call the external step") + }) + + It("resets the failure count on a successful call", func() { + failing := func() (io.ReadCloser, string, error) { return nil, "", errors.New("boom") } + ok := func() (io.ReadCloser, string, error) { return io.NopCloser(nil), "p", nil } + for range 4 { + _, _, _ = w.gate(failing) + } + _, _, err := w.gate(ok) + Expect(err).ToNot(HaveOccurred()) + + var calls int + counting := func() (io.ReadCloser, string, error) { + calls++ + return nil, "", errors.New("boom") + } + for range 5 { + _, _, _ = w.gate(counting) + } + Expect(calls).To(Equal(5), "the breaker should have re-closed after the success") + }) + }) + + Describe("RunPrune", func() { + It("runs a prune under the worker mutex", func() { + Expect(w.RunPrune(ctx)).To(Succeed()) + }) + }) + + Describe("Run", func() { + It("exits cleanly when the context is cancelled", func() { + runCtx, cancel := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- w.Run(runCtx) }() + + cancel() + Eventually(done, time.Second).Should(Receive(BeNil())) + }) + }) +}) diff --git a/core/artwork/worker_timing_test.go b/core/artwork/worker_timing_test.go new file mode 100644 index 000000000..a721459c2 --- /dev/null +++ b/core/artwork/worker_timing_test.go @@ -0,0 +1,95 @@ +package artwork + +import ( + "context" + "errors" + "testing" + "testing/synctest" + "time" + + "github.com/navidrome/navidrome/conf/configtest" + "github.com/navidrome/navidrome/tests" + . "github.com/onsi/gomega" + "go.uber.org/goleak" +) + +func TestArtworkBackoffSchedule(t *testing.T) { + g := NewWithT(t) + for _, c := range []struct { + attempts int + want time.Duration + }{ + {0, 5 * time.Minute}, + {1, 20 * time.Minute}, + {2, 80 * time.Minute}, + {3, 320 * time.Minute}, + {4, 1280 * time.Minute}, + {5, 48 * time.Hour}, + {6, 48 * time.Hour}, + } { + g.Expect(backoffFor(c.attempts, 0)).To(Equal(c.want), "attempt %d", c.attempts) + } + + base := backoffFor(2, 0) + g.Expect(backoffFor(2, 0.2)).To(Equal(time.Duration(float64(base) * 1.2))) + g.Expect(backoffFor(2, -0.2)).To(Equal(time.Duration(float64(base) * 0.8))) + + lo := time.Duration(float64(320*time.Minute) * 0.8) + hi := time.Duration(float64(320*time.Minute) * 1.2) + for range 200 { + d := backoff(3) + g.Expect(d).To(BeNumerically(">=", lo)) + g.Expect(d).To(BeNumerically("<=", hi)) + } +} + +func TestArtworkBreakerHalfOpen(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + g := NewWithT(t) + b := newBreaker() + + for range breakerThreshold { + b.record(errors.New("boom")) + } + g.Expect(b.allow()).To(BeFalse(), "breaker opens after consecutive errors") + + time.Sleep(breakerProbeAfter - time.Nanosecond) + g.Expect(b.allow()).To(BeFalse(), "still open before the probe interval") + + time.Sleep(time.Nanosecond) + g.Expect(b.allow()).To(BeTrue(), "half-open: one probe is granted") + g.Expect(b.allow()).To(BeFalse(), "only a single probe per interval") + + b.record(errors.New("boom")) // probe fails -> stay open + time.Sleep(breakerProbeAfter) + g.Expect(b.allow()).To(BeTrue(), "another probe after the next interval") + + b.record(nil) // probe succeeds -> close + g.Expect(b.allow()).To(BeTrue(), "closed breaker admits freely") + g.Expect(b.allow()).To(BeTrue()) + }) +} + +func TestArtworkWorkerRunNoLeak(t *testing.T) { + ignore := goleak.IgnoreCurrent() + defer goleak.VerifyNone(t, ignore) + + t.Cleanup(configtest.SetupConfig()) + ds := &tests.MockDataStore{MockedArtworkQueue: tests.CreateMockArtworkQueueRepo()} + w := NewWorker(ds, NewImageStore(t.TempDir()), &fakeExternalProvider{}, tests.NewMockFFmpeg("")) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- w.Run(ctx) }() + + time.Sleep(20 * time.Millisecond) // let the loop settle on the idle select + cancel() + select { + case err := <-done: + if err != nil { + t.Fatalf("Run returned error on cancel: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Run did not exit after context cancel") + } +}