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.