diff --git a/core/artwork/worker.go b/core/artwork/worker.go index 28e51958c..198fcc87a 100644 --- a/core/artwork/worker.go +++ b/core/artwork/worker.go @@ -26,6 +26,8 @@ const ( // giveUpAfter bounds the retry budget from enqueue; past it the item settles and only an // explicit reprocess retries it. giveUpAfter = 12 * time.Hour + // itemTimeout is how long a drain waits on one item before rescheduling it. + itemTimeout = 5 * time.Minute ) // drainPool drains one class of work with its own slot budget, so a blocking kind cannot @@ -50,8 +52,14 @@ 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{} } +type itemKey struct{ kind, id, imageType string } + func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg ffmpeg.FFmpeg, broker events.Broker, imgCache cache.FileCache) *Worker { w := &Worker{ proc: &processor{ds: ds, store: store}, @@ -62,6 +70,7 @@ 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() @@ -244,8 +253,61 @@ func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) (outc item.ImageType = cmp.Or(item.ImageType, model.ImageTypePrimary) trace := &ChainTrace{} ctx = withTrace(ctx, trace) - out, got, retryIn := w.proc.acquire(ctx, item) + out, got, retryIn := w.acquireWithTimeout(ctx, item) + w.settle(ctx, item, out, retryIn, trace) + 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) { + 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"}) + return outcomeFailed, nil, 0 + } + type result struct { + out outcome + got *acquired + retryIn time.Duration + } + done := make(chan result, 1) + go func() { + out, got, retryIn := w.proc.acquire(ctx, item) + // Cleared before the send, so the next drain never sees a finished item as busy. + w.clearBusy(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: + 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()}) + 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) { queue := w.proc.ds.ArtworkQueue() switch out { case outcomeFound, outcomeAbsent: @@ -283,7 +345,6 @@ func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) (outc log.Warn(ctx, "Artwork: Could not remove exhausted queue item", "kind", item.ItemKind, "id", item.ItemID, err) } } - return out, got } // recordGiveUp keeps the last failure on the state row after the queue row is deleted. An item diff --git a/core/artwork/worker_timing_test.go b/core/artwork/worker_timing_test.go index 63f12d03c..d991fc38d 100644 --- a/core/artwork/worker_timing_test.go +++ b/core/artwork/worker_timing_test.go @@ -7,7 +7,10 @@ import ( "testing/synctest" "time" + "github.com/navidrome/navidrome/conf" + "github.com/navidrome/navidrome/conf/configtest" "github.com/navidrome/navidrome/core/agents" + "github.com/navidrome/navidrome/model" "github.com/navidrome/navidrome/tests" . "github.com/onsi/gomega" ) @@ -113,3 +116,54 @@ func TestArtworkGatePerAgentBreakerIsolation(t *testing.T) { g.Expect(aCalls).To(Equal(1)) }) } + +// Go can't stop a stuck item, so the drain must stop waiting for it, or the whole pool stalls. +func TestArtworkDrainMovesOnFromAStuckItem(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + g := NewWithT(t) + defer configtest.SetupConfig()() + conf.Server.ArtistArtPriority = "external" + conf.Server.DevArtworkExternalMaxRPS = 1000 + + block := make(chan struct{}) + agent := &fakeImageAgent{name: "stuckAgent", block: block} + ag := imageAgents(agent) + artists := tests.CreateMockArtistRepo() + artists.SetData(model.Artists{{ID: "ar1", Name: "Artist"}}) + queue := tests.CreateMockArtworkQueueRepo() + ds := &tests.MockDataStore{MockedArtist: artists, MockedFolder: &fakeFolderRepo{}, MockedArtwork: tests.CreateMockArtworkRepo(), MockedArtworkQueue: queue} + w := NewWorker(ds, NewImageStore(t.TempDir()), ag, tests.NewMockFFmpeg(""), &fakeEventBroker{}, nil) + ctx := t.Context() + g.Expect(queue.Enqueue(ctx, model.ArtworkQueueItem{ItemKind: "ar", ItemID: "ar1"})).To(Succeed()) + + start := time.Now() + n, err := w.drain(ctx, 1) + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(n).To(Equal(1)) + g.Expect(time.Since(start)).To(Equal(itemTimeout), "the drain stops waiting once the item times out") + row := findQueued(queue, "ar", "ar1") + g.Expect(row).ToNot(BeNil(), "a timed-out item is rescheduled, not dropped") + g.Expect(row.Attempts).To(Equal(1)) + g.Expect(row.RetryAt).To(BeTemporally(">", time.Now())) + g.Expect(row.Trace).To(ContainSubstring("timed out")) + + // Due again while the first run is still stuck: it must not start a second one. + time.Sleep(time.Until(row.RetryAt)) + start = time.Now() + _, err = w.drain(ctx, 1) + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(time.Since(start)).To(BeZero(), "a still-running item is rescheduled without waiting") + row = findQueued(queue, "ar", "ar1") + g.Expect(row.Attempts).To(Equal(2)) + g.Expect(row.Trace).To(ContainSubstring("still running")) + + // Once the stuck run returns, the next attempt runs normally and settles the row. + close(block) + synctest.Wait() + time.Sleep(time.Until(row.RetryAt)) + _, err = w.drain(ctx, 1) + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(findQueued(queue, "ar", "ar1")).To(BeNil()) + g.Expect(agent.artistCalls).To(Equal(2), "one stuck call and one retry, never two at once") + }) +}