diff --git a/core/artwork/worker.go b/core/artwork/worker.go index bc3e47af0..4be32fee8 100644 --- a/core/artwork/worker.go +++ b/core/artwork/worker.go @@ -155,8 +155,10 @@ func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) { log.Warn(ctx, "artwork: could not delete processed queue item", "kind", item.ItemKind, "id", item.ItemID, err) } case outcomeFoundStale, outcomeFailed: + // MarkFailedIfUnchanged, not MarkFailed: a scan that re-enqueued this row mid-flight reset + // retry_at, so stale backoff must not stomp its fresh, immediate eligibility. retryAt := time.Now().Add(backoff(item.Attempts)) - if err := queue.MarkFailed(item.ItemKind, item.ItemID, item.ImageType, retryAt); err != nil { + if err := queue.MarkFailedIfUnchanged(item.ItemKind, item.ItemID, item.ImageType, item.RetryAt, retryAt); err != nil { log.Warn(ctx, "artwork: could not reschedule failed queue item", "kind", item.ItemKind, "id", item.ItemID, err) } } diff --git a/core/artwork/worker_test.go b/core/artwork/worker_test.go index 7865b6cfc..1fe904e21 100644 --- a/core/artwork/worker_test.go +++ b/core/artwork/worker_test.go @@ -189,6 +189,30 @@ var _ = Describe("Worker", func() { Expect(ia.Source).To(Equal("folder")) }) + It("keeps a fresh re-enqueue ahead of a stale failure backoff", func() { + conf.Server.CoverArtPriority = "external" + ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{{ID: "al8", Name: "Album"}}) + prov.albumImage = func(context.Context, string) (*url.URL, error) { + return nil, errors.New("agent timed out") + } + racing := &reenqueueOnDequeue{MockArtworkQueueRepo: queueRepo} + ds.MockedArtworkQueue = racing + w = NewWorker(ds, store, prov, ffm) + Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "al", ItemID: "al8"})).To(Succeed()) + dequeued := findQueued(queueRepo, "al", "al8").RetryAt + + n, err := w.drain(ctx, 1) + Expect(err).ToNot(HaveOccurred()) + Expect(n).To(Equal(1)) + + // The concurrent re-enqueue reset retry_at; the failure path must not stomp it + // with stale backoff nor bump attempts, so the row stays immediately eligible. + it := findQueued(queueRepo, "al", "al8") + Expect(it).ToNot(BeNil()) + Expect(it.Attempts).To(BeZero()) + Expect(it.RetryAt).To(BeTemporally("==", dequeued.Add(time.Minute))) + }) + It("returns zero when the queue is empty", func() { n, err := w.drain(ctx, 2) Expect(err).ToNot(HaveOccurred()) diff --git a/model/artwork.go b/model/artwork.go index 0aec6489a..8db11ce69 100644 --- a/model/artwork.go +++ b/model/artwork.go @@ -88,6 +88,9 @@ type ArtworkQueueRepository interface { DequeueBatch(n int) ([]ArtworkQueueItem, error) // MarkFailed increments attempts and pushes retry_at into the future. MarkFailed(kind, id, imageType string, retryAt time.Time) error + // MarkFailedIfUnchanged applies the failure backoff only while retry_at still matches + // seenRetryAt; a concurrent re-enqueue (which resets retry_at) keeps its fresh eligibility. + MarkFailedIfUnchanged(kind, id, imageType string, seenRetryAt, retryAt time.Time) error Delete(kind, id, imageType string) error // DeleteIfUnchanged deletes the row only if its retry_at still matches retryAt, so a // concurrent re-enqueue (which resets retry_at) survives instead of being erased. diff --git a/persistence/artwork_queue_repository.go b/persistence/artwork_queue_repository.go index 4a6a68eb1..59e130204 100644 --- a/persistence/artwork_queue_repository.go +++ b/persistence/artwork_queue_repository.go @@ -66,6 +66,17 @@ func (r *artworkQueueRepository) MarkFailed(kind, id, imageType string, retryAt return err } +// MarkFailedIfUnchanged applies the backoff only while retry_at still equals seenRetryAt; +// a concurrent Enqueue resets retry_at, so its fresh eligibility survives untouched. +func (r *artworkQueueRepository) MarkFailedIfUnchanged(kind, id, imageType string, seenRetryAt, retryAt time.Time) error { + upd := Update(r.tableName). + Set("attempts", Expr("attempts + 1")). + Set("retry_at", retryAt). + Where(Eq{"item_kind": kind, "item_id": id, "image_type": imageType, "retry_at": seenRetryAt}) + _, err := r.executeSQL(upd) + return err +} + func (r *artworkQueueRepository) Delete(kind, id, imageType string) error { return r.delete(Eq{"item_kind": kind, "item_id": id, "image_type": imageType}) } diff --git a/persistence/artwork_queue_repository_test.go b/persistence/artwork_queue_repository_test.go index e96802284..0a6ddefdc 100644 --- a/persistence/artwork_queue_repository_test.go +++ b/persistence/artwork_queue_repository_test.go @@ -54,6 +54,34 @@ var _ = Describe("ArtworkQueueRepository", func() { Expect(got[0].Attempts).To(Equal(2)) }) + It("MarkFailedIfUnchanged applies backoff only while retry_at is unchanged", func() { + Expect(repo.Enqueue(item("al", "m1", model.ArtworkPriorityScan))).To(Succeed()) + // Anchor retry_at in the past (attempts -> 1) so it can never collide with the re-enqueue's now. + Expect(repo.MarkFailed("al", "m1", model.ImageTypePrimary, time.Now().Add(-time.Hour))).To(Succeed()) + got, err := repo.DequeueBatch(10) + Expect(err).ToNot(HaveOccurred()) + Expect(got).To(HaveLen(1)) + original := got[0].RetryAt + + // A concurrent scan re-enqueues, resetting retry_at to now. + Expect(repo.Enqueue(item("al", "m1", model.ArtworkPriorityScan))).To(Succeed()) + + // Failing with the stale retry_at is a no-op: the re-enqueued row keeps its fresh state. + future := time.Now().Add(48 * time.Hour) + Expect(repo.MarkFailedIfUnchanged("al", "m1", model.ImageTypePrimary, original, future)).To(Succeed()) + got, _ = repo.DequeueBatch(10) + Expect(got).To(HaveLen(1), "the fresh re-enqueue stays immediately eligible") + Expect(got[0].Attempts).To(Equal(1), "the stale failure must not bump attempts") + current := got[0].RetryAt + + // Failing with the current retry_at applies the backoff and bumps attempts. + Expect(repo.MarkFailedIfUnchanged("al", "m1", model.ImageTypePrimary, current, future)).To(Succeed()) + got, _ = repo.DequeueBatch(10) + Expect(got).To(BeEmpty(), "backed-off row is hidden until the future retry_at") + all, _ := repo.Count() + Expect(all).To(Equal(int64(1))) + }) + It("deletes on completion and counts", func() { Expect(repo.Enqueue(item("al", "c1", 0))).To(Succeed()) n, _ := repo.Count() diff --git a/tests/mock_artwork_queue_repo.go b/tests/mock_artwork_queue_repo.go index 8d01ef9c0..421e0dd15 100644 --- a/tests/mock_artwork_queue_repo.go +++ b/tests/mock_artwork_queue_repo.go @@ -84,6 +84,19 @@ func (m *MockArtworkQueueRepo) MarkFailed(kind, id, imageType string, retryAt ti return nil } +func (m *MockArtworkQueueRepo) MarkFailedIfUnchanged(kind, id, imageType string, seenRetryAt, retryAt time.Time) error { + if m.Err != nil { + return m.Err + } + k := iaKey(kind, id, imageType) + if it, ok := m.Data[k]; ok && it.RetryAt.Equal(seenRetryAt) { + it.Attempts++ + it.RetryAt = retryAt + m.Data[k] = it + } + return nil +} + func (m *MockArtworkQueueRepo) Delete(kind, id, imageType string) error { if m.Err != nil { return m.Err