diff --git a/core/artwork/artwork.go b/core/artwork/artwork.go index 218c4d665..8ac8d9c9d 100644 --- a/core/artwork/artwork.go +++ b/core/artwork/artwork.go @@ -129,8 +129,7 @@ func (s *service) serveEntity(ctx context.Context, artID model.ArtworkID, size i case err != nil: return nil, err case ia.Hash == "": - // EnqueueBump preserves an existing backoff row's retry_at, and inserts an - // immediately-eligible recheck for a settled absent row. + // Inserts an immediately-eligible recheck for a settled absent row. if time.Since(ia.AttemptedAt) > requestRecheckAge { s.enqueue(ctx, artID, model.ArtworkPriorityBump) } @@ -344,9 +343,8 @@ func (s *service) dangling(ctx context.Context, artID model.ArtworkID) (*Image, return nil, ErrUnavailable } -// enqueue uses EnqueueBump so an incidental read-through never resets a failed resolution's backoff. func (s *service) enqueue(ctx context.Context, artID model.ArtworkID, priority int) { - err := s.ds.ArtworkQueue(ctx).EnqueueBump(model.ArtworkQueueItem{ + err := s.ds.ArtworkQueue(ctx).EnqueuePreservingBackoff(model.ArtworkQueueItem{ ItemKind: artID.Kind.Prefix(), ItemID: artID.ID, ImageType: model.ImageTypePrimary, diff --git a/core/artwork/e2e/acquire_serve_test.go b/core/artwork/e2e/acquire_serve_test.go index ec6ba342e..4fea9ff6c 100644 --- a/core/artwork/e2e/acquire_serve_test.go +++ b/core/artwork/e2e/acquire_serve_test.go @@ -55,7 +55,7 @@ var _ = Describe("Acquisition → serve loop", func() { // Enqueues the way the serving paths do, so the drain is driven by a plain queue row. bump := func(kind, id string) { GinkgoHelper() - Expect(ds.ArtworkQueue(ctx).EnqueueBump(model.ArtworkQueueItem{ + Expect(ds.ArtworkQueue(ctx).EnqueuePreservingBackoff(model.ArtworkQueueItem{ ItemKind: kind, ItemID: id, ImageType: model.ImageTypePrimary, Priority: model.ArtworkPriorityBump, })).To(Succeed()) diff --git a/core/artwork/e2e/resolution_harness_test.go b/core/artwork/e2e/resolution_harness_test.go index ac52e5b50..1168393df 100644 --- a/core/artwork/e2e/resolution_harness_test.go +++ b/core/artwork/e2e/resolution_harness_test.go @@ -140,7 +140,7 @@ func scan() { func acquire(kind model.Kind, id string) model.ItemArtwork { GinkgoHelper() // Enqueues the way the serving paths do, so the drain is driven by a plain queue row. - Expect(rds.ArtworkQueue(rctx).EnqueueBump(model.ArtworkQueueItem{ + Expect(rds.ArtworkQueue(rctx).EnqueuePreservingBackoff(model.ArtworkQueueItem{ ItemKind: kind.Prefix(), ItemID: id, ImageType: model.ImageTypePrimary, Priority: model.ArtworkPriorityBump, })).To(Succeed()) diff --git a/model/artwork.go b/model/artwork.go index 4f3d4e6e8..458888d23 100644 --- a/model/artwork.go +++ b/model/artwork.go @@ -113,9 +113,9 @@ type ArtworkRepository interface { type ArtworkQueueRepository interface { // Enqueue upserts; an existing row keeps the higher priority and has its retry_at reset. Enqueue(items ...ArtworkQueueItem) error - // EnqueueBump upserts like Enqueue but preserves an existing row's retry_at, so a + // EnqueuePreservingBackoff 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 + EnqueuePreservingBackoff(items ...ArtworkQueueItem) error // EnqueueStaleAbsent inserts queue rows (priority Recheck) for absent states older than cutoff. EnqueueStaleAbsent(kind Kind, attemptedBefore time.Time) (int64, error) // EnqueueAllMissing inserts queue rows for all entities with no item_artwork row, at the given priority. diff --git a/persistence/artwork_queue_repository.go b/persistence/artwork_queue_repository.go index 69cb17d2f..7091a8817 100644 --- a/persistence/artwork_queue_repository.go +++ b/persistence/artwork_queue_repository.go @@ -16,6 +16,9 @@ import ( // Keeps each multi-row insert under SQLite's bind-variable limit (at most 7 vars per row). const enqueueChunkSize = 100 +// Every insert writes these, in this order; the INSERT..SELECT forms must project them to match. +var enqueueColumns = []string{"item_kind", "item_id", "image_type", "priority", "attempts", "retry_at", "enqueued_at"} + type artworkQueueRepository struct { sqlRepository } @@ -35,20 +38,16 @@ func (r *artworkQueueRepository) Enqueue(items ...model.ArtworkQueueItem) error attempts = 0, enqueued_at = excluded.enqueued_at`, items) } -func (r *artworkQueueRepository) EnqueueBump(items ...model.ArtworkQueueItem) error { +func (r *artworkQueueRepository) EnqueuePreservingBackoff(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) EnqueueStaleAbsent(kind model.Kind, attemptedBefore time.Time) (int64, error) { now := time.Now() - // DO NOTHING is deliberate: rechecks must not bump priority/retry_at of already-queued items. - ins := Expr(`INSERT INTO `+r.tableName+` (item_kind, item_id, image_type, priority, attempts, retry_at, enqueued_at) - SELECT item_kind, item_id, image_type, ?, 0, ?, ? - FROM `+itemArtworkTable+` WHERE item_kind = ? AND hash = '' AND attempted_at < ? - ON CONFLICT (item_kind, item_id, image_type) DO NOTHING`, + return r.insertIfNotQueued("", `SELECT item_kind, item_id, image_type, ?, 0, ?, ? + FROM `+itemArtworkTable+` WHERE item_kind = ? AND hash = '' AND attempted_at < ?`, model.ArtworkPriorityRecheck, now, now, kind.Prefix(), attemptedBefore) - return r.executeSQL(ins) } func (r *artworkQueueRepository) EnqueueAllMissing(kind model.Kind, priority int) (int64, error) { @@ -57,14 +56,10 @@ func (r *artworkQueueRepository) EnqueueAllMissing(kind model.Kind, priority int return 0, fmt.Errorf("artwork queue: no entity table for kind %q", kind.Prefix()) } now := time.Now() - // DO NOTHING is deliberate: rechecks must not bump priority/retry_at of already-queued items. - ins := Expr(`INSERT INTO `+r.tableName+` (item_kind, item_id, image_type, priority, attempts, retry_at, enqueued_at) - SELECT ?, id, ?, ?, 0, ?, ? + return r.insertIfNotQueued("", `SELECT ?, id, ?, ?, 0, ?, ? FROM `+entityTable+` - WHERE id NOT IN (SELECT item_id FROM `+itemArtworkTable+` WHERE item_kind = ?) - ON CONFLICT (item_kind, item_id, image_type) DO NOTHING`, + WHERE id NOT IN (SELECT item_id FROM `+itemArtworkTable+` WHERE item_kind = ?)`, kind.Prefix(), model.ImageTypePrimary, priority, now, now, kind.Prefix()) - return r.executeSQL(ins) } func (r *artworkQueueRepository) EnqueueIfMissing(items ...model.ArtworkQueueItem) error { @@ -77,26 +72,33 @@ func (r *artworkQueueRepository) EnqueueIfMissing(items ...model.ArtworkQueueIte args = append(args, it.ItemKind, it.ItemID, cmp.Or(it.ImageType, model.ImageTypePrimary), it.Priority) } args = append(args, now, now) - ins := Expr(`WITH new_items(item_kind, item_id, image_type, priority) AS (VALUES `+ - strings.Join(rows, ",")+`) - INSERT INTO `+r.tableName+` (item_kind, item_id, image_type, priority, attempts, retry_at, enqueued_at) - SELECT n.item_kind, n.item_id, n.image_type, n.priority, 0, ?, ? + _, err := r.insertIfNotQueued( + `WITH new_items(item_kind, item_id, image_type, priority) AS (VALUES `+strings.Join(rows, ",")+`) `, + `SELECT n.item_kind, n.item_id, n.image_type, n.priority, 0, ?, ? FROM new_items n WHERE NOT EXISTS ( SELECT 1 FROM `+itemArtworkTable+` ia - WHERE ia.item_kind = n.item_kind AND ia.item_id = n.item_id AND ia.image_type = n.image_type) - ON CONFLICT (item_kind, item_id, image_type) DO NOTHING`, args...) - if _, err := r.executeSQL(ins); err != nil { + WHERE ia.item_kind = n.item_kind AND ia.item_id = n.item_id AND ia.image_type = n.image_type)`, + args...) + if err != nil { return err } } return nil } +// insertIfNotQueued inserts the rows selected by the given SQL, optionally prefixed by a CTE. DO NOTHING is +// deliberate: a recheck must not bump the priority or retry_at of an already-queued item. +func (r *artworkQueueRepository) insertIfNotQueued(with, sql string, args ...any) (int64, error) { + return r.executeSQL(Expr(with+`INSERT INTO `+r.tableName+ + ` (`+strings.Join(enqueueColumns, ", ")+`) `+sql+ + ` ON CONFLICT (item_kind, item_id, image_type) DO NOTHING`, args...)) +} + 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") + ins := Insert(r.tableName).Columns(enqueueColumns...) for _, it := range chunk { ins = ins.Values(it.ItemKind, it.ItemID, cmp.Or(it.ImageType, model.ImageTypePrimary), it.Priority, 0, now, now) } diff --git a/persistence/artwork_queue_repository_test.go b/persistence/artwork_queue_repository_test.go index bae5cdb16..6638a204e 100644 --- a/persistence/artwork_queue_repository_test.go +++ b/persistence/artwork_queue_repository_test.go @@ -62,12 +62,12 @@ 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() { + It("EnqueuePreservingBackoff raises priority without resetting a backing-off row's retry_at", func() { Expect(repo.Enqueue(item("al", "b1", model.ArtworkPriorityScan))).To(Succeed()) backOff("al", "b1", time.Now().Add(time.Hour)) Expect(repo.DequeueBatch(10)).To(BeEmpty()) - Expect(repo.EnqueueBump(item("al", "b1", model.ArtworkPriorityBump))).To(Succeed()) + Expect(repo.EnqueuePreservingBackoff(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. @@ -77,8 +77,8 @@ var _ = Describe("ArtworkQueueRepository", func() { 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()) + It("EnqueuePreservingBackoff inserts a brand-new row eligible immediately", func() { + Expect(repo.EnqueuePreservingBackoff(item("ar", "n1", model.ArtworkPriorityBump))).To(Succeed()) got, _ := repo.DequeueBatch(10) Expect(got).To(HaveLen(1)) Expect(got[0].ItemID).To(Equal("n1")) @@ -238,6 +238,33 @@ var _ = Describe("ArtworkQueueRepository", func() { Expect(ids).ToNot(ContainElement(albumAbbeyRoad.ID), "an absent-state album must not be enqueued as missing") }) + It("EnqueueIfMissing skips items that already have an item_artwork row", func() { + awRepo := NewArtworkRepository(context.Background(), GetDBXBuilder()) + Expect(awRepo.PutItemArtwork(&model.ItemArtwork{ItemKind: "al", ItemID: "resolved", ImageType: model.ImageTypePrimary, Hash: "hX", AttemptedAt: time.Now()})).To(Succeed()) + Expect(awRepo.PutItemArtwork(&model.ItemArtwork{ItemKind: "al", ItemID: "absent", ImageType: model.ImageTypePrimary, Hash: "", AttemptedAt: time.Now()})).To(Succeed()) + + Expect(repo.EnqueueIfMissing( + item("al", "resolved", model.ArtworkPriorityScan), + item("al", "absent", model.ArtworkPriorityScan), + item("al", "brandnew", model.ArtworkPriorityScan), + )).To(Succeed()) + + got, err := repo.DequeueBatch(100) + Expect(err).ToNot(HaveOccurred()) + ids := slice.Map(got, func(it model.ArtworkQueueItem) string { return it.ItemID }) + Expect(ids).To(ConsistOf("brandnew"), "only an item with no state row may be enqueued") + }) + + It("EnqueueIfMissing leaves an already-queued row untouched", func() { + Expect(repo.Enqueue(item("al", "queued", model.ArtworkPriorityBump))).To(Succeed()) + + Expect(repo.EnqueueIfMissing(item("al", "queued", model.ArtworkPriorityScan))).To(Succeed()) + + got, _ := repo.DequeueBatch(100) + Expect(got).To(HaveLen(1)) + Expect(got[0].Priority).To(Equal(model.ArtworkPriorityBump), "the existing priority must survive") + }) + It("does not disturb an already-queued entity when enqueueing missing rows", func() { Expect(repo.Enqueue(item("al", albumRadioactivity.ID, model.ArtworkPriorityBump))).To(Succeed()) diff --git a/server/subsonic/e2e/subsonic_artwork_test.go b/server/subsonic/e2e/subsonic_artwork_test.go index c1976983b..619619810 100644 --- a/server/subsonic/e2e/subsonic_artwork_test.go +++ b/server/subsonic/e2e/subsonic_artwork_test.go @@ -148,7 +148,7 @@ var _ = Describe("Artwork Serving", Ordered, func() { It("drains the queue: folder art is acquired, the artless album settles absent", func() { // Enqueues the way the serving paths do, so the drain is driven by a plain queue row. for _, id := range []string{artfulID, artlessID} { - Expect(ds.ArtworkQueue(ctx).EnqueueBump(model.ArtworkQueueItem{ + Expect(ds.ArtworkQueue(ctx).EnqueuePreservingBackoff(model.ArtworkQueueItem{ ItemKind: model.KindAlbumArtwork.Prefix(), ItemID: id, ImageType: model.ImageTypePrimary, Priority: model.ArtworkPriorityBump, })).To(Succeed()) diff --git a/tests/mock_artwork_queue_repo.go b/tests/mock_artwork_queue_repo.go index a6e5173a6..1b097ca32 100644 --- a/tests/mock_artwork_queue_repo.go +++ b/tests/mock_artwork_queue_repo.go @@ -161,7 +161,7 @@ func (m *MockArtworkQueueRepo) Count() (int64, error) { return int64(len(m.Data)), nil } -func (m *MockArtworkQueueRepo) EnqueueBump(items ...model.ArtworkQueueItem) error { +func (m *MockArtworkQueueRepo) EnqueuePreservingBackoff(items ...model.ArtworkQueueItem) error { m.mu.Lock() defer m.mu.Unlock() if m.Err != nil {