remote/navidrome
1
0
Fork 0
mirror of https://github.com/navidrome/navidrome.git synced 2026-10-13 21:07:10 +02:00
Code Issues Projects Releases Packages Wiki Activity

refactor(artwork): cancel a timed-out item's context and simplify the busy set

Browse source 
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.
This commit is contained in:
Deluan 2026-09-27 15:51:50 -04:00
commit 89e17f7ed8
1 changed files with 17 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

50
core/artwork/worker.go
View file

@ -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.