refactor(artwork): narrow the prune lock and drop redundant in-flight tracking

The prune read-lock wrapped all of processItem, including external fetches under
their own timeout. Since a pending RWMutex writer blocks new readers, one prune
arriving behind a slow provider stalled every subsequent item in both drain
pools. Extract persist() so the lock covers only the window it protects: store
placement plus the two row writes.

The in-flight set guarded against a queue row appearing twice in one batch, but
artwork_queue's primary key makes that impossible, drains are serial per pool,
and the pools' kind lists are disjoint. Removing it also retires the cancellation
unwind loop that existed only to release those claims.

core/artwork/processor.go

@@ -9,6 +9,7 @@ import (
 	"image/draw"
 	_ "image/gif" // the only artwork format with no other importer in this package
 	"io"
+	"sync"
 	"time"
 
 	"github.com/navidrome/navidrome/core/agents"
@@ -44,15 +45,17 @@ const maxImageBytes = 20 << 20
 // huge canvas that image.Decode would expand into gigabytes (decompression bomb).
 const maxImagePixels = 64 << 20
 
-// workerDeps are the collaborators processItem needs; gate is set by NewWorker in
-// production and nil only in tests, where resolveItem falls back to a plain passthrough.
+// workerDeps are the collaborators processItem needs; gate and pruneLock are set by NewWorker
+// in production and nil only in tests, where resolveItem falls back to a plain passthrough and
+// nothing prunes.
 type workerDeps struct {
-	ds     model.DataStore
-	store  *ImageStore
-	agents *agents.Agents
-	ffmpeg ffmpeg.FFmpeg
-	cache  cache.FileCache
-	gate   gateFunc
+	ds        model.DataStore
+	store     *ImageStore
+	agents    *agents.Agents
+	ffmpeg    ffmpeg.FFmpeg
+	cache     cache.FileCache
+	gate      gateFunc
+	pruneLock sync.Locker
 }
 
 // acquired is what processItem persisted, handed back so the caller can warm the resize
@@ -112,14 +115,34 @@ func processItem(ctx context.Context, deps *workerDeps, item model.ArtworkQueueI
 	}
 	art.SizeBytes = int64(len(data))
 
-	sourcePath, refMtime, err := placeBytes(deps.store, art, res, data)
+	ia, err := persist(deps, repo, item, hash, art, res, data)
 	if err != nil {
-		log.Warn(ctx, "artwork: failed to write image store", "kind", item.ItemKind, "id", item.ItemID, err)
+		log.Warn(ctx, "artwork: failed to persist resolved image", "kind", item.ItemKind, "id", item.ItemID, err)
 		return outcomeFailed, nil
 	}
+	got := &acquired{ia: ia, mime: art.Mime, data: data}
+	if res.extError {
+		return outcomeFoundStale, got
+	}
+	return outcomeFound, got
+}
+
+// persist places the bytes and commits the rows referencing them. Only this window excludes
+// Prune, which reclaims store files no row points at; resolution stays outside so a slow fetch
+// cannot hold prune off.
+func persist(deps *workerDeps, repo model.ArtworkRepository, item model.ArtworkQueueItem, hash string,
+	art *model.Artwork, res resolution, data []byte,
+) (*model.ItemArtwork, error) {
+	if deps.pruneLock != nil {
+		deps.pruneLock.Lock()
+		defer deps.pruneLock.Unlock()
+	}
+	sourcePath, refMtime, err := placeBytes(deps.store, art, res, data)
+	if err != nil {
+		return nil, fmt.Errorf("writing image store: %w", err)
+	}
 	if err := repo.PutImage(art); err != nil {
-		log.Warn(ctx, "artwork: failed to persist artwork image", "kind", item.ItemKind, "id", item.ItemID, err)
-		return outcomeFailed, nil
+		return nil, fmt.Errorf("persisting artwork image: %w", err)
 	}
 	ia := &model.ItemArtwork{
 		ItemKind:    item.ItemKind,
@@ -133,14 +156,9 @@ func processItem(ctx context.Context, deps *workerDeps, item model.ArtworkQueueI
 	}
 	// PutItemArtwork stamps UpdatedAt on ia, so what it holds now matches the persisted row.
 	if err := repo.PutItemArtwork(ia); err != nil {
-		log.Warn(ctx, "artwork: failed to persist item artwork state", "kind", item.ItemKind, "id", item.ItemID, err)
-		return outcomeFailed, nil
+		return nil, fmt.Errorf("persisting item artwork state: %w", err)
 	}
-	got := &acquired{ia: ia, mime: art.Mime, data: data}
-	if res.extError {
-		return outcomeFoundStale, got
-	}
-	return outcomeFound, got
+	return ia, nil
 }
 
 // writeAbsent records a known-absent state: every local/external source answered definitively "no".

core/artwork/processor_test.go

@@ -101,6 +101,43 @@ var _ = Describe("processItem", func() {
 		Expect(os.IsNotExist(err)).To(BeTrue(), "folder-backed art must not be duplicated into the store")
 	})
 
+	Describe("prune lock scope", func() {
+		var lock *countingLocker
+
+		BeforeEach(func() {
+			lock = &countingLocker{}
+			deps.pruneLock = lock
+		})
+
+		It("holds it once across the write window", func() {
+			folderRepo.result = []model.Folder{{
+				Path:       "tests/fixtures/artist/an-album",
+				ImageFiles: []string{"cover.jpg"},
+			}}
+			ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
+				{ID: "alL1", Name: "Album", FolderIDs: []string{"f1"}},
+			})
+
+			out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "alL1"})
+			Expect(out).To(Equal(outcomeFound))
+			Expect(lock.locks).To(BeNumerically(">", 0), "the write window must exclude prune")
+			Expect(lock.held()).To(BeFalse(), "the window must close before processItem returns")
+		})
+
+		// Resolution can reach the network under its own timeout, so holding the lock across it
+		// would let one slow provider block prune, and every drain queued behind prune's writer.
+		It("never takes it while only resolving", func() {
+			folderRepo.result = nil
+			ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
+				{ID: "alL2", Name: "Album", FolderIDs: []string{"f1"}},
+			})
+
+			out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "alL2"})
+			Expect(out).To(Equal(outcomeAbsent))
+			Expect(lock.locks).To(BeZero())
+		})
+	})
+
 	It("found-embedded: writes a store file and computes a non-empty blurhash from a real fixture", func() {
 		ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
 			{ID: "al2", Name: "Album", EmbedArtPath: "tests/fixtures/artist/an-album/test.mp3", FolderIDs: []string{"f1"}},
@@ -413,3 +450,14 @@ var _ = Describe("processItem", func() {
 		Expect(err).To(MatchError(model.ErrNotFound))
 	})
 })
