nik/navidrome
1
0
Fork 0
mirror of https://github.com/navidrome/navidrome.git synced 2026-08-01 07:21:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(persistence): name the backoff-preserving enqueue for what it does

Browse Source 
EnqueueBump was named for a priority, but its behaviour is preserving an
existing row's retry_at; the priority comes from the item. Its one caller
passes ArtworkPriorityScan, which read like a bug at the call site and is not.

The three DO NOTHING inserts each repeated the INSERT prefix, the column list
and the conflict clause. They now share insertIfNotQueued, with the CTE that
EnqueueIfMissing needs passed as a prefix, and the column list lives in one
slice used by both squirrel and the raw SQL.

EnqueueIfMissing had no test against real SQL, only a mock mirroring the
anti-join by hand, so the rewritten statement had nothing verifying it. Two
specs now cover it: items with a state row are skipped, and an already-queued
row keeps its priority.

Signed-off-by: Deluan <deluan@navidrome.org>
This commit is contained in:
Deluan 2026-07-30 12:16:17 -04:00
parent 3ce92cc055
commit fb3efe8a61
8 changed files with 62 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
core/artwork/artwork.go
View File

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

2
core/artwork/e2e/acquire_serve_test.go
View File

@ -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())

2
core/artwork/e2e/resolution_harness_test.go
View File

@ -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())

4
model/artwork.go
View File

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

44
persistence/artwork_queue_repository.go
View File

@ -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)
}

35
persistence/artwork_queue_repository_test.go
View File

@ -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())

2
server/subsonic/e2e/subsonic_artwork_test.go
View File

@ -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())

2
tests/mock_artwork_queue_repo.go
View File

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