feat(artwork): add acquisition worker service

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 {

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)

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,

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)

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 {

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)
 			}

core/artwork/wire_providers.go

@@ -8,4 +8,6 @@ var Set = wire.NewSet(
 	NewArtwork,
 	GetImageCache,
 	NewCacheWarmer,
+	NewWorker,
+	ProvideImageStore,
 )

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()
+	}
+}

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()))
+		})
+	})
+})

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")
+	}
+}