+
+// countingLocker stands in for the worker's prune read-lock, recording how often and how long
+// processItem holds it.
+type countingLocker struct {
+	locks   int
+	unlocks int
+}
+
+func (l *countingLocker) Lock()      { l.locks++ }
+func (l *countingLocker) Unlock()    { l.unlocks++ }
+func (l *countingLocker) held() bool { return l.locks != l.unlocks }

core/artwork/worker.go

@@ -50,7 +50,7 @@ type drainPool struct {
 }
 
 // Worker drains the artwork queue through processItem: each external agent is rate-limited
-// and circuit-broken independently, and prune is serialized against in-flight acquisitions
+// and circuit-broken independently, and prune is serialized against the store-write window
 // via pruneMu.
 type Worker struct {
 	deps    workerDeps
@@ -61,21 +61,18 @@ type Worker struct {
 
 	gatesMu sync.Mutex
 	gates   map[string]*extGate
-
-	mu       sync.Mutex
-	inFlight map[string]struct{}
 }
 
 func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg ffmpeg.FFmpeg, broker events.Broker, imgCache cache.FileCache) *Worker {
 	w := &Worker{
-		deps:     workerDeps{ds: ds, store: store, agents: ag, ffmpeg: ffmpeg, cache: imgCache},
-		broker:   broker,
-		pools:    newDrainPools(),
-		runCtx:   context.Background(),
-		gates:    map[string]*extGate{},
-		inFlight: map[string]struct{}{},
+		deps:   workerDeps{ds: ds, store: store, agents: ag, ffmpeg: ffmpeg, cache: imgCache},
+		broker: broker,
+		pools:  newDrainPools(),
+		runCtx: context.Background(),
+		gates:  map[string]*extGate{},
 	}
 	w.deps.gate = w.gate
+	w.deps.pruneLock = w.pruneMu.RLocker()
 	return w
 }
 
