diff --git a/model/artwork.go b/model/artwork.go index 1efa4c1e9..204fe5ebb 100644 --- a/model/artwork.go +++ b/model/artwork.go @@ -129,12 +129,9 @@ type ArtworkQueueRepository interface { // Restricted to the given item kinds when any are passed, so a drain pool sees only its own // work and cannot be held up behind another kind's backlog. DequeueBatch(n int, kinds ...string) ([]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. DeleteIfUnchanged(kind, id, imageType string, retryAt time.Time) error diff --git a/persistence/artwork_queue_repository.go b/persistence/artwork_queue_repository.go index 69af9c66a..e3e1f735b 100644 --- a/persistence/artwork_queue_repository.go +++ b/persistence/artwork_queue_repository.go @@ -72,18 +72,6 @@ func (r *artworkQueueRepository) DequeueBatch(n int, kinds ...string) ([]model.A return res, err } -func (r *artworkQueueRepository) MarkFailed(kind, id, imageType string, 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}) - c, err := r.executeSQL(upd) - if err == nil && c == 0 { - return model.ErrNotFound - } - 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 { @@ -95,10 +83,6 @@ func (r *artworkQueueRepository) MarkFailedIfUnchanged(kind, id, imageType strin return err } -func (r *artworkQueueRepository) Delete(kind, id, imageType string) error { - return r.delete(Eq{"item_kind": kind, "item_id": id, "image_type": imageType}) -} - // DeleteIfUnchanged deletes the row only while its retry_at still equals the dequeued // value; a concurrent Enqueue resets retry_at, so the row survives to be re-resolved. func (r *artworkQueueRepository) DeleteIfUnchanged(kind, id, imageType string, retryAt time.Time) error { diff --git a/persistence/artwork_queue_repository_test.go b/persistence/artwork_queue_repository_test.go index 402b4274a..a2843bf14 100644 --- a/persistence/artwork_queue_repository_test.go +++ b/persistence/artwork_queue_repository_test.go @@ -4,6 +4,7 @@ import ( "context" "time" + "github.com/Masterminds/squirrel" "github.com/navidrome/navidrome/model" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -18,6 +19,24 @@ var _ = Describe("ArtworkQueueRepository", func() { ImageType: model.ImageTypePrimary, Priority: prio} } + // backOff puts a row into the state a failed resolution leaves behind. Production only ever + // reaches that state through MarkFailedIfUnchanged, which needs the retry_at it dequeued. + backOff := func(kind, id string, retryAt time.Time) { + GinkgoHelper() + r := repo.(*artworkQueueRepository) + _, err := r.executeSQL(squirrel.Update(r.tableName). + Set("attempts", squirrel.Expr("attempts + 1")). + Set("retry_at", retryAt). + Where(squirrel.Eq{"item_kind": kind, "item_id": id, "image_type": model.ImageTypePrimary})) + Expect(err).ToNot(HaveOccurred()) + } + + remove := func(kind, id string) { + GinkgoHelper() + r := repo.(*artworkQueueRepository) + Expect(r.delete(squirrel.Eq{"item_kind": kind, "item_id": id, "image_type": model.ImageTypePrimary})).To(Succeed()) + } + BeforeEach(func() { clearArtworkTables() DeferCleanup(clearArtworkTables) @@ -45,7 +64,7 @@ var _ = Describe("ArtworkQueueRepository", func() { 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()) + backOff("al", "b1", time.Now().Add(time.Hour)) Expect(repo.DequeueBatch(10)).To(BeEmpty()) // A request-triggered bump raises priority but must leave the backoff intact. @@ -68,13 +87,13 @@ var _ = Describe("ArtworkQueueRepository", func() { 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()) + backOff("al", "f1", time.Now().Add(time.Hour)) got, err := repo.DequeueBatch(10) Expect(err).ToNot(HaveOccurred()) Expect(got).To(BeEmpty()) - Expect(repo.MarkFailed("al", "f1", model.ImageTypePrimary, time.Now().Add(-time.Minute))).To(Succeed()) + backOff("al", "f1", time.Now().Add(-time.Minute)) got, _ = repo.DequeueBatch(10) Expect(got).To(HaveLen(1)) Expect(got[0].Attempts).To(Equal(2)) @@ -83,7 +102,7 @@ var _ = Describe("ArtworkQueueRepository", func() { 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()) + backOff("al", "m1", time.Now().Add(-time.Hour)) got, err := repo.DequeueBatch(10) Expect(err).ToNot(HaveOccurred()) Expect(got).To(HaveLen(1)) @@ -110,7 +129,7 @@ var _ = Describe("ArtworkQueueRepository", func() { It("Enqueue restarts the retry budget an existing row had spent", func() { Expect(repo.Enqueue(item("al", "e1", model.ArtworkPriorityScan))).To(Succeed()) - Expect(repo.MarkFailed("al", "e1", model.ImageTypePrimary, time.Now().Add(-time.Hour))).To(Succeed()) + backOff("al", "e1", time.Now().Add(-time.Hour)) stale := time.Now().Add(-48 * time.Hour) _, err := GetDBXBuilder().NewQuery("UPDATE artwork_queue SET enqueued_at = {:t} WHERE item_id = 'e1'"). Bind(dbx.Params{"t": stale}).Execute() @@ -130,7 +149,7 @@ var _ = Describe("ArtworkQueueRepository", func() { Expect(repo.Enqueue(item("al", "c1", 0))).To(Succeed()) n, _ := repo.Count() Expect(n).To(Equal(int64(1))) - Expect(repo.Delete("al", "c1", model.ImageTypePrimary)).To(Succeed()) + remove("al", "c1") n, _ = repo.Count() Expect(n).To(BeZero()) }) @@ -138,7 +157,7 @@ var _ = Describe("ArtworkQueueRepository", func() { It("DeleteIfUnchanged deletes only while retry_at is unchanged", func() { Expect(repo.Enqueue(item("al", "d1", model.ArtworkPriorityScan))).To(Succeed()) // Anchor retry_at in the past so it can never collide with the re-enqueue's now. - Expect(repo.MarkFailed("al", "d1", model.ImageTypePrimary, time.Now().Add(-time.Hour))).To(Succeed()) + backOff("al", "d1", time.Now().Add(-time.Hour)) got, err := repo.DequeueBatch(10) Expect(err).ToNot(HaveOccurred()) Expect(got).To(HaveLen(1)) diff --git a/tests/mock_artwork_queue_repo.go b/tests/mock_artwork_queue_repo.go index b85515dab..0068d6087 100644 --- a/tests/mock_artwork_queue_repo.go +++ b/tests/mock_artwork_queue_repo.go @@ -79,23 +79,6 @@ func (m *MockArtworkQueueRepo) DequeueBatch(n int, kinds ...string) ([]model.Art return res, nil } -func (m *MockArtworkQueueRepo) MarkFailed(kind, id, imageType string, retryAt time.Time) error { - m.mu.Lock() - defer m.mu.Unlock() - if m.Err != nil { - return m.Err - } - k := iaKey(kind, id, imageType) - it, ok := m.Data[k] - if !ok { - return model.ErrNotFound - } - it.Attempts++ - it.RetryAt = retryAt - m.Data[k] = it - return nil -} - func (m *MockArtworkQueueRepo) MarkFailedIfUnchanged(kind, id, imageType string, seenRetryAt, retryAt time.Time) error { m.mu.Lock() defer m.mu.Unlock() @@ -111,16 +94,6 @@ func (m *MockArtworkQueueRepo) MarkFailedIfUnchanged(kind, id, imageType string, return nil } -func (m *MockArtworkQueueRepo) Delete(kind, id, imageType string) error { - m.mu.Lock() - defer m.mu.Unlock() - if m.Err != nil { - return m.Err - } - delete(m.Data, iaKey(kind, id, imageType)) - return nil -} - func (m *MockArtworkQueueRepo) DeleteIfUnchanged(kind, id, imageType string, retryAt time.Time) error { m.mu.Lock() defer m.mu.Unlock()