From 89e17f7ed888bbc67d1aa8fea9e3903af16d1e62 Mon Sep 17 00:00:00 2001
From: Deluan <deluan@navidrome.org>
Date: Sun, 27 Sep 2026 15:51:50 -0400
Subject: [PATCH] refactor(artwork): cancel a timed-out item's context and
 simplify the busy set

Give acquire a context with the same itemTimeout deadline, so network and
ffmpeg work that honors ctx stops when the drain moves on. The select still
covers work that ignores ctx, like the filepath.Rel loop.

Replace the busy mutex, map and helpers with a sync.Map, use time.After,
record the timeout and still-running steps with traceStage, and let settle
read the trace from ctx.
---
 core/artwork/worker.go | 50 ++++++++++++++----------------------------
 1 file changed, 17 insertions(+), 33 deletions(-)

diff --git a/core/artwork/worker.go b/core/artwork/worker.go
index 198fcc87a..8c2f2bd40 100644
--- a/core/artwork/worker.go
+++ b/core/artwork/worker.go
@@ -4,6 +4,8 @@ import (
 	"bytes"
 	"cmp"
 	"context"
+	"errors"
+	"fmt"
 	"io"
 	"math"
 	"math/rand/v2"
@@ -53,9 +55,8 @@ type Worker struct {
 	gatesMu sync.Mutex
 	gates   map[string]*extGate
 
-	// busy holds the items still being acquired, including ones a drain stopped waiting for.
-	busyMu sync.Mutex
-	busy   map[itemKey]struct{}
+	// busy holds the itemKeys still being acquired, including ones a drain stopped waiting for.
+	busy sync.Map
 }
 
 type itemKey struct{ kind, id, imageType string }
@@ -70,7 +71,6 @@ func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg
 		runCtx: context.Background(),
 		paused: func() bool { return false },
 		gates:  map[string]*extGate{},
-		busy:   map[itemKey]struct{}{},
 	}
 	w.proc.resolver = newResolver(ds, ag, ffmpeg, w.gate)
 	w.proc.pruneLock = w.pruneMu.RLocker()
@@ -251,19 +251,18 @@ func (w *Worker) broadcastRefresh(ctx context.Context, found []model.ArtworkQueu
 
 func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) (outcome, *acquired) {
 	item.ImageType = cmp.Or(item.ImageType, model.ImageTypePrimary)
-	trace := &ChainTrace{}
-	ctx = withTrace(ctx, trace)
+	ctx = withTrace(ctx, &ChainTrace{})
 	out, got, retryIn := w.acquireWithTimeout(ctx, item)
-	w.settle(ctx, item, out, retryIn, trace)
+	w.settle(ctx, item, out, retryIn)
 	return out, got
 }
 
 // acquireWithTimeout stops waiting on a stuck item, which Go can't stop, and never runs one item twice at once.
 func (w *Worker) acquireWithTimeout(ctx context.Context, item model.ArtworkQueueItem) (outcome, *acquired, time.Duration) {
 	key := itemKey{item.ItemKind, item.ItemID, item.ImageType}
-	if !w.markBusy(key) {
+	if _, running := w.busy.LoadOrStore(key, struct{}{}); running {
 		log.Debug(ctx, "Artwork: Item still running from an earlier drain", "kind", item.ItemKind, "id", item.ItemID)
-		traceFrom(ctx).add(TraceStep{Candidate: "worker", Outcome: OutcomeError, Detail: "previous attempt still running"})
+		traceStage(ctx, "worker", errors.New("previous attempt still running"))
 		return outcomeFailed, nil, 0
 	}
 	type result struct {
@@ -273,41 +272,26 @@ func (w *Worker) acquireWithTimeout(ctx context.Context, item model.ArtworkQueue
 	}
 	done := make(chan result, 1)
 	go func() {
-		out, got, retryIn := w.proc.acquire(ctx, item)
+		// The deadline stops work that honors ctx; the select below covers work that doesn't.
+		actx, cancel := context.WithTimeout(ctx, itemTimeout)
+		defer cancel()
+		out, got, retryIn := w.proc.acquire(actx, item)
 		// Cleared before the send, so the next drain never sees a finished item as busy.
-		w.clearBusy(key)
+		w.busy.Delete(key)
 		done <- result{out, got, retryIn}
 	}()
-	timer := time.NewTimer(itemTimeout)
-	defer timer.Stop()
 	select {
 	case r := <-done:
 		return r.out, r.got, r.retryIn
-	case <-timer.C:
+	case <-time.After(itemTimeout):
 		log.Error(ctx, "Artwork: Item timed out, moving on", "kind", item.ItemKind, "id", item.ItemID, "timeout", itemTimeout)
-		traceFrom(ctx).add(TraceStep{Candidate: "worker", Outcome: OutcomeError, Detail: "timed out after " + itemTimeout.String()})
+		traceStage(ctx, "worker", fmt.Errorf("timed out after %s", itemTimeout))
 		return outcomeFailed, nil, 0
 	}
 }
 
-func (w *Worker) markBusy(key itemKey) bool {
-	w.busyMu.Lock()
-	defer w.busyMu.Unlock()
-	if _, ok := w.busy[key]; ok {
-		return false
-	}
-	w.busy[key] = struct{}{}
-	return true
-}
-
-func (w *Worker) clearBusy(key itemKey) {
-	w.busyMu.Lock()
-	defer w.busyMu.Unlock()
-	delete(w.busy, key)
-}
-
 // settle deletes or reschedules the queue row for an acquire outcome.
-func (w *Worker) settle(ctx context.Context, item model.ArtworkQueueItem, out outcome, retryIn time.Duration, trace *ChainTrace) {
+func (w *Worker) settle(ctx context.Context, item model.ArtworkQueueItem, out outcome, retryIn time.Duration) {
 	queue := w.proc.ds.ArtworkQueue()
 	switch out {
 	case outcomeFound, outcomeAbsent:
@@ -318,7 +302,7 @@ func (w *Worker) settle(ctx context.Context, item model.ArtworkQueueItem, out ou
 		}
 	case outcomeFoundStale, outcomeFailed:
 		retryAt := time.Now().Add(retryDelay(item.Attempts, retryIn))
-		encoded := trace.encode("")
+		encoded := traceFrom(ctx).encode("")
 		if retryAt.Before(item.EnqueuedAt.Add(giveUpAfter)) {
 			// A mid-flight re-enqueue reset retry_at; stale backoff must not stomp its
 			// fresh, immediate eligibility.