@@ -175,11 +172,10 @@ func (w *Worker) drain(ctx context.Context, concurrency int, kinds ...string) (i
 	// Dequeue well past the worker pool so a slow item (an external lookup burning its
 	// timeout) never idles the other slots: the pool stays fed until the batch runs out.
 	// DequeueBatch does not mark rows taken, so this is one query per pass, not per slot.
-	batch, err := w.deps.ds.ArtworkQueue(ctx).DequeueBatch(max(16, 4*concurrency), kinds...)
+	items, err := w.deps.ds.ArtworkQueue(ctx).DequeueBatch(max(16, 4*concurrency), kinds...)
 	if err != nil {
 		return 0, err
 	}
-	items := w.claim(batch)
 	if len(items) == 0 {
 		return 0, nil
 	}
@@ -190,30 +186,22 @@ func (w *Worker) drain(ctx context.Context, concurrency int, kinds ...string) (i
 	var wg sync.WaitGroup
 	var refreshMu sync.Mutex
 	var refresh []model.ArtworkQueueItem
-	for i, item := range items {
+	for _, item := range items {
 		select {
 		case sem <- struct{}{}:
 		case <-ctx.Done():
-			// claim() reserved the whole batch; anything not dispatched has to go back, or it
-			// stays in flight forever and no later drain can pick it up.
-			for _, undispatched := range items[i:] {
-				w.release(undispatched)
-			}
 			wg.Wait()
 			return len(items), nil
 		}
-		wg.Add(1)
-		go func(it model.ArtworkQueueItem) {
-			defer wg.Done()
+		wg.Go(func() {
 			defer func() { <-sem }()
-			defer w.release(it)
-			out, got := w.process(ctx, it)
+			out, got := w.process(ctx, item)
 			// Refresh clients on any visible state change: found/foundStale (new art) and absent
 			// (removed art — clients must drop a previously-served immutable cover). foundStale
 			// also wrote a served state row.
 			if out == outcomeFound || out == outcomeFoundStale || out == outcomeAbsent {
 				refreshMu.Lock()
-				refresh = append(refresh, it)
+				refresh = append(refresh, item)
 				refreshMu.Unlock()
 			}
 			// Precache only actual images. Post-outcome only: the queue row was already settled
@@ -221,7 +209,7 @@ func (w *Worker) drain(ctx context.Context, concurrency int, kinds ...string) (i
 			if got != nil {
 				w.precache(ctx, got)
 			}
-		}(item)
+		})
 	}
 	wg.Wait()
 	w.broadcastRefresh(ctx, refresh)
@@ -265,9 +253,7 @@ func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) (outc
 	if item.ImageType == "" {
 		item.ImageType = model.ImageTypePrimary
 	}
-	w.pruneMu.RLock()
 	out, got := processItem(ctx, &w.deps, item)
-	w.pruneMu.RUnlock()
 
 	queue := w.deps.ds.ArtworkQueue(ctx)
 	switch out {
@@ -337,33 +323,6 @@ func (w *Worker) precache(ctx context.Context, got *acquired) {
 	_ = stream.Close()
 }
 
-// claim reserves items not already in flight, so a row appearing twice within a single
-// batch is processed once.
-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 a named external step with that agent's own rate limiter and circuit
 // breaker, matching gateFunc so it can be injected via workerDeps.gate.
 func (w *Worker) gate(name string, f func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) {

core/artwork/worker_test.go

@@ -673,9 +673,9 @@ var _ = Describe("Worker", func() {
 	})
 
 	Describe("batching", func() {
-		// claim() reserves the whole batch before dispatching, so a cancel mid-batch used to
-		// strand the rest in the in-flight set, where no later drain could ever claim them.
-		It("returns undispatched items to the pool when cancelled mid-batch", func() {
+		// A cancelled drain leaves its undispatched rows untouched in the queue, so a later
+		// drain picks them up unchanged.
+		It("leaves undispatched items queued when cancelled mid-batch", func() {
 			for i := range 8 {
 				id := fmt.Sprintf("alc%d", i)
 				ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{{ID: id, Name: "Album"}})
@@ -689,11 +689,10 @@ var _ = Describe("Worker", func() {
 			_, err := w.drain(cancelledCtx, 1)
 			Expect(err).ToNot(HaveOccurred())
 
-			// Every item must be claimable again; a stranded one would be silently skipped.
-			w.mu.Lock()
-			stranded := len(w.inFlight)
-			w.mu.Unlock()
-			Expect(stranded).To(BeZero(), "a cancelled drain must not strand claimed items")
+			for i := range 8 {
+				id := fmt.Sprintf("alc%d", i)
+				Expect(findQueued(queueRepo, "al", id)).ToNot(BeNil(), "row "+id+" must survive a cancelled drain")
+			}
 		})
 
 		// The pool is fed from one dequeue per pass: a batch sized to the pool would make a