diff --git a/core/artwork/serving.go b/core/artwork/serving.go index 38a474355..109d0033d 100644 --- a/core/artwork/serving.go +++ b/core/artwork/serving.go @@ -286,8 +286,10 @@ func (s *service) dangling(ctx context.Context, artID model.ArtworkID) (*Image, return nil, ErrUnavailable } +// enqueue schedules a request-triggered re-resolution. It uses EnqueueBump so an incidental +// read-through never resets a failed resolution's backoff (unlike scan/manual re-resolve). func (s *service) enqueue(ctx context.Context, artID model.ArtworkID, priority int) { - err := s.ds.ArtworkQueue(ctx).Enqueue(model.ArtworkQueueItem{ + err := s.ds.ArtworkQueue(ctx).EnqueueBump(model.ArtworkQueueItem{ ItemKind: artID.Kind.Prefix(), ItemID: artID.ID, ImageType: model.ImageTypePrimary, diff --git a/model/artwork.go b/model/artwork.go index 88b2d009d..6d343946e 100644 --- a/model/artwork.go +++ b/model/artwork.go @@ -91,8 +91,12 @@ type ArtworkRepository interface { } type ArtworkQueueRepository interface { - // Enqueue upserts; an existing row keeps the higher of the two priorities. + // Enqueue upserts; an existing row keeps the higher of the two priorities and has its + // retry_at reset (a detected change wants immediate re-resolution). Enqueue(items ...ArtworkQueueItem) error + // EnqueueBump upserts like Enqueue but preserves an existing row's retry_at, so a + // request-triggered read-through never resets a failed resolution's backoff. + EnqueueBump(items ...ArtworkQueueItem) error // DequeueBatch returns up to n items with retry_at <= now, priority desc, enqueued_at asc. DequeueBatch(n int) ([]ArtworkQueueItem, error) // MarkFailed increments attempts and pushes retry_at into the future. diff --git a/persistence/artwork_queue_repository.go b/persistence/artwork_queue_repository.go index 59e130204..81f674dd9 100644 --- a/persistence/artwork_queue_repository.go +++ b/persistence/artwork_queue_repository.go @@ -26,6 +26,18 @@ func NewArtworkQueueRepository(ctx context.Context, db dbx.Builder) model.Artwor } func (r *artworkQueueRepository) Enqueue(items ...model.ArtworkQueueItem) error { + return r.enqueue(`ON CONFLICT (item_kind, item_id, image_type) DO UPDATE SET + priority = MAX(priority, excluded.priority), retry_at = excluded.retry_at`, items) +} + +// EnqueueBump raises priority like Enqueue but leaves an existing row's retry_at intact, so a +// request-triggered read-through never resets a failed resolution's backoff. New rows insert eligible. +func (r *artworkQueueRepository) EnqueueBump(items ...model.ArtworkQueueItem) error { + return r.enqueue(`ON CONFLICT (item_kind, item_id, image_type) DO UPDATE SET + priority = MAX(priority, excluded.priority)`, items) +} + +func (r *artworkQueueRepository) enqueue(conflict string, items []model.ArtworkQueueItem) error { now := time.Now() for chunk := range slices.Chunk(items, enqueueChunkSize) { ins := Insert(r.tableName).Columns("item_kind", "item_id", "image_type", "priority", "attempts", "retry_at", "enqueued_at") @@ -35,8 +47,7 @@ func (r *artworkQueueRepository) Enqueue(items ...model.ArtworkQueueItem) error } ins = ins.Values(it.ItemKind, it.ItemID, it.ImageType, it.Priority, 0, now, now) } - ins = ins.Suffix(`ON CONFLICT (item_kind, item_id, image_type) DO UPDATE SET - priority = MAX(priority, excluded.priority), retry_at = excluded.retry_at`) + ins = ins.Suffix(conflict) if _, err := r.executeSQL(ins); err != nil { return err } diff --git a/persistence/artwork_queue_repository_test.go b/persistence/artwork_queue_repository_test.go index 9a4054cb9..f5eafec00 100644 --- a/persistence/artwork_queue_repository_test.go +++ b/persistence/artwork_queue_repository_test.go @@ -41,6 +41,30 @@ var _ = Describe("ArtworkQueueRepository", func() { Expect(got[0].Priority).To(Equal(model.ArtworkPriorityBump)) }) + It("EnqueueBump raises priority without resetting a backing-off row's retry_at", func() { + Expect(repo.Enqueue(item("al", "b1", model.ArtworkPriorityScan))).To(Succeed()) + // Push retry_at into the future so the row is backing off and hidden from dequeue. + Expect(repo.MarkFailed("al", "b1", model.ImageTypePrimary, time.Now().Add(time.Hour))).To(Succeed()) + Expect(repo.DequeueBatch(10)).To(BeEmpty()) + + // A request-triggered bump raises priority but must leave the backoff intact. + Expect(repo.EnqueueBump(item("al", "b1", model.ArtworkPriorityBump))).To(Succeed()) + Expect(repo.DequeueBatch(10)).To(BeEmpty(), "bump must not reset retry_at") + + // Enqueue (scan/manual), by contrast, resets retry_at and makes it eligible now. + Expect(repo.Enqueue(item("al", "b1", model.ArtworkPriorityScan))).To(Succeed()) + got, _ := repo.DequeueBatch(10) + Expect(got).To(HaveLen(1)) + Expect(got[0].Priority).To(Equal(model.ArtworkPriorityBump), "bump's higher priority is preserved") + }) + + It("EnqueueBump inserts a brand-new row eligible immediately", func() { + Expect(repo.EnqueueBump(item("ar", "n1", model.ArtworkPriorityBump))).To(Succeed()) + got, _ := repo.DequeueBatch(10) + Expect(got).To(HaveLen(1)) + Expect(got[0].ItemID).To(Equal("n1")) + }) + It("hides failed items until retry_at", func() { Expect(repo.Enqueue(item("al", "f1", model.ArtworkPriorityScan))).To(Succeed()) Expect(repo.MarkFailed("al", "f1", model.ImageTypePrimary, time.Now().Add(time.Hour))).To(Succeed()) diff --git a/tests/mock_artwork_queue_repo.go b/tests/mock_artwork_queue_repo.go index 500fdf4fd..cbb2f3f58 100644 --- a/tests/mock_artwork_queue_repo.go +++ b/tests/mock_artwork_queue_repo.go @@ -160,6 +160,32 @@ func (m *MockArtworkQueueRepo) Count() (int64, error) { return int64(len(m.Data)), nil } +func (m *MockArtworkQueueRepo) EnqueueBump(items ...model.ArtworkQueueItem) error { + m.mu.Lock() + defer m.mu.Unlock() + if m.Err != nil { + return m.Err + } + now := time.Now() + for _, it := range items { + if it.ImageType == "" { + it.ImageType = model.ImageTypePrimary + } + k := iaKey(it.ItemKind, it.ItemID, it.ImageType) + // Preserve an existing row's retry_at: a bump raises priority without resetting backoff. + if prev, ok := m.Data[k]; ok { + prev.Priority = max(prev.Priority, it.Priority) + m.Data[k] = prev + continue + } + it.Attempts = 0 + it.RetryAt = now + it.EnqueuedAt = now + m.Data[k] = it + } + return nil +} + func (m *MockArtworkQueueRepo) EnqueueStaleAbsent(kind string, attemptedBefore time.Time) (int64, error) { m.mu.Lock() defer m.mu.Unlock()