diff --git a/core/artwork/prune.go b/core/artwork/prune.go index 5278beaa4..35eddb525 100644 --- a/core/artwork/prune.go +++ b/core/artwork/prune.go @@ -22,6 +22,15 @@ func Prune(ctx context.Context, ds model.DataStore, store *ImageStore) error { log.Info(ctx, "Prune: purged dangling item artwork state", "count", purged) } + // Queue rows for deleted entities would otherwise retry forever (Get -> not found -> failed). + queuePurged, err := ds.ArtworkQueue(ctx).PurgeDangling() + if err != nil { + return err + } + if queuePurged > 0 { + log.Info(ctx, "Prune: purged dangling artwork queue rows", "count", queuePurged) + } + // One grace cutoff for both the DB orphan check and the file sweep: files younger // than the window may belong to acquisitions whose rows aren't committed yet. cutoff := time.Now().Add(-pruneMinAge) diff --git a/core/artwork/prune_test.go b/core/artwork/prune_test.go index 3de46942a..5511b2305 100644 --- a/core/artwork/prune_test.go +++ b/core/artwork/prune_test.go @@ -59,6 +59,21 @@ var _ = Describe("Prune", func() { Expect(err).ToNot(HaveOccurred()) }) + It("purges dangling artwork_queue rows for gone entities", func() { + queueRepo := tests.CreateMockArtworkQueueRepo() + Expect(queueRepo.Enqueue( + model.ArtworkQueueItem{ItemKind: "al", ItemID: "gone-album", ImageType: model.ImageTypePrimary}, + model.ArtworkQueueItem{ItemKind: "al", ItemID: "live-album", ImageType: model.ImageTypePrimary}, + )).To(Succeed()) + queueRepo.ExistingIDs = map[string]map[string]bool{"al": {"live-album": true}} + ds.MockedArtworkQueue = queueRepo + + Expect(Prune(context.Background(), ds, store)).To(Succeed()) + + Expect(findQueued(queueRepo, "al", "gone-album")).To(BeNil()) + Expect(findQueued(queueRepo, "al", "live-album")).ToNot(BeNil()) + }) + It("deletes orphan rows and their store files, keeps referenced ones", func() { data := []byte("orphan-bytes") h, _ := HashImage(bytes.NewReader(data)) diff --git a/core/artwork/worker.go b/core/artwork/worker.go index 1ebe15527..fb20a2bbc 100644 --- a/core/artwork/worker.go +++ b/core/artwork/worker.go @@ -27,9 +27,8 @@ const ( var errBreakerOpen = errors.New("artwork: external circuit breaker open") -// Worker drains the artwork queue and runs each item through processItem. The -// external step is rate-limited and circuit-broken; prune is serialized against -// in-flight acquisitions via pruneMu (acquisitions RLock, prune Lock). +// Worker drains the artwork queue through processItem: the external step is rate-limited +// and circuit-broken, and prune is serialized against in-flight acquisitions via pruneMu. type Worker struct { deps workerDeps limiter *rate.Limiter @@ -150,7 +149,9 @@ func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) { queue := w.deps.ds.ArtworkQueue(ctx) switch out { case outcomeFound, outcomeAbsent: - if err := queue.Delete(item.ItemKind, item.ItemID, item.ImageType); err != nil { + // DeleteIfUnchanged, not Delete: a scan that re-enqueued this row mid-flight reset + // its retry_at, so the row survives here and the next drain re-resolves it. + if err := queue.DeleteIfUnchanged(item.ItemKind, item.ItemID, item.ImageType, item.RetryAt); err != nil { log.Warn(ctx, "artwork: could not delete processed queue item", "kind", item.ItemKind, "id", item.ItemID, err) } case outcomeFailed: @@ -161,8 +162,8 @@ func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) { } } -// claim reserves items not already in flight, so a wake-triggered re-drain never -// double-processes an item still running from a previous cycle. +// claim reserves items not already in flight, so a row appearing twice within a single +// batch is processed once. func (w *Worker) claim(batch []model.ArtworkQueueItem) []model.ArtworkQueueItem { w.mu.Lock() defer w.mu.Unlock() diff --git a/core/artwork/worker_test.go b/core/artwork/worker_test.go index d41249fa2..058d707b8 100644 --- a/core/artwork/worker_test.go +++ b/core/artwork/worker_test.go @@ -16,6 +16,27 @@ import ( . "github.com/onsi/gomega" ) +// reenqueueOnDequeue simulates a concurrent scan Enqueue between DequeueBatch and the +// worker's delete by bumping retry_at, so a DeleteIfUnchanged on the dequeued value no-ops. +type reenqueueOnDequeue struct { + *tests.MockArtworkQueueRepo + done bool +} + +func (r *reenqueueOnDequeue) DequeueBatch(n int) ([]model.ArtworkQueueItem, error) { + items, err := r.MockArtworkQueueRepo.DequeueBatch(n) + if !r.done && len(items) > 0 { + r.done = true + for k, it := range r.Data { + if it.ItemKind == items[0].ItemKind && it.ItemID == items[0].ItemID { + it.RetryAt = items[0].RetryAt.Add(time.Minute) + r.Data[k] = it + } + } + } + return items, err +} + func findQueued(q *tests.MockArtworkQueueRepo, kind, id string) *model.ArtworkQueueItem { for _, it := range q.Data { if it.ItemKind == kind && it.ItemID == id { @@ -114,6 +135,32 @@ var _ = Describe("Worker", func() { Expect(err).To(MatchError(model.ErrNotFound), "a timeout must never settle on absent") }) + It("keeps a row re-enqueued between dequeue and delete", func() { + folderRepo.result = []model.Folder{{ + Path: "tests/fixtures/artist/an-album", + ImageFiles: []string{"cover.jpg"}, + }} + ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{ + {ID: "al7", Name: "Album", FolderIDs: []string{"f1"}}, + }) + racing := &reenqueueOnDequeue{MockArtworkQueueRepo: queueRepo} + ds.MockedArtworkQueue = racing + w = NewWorker(ds, store, prov, ffm) + Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ + ItemKind: "al", ItemID: "al7", Priority: model.ArtworkPriorityScan, + })).To(Succeed()) + + n, err := w.drain(ctx, 1) + Expect(err).ToNot(HaveOccurred()) + Expect(n).To(Equal(1)) + + // The concurrent re-enqueue changed retry_at, so the found-path delete was a no-op. + Expect(findQueued(queueRepo, "al", "al7")).ToNot(BeNil()) + ia, err := artRepo.GetItemArtwork("al", "al7", model.ImageTypePrimary) + Expect(err).ToNot(HaveOccurred()) + Expect(ia.Source).To(Equal("folder")) + }) + 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 73db00ca5..0aec6489a 100644 --- a/model/artwork.go +++ b/model/artwork.go @@ -89,7 +89,12 @@ type ArtworkQueueRepository interface { // MarkFailed increments attempts and pushes retry_at into the future. MarkFailed(kind, id, imageType string, 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 Count() (int64, error) // EnqueueStaleAbsent inserts queue rows (priority Recheck) for absent states older than cutoff. EnqueueStaleAbsent(kind string, attemptedBefore time.Time) (int64, error) + // PurgeDangling removes queue rows whose entity no longer exists. + PurgeDangling() (int64, error) } diff --git a/persistence/artwork_queue_repository.go b/persistence/artwork_queue_repository.go index ced5e9913..eb5f21ecd 100644 --- a/persistence/artwork_queue_repository.go +++ b/persistence/artwork_queue_repository.go @@ -70,6 +70,29 @@ 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 { + return r.delete(Eq{"item_kind": kind, "item_id": id, "image_type": imageType, "retry_at": retryAt}) +} + +// PurgeDangling removes queue rows whose entity no longer exists, per kind. +func (r *artworkQueueRepository) PurgeDangling() (int64, error) { + var total int64 + for kind, table := range danglingItemArtworkKinds { + del := Delete(r.tableName).Where(And{ + Eq{"item_kind": kind}, + Expr("item_id NOT IN (SELECT id FROM " + table + ")"), + }) + c, err := r.executeSQL(del) + if err != nil { + return total, err + } + total += c + } + return total, nil +} + func (r *artworkQueueRepository) Count() (int64, error) { var res struct{ Count int64 } err := r.queryOne(Select("count(*) as count").From(r.tableName), &res) diff --git a/persistence/artwork_queue_repository_test.go b/persistence/artwork_queue_repository_test.go index 2b6f0d8c5..e96802284 100644 --- a/persistence/artwork_queue_repository_test.go +++ b/persistence/artwork_queue_repository_test.go @@ -63,6 +63,55 @@ var _ = Describe("ArtworkQueueRepository", func() { Expect(n).To(BeZero()) }) + 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()) + 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", "d1", model.ArtworkPriorityScan))).To(Succeed()) + + // Deleting with the stale retry_at is a no-op: the re-enqueued row survives. + Expect(repo.DeleteIfUnchanged("al", "d1", model.ImageTypePrimary, original)).To(Succeed()) + n, _ := repo.Count() + Expect(n).To(Equal(int64(1))) + + // Deleting with the current retry_at removes it. + got, _ = repo.DequeueBatch(10) + Expect(got).To(HaveLen(1)) + Expect(repo.DeleteIfUnchanged("al", "d1", model.ImageTypePrimary, got[0].RetryAt)).To(Succeed()) + n, _ = repo.Count() + Expect(n).To(BeZero()) + }) + + It("purges queue rows whose entity no longer exists, per kind", func() { + Expect(repo.Enqueue( + item("al", albumSgtPeppers.ID, model.ArtworkPriorityScan), + item("al", "no-such-album", model.ArtworkPriorityScan), + item("ar", artistKraftwerk.ID, model.ArtworkPriorityScan), + item("ar", "no-such-artist", model.ArtworkPriorityScan), + item("pl", plsBest.ID, model.ArtworkPriorityScan), + item("pl", "no-such-playlist", model.ArtworkPriorityScan), + item("ra", radioWithHomePage.ID, model.ArtworkPriorityScan), + item("ra", "no-such-radio", model.ArtworkPriorityScan), + )).To(Succeed()) + + purged, err := repo.PurgeDangling() + Expect(err).ToNot(HaveOccurred()) + Expect(purged).To(Equal(int64(4))) + + got, _ := repo.DequeueBatch(100) + ids := make([]string, 0, len(got)) + for _, it := range got { + ids = append(ids, it.ItemID) + } + Expect(ids).To(ConsistOf(albumSgtPeppers.ID, artistKraftwerk.ID, plsBest.ID, radioWithHomePage.ID)) + }) + It("enqueues stale absent states for recheck", func() { awRepo := NewArtworkRepository(context.Background(), GetDBXBuilder()) old := time.Now().Add(-48 * time.Hour) diff --git a/tests/mock_artwork_queue_repo.go b/tests/mock_artwork_queue_repo.go index f62afe43a..8d01ef9c0 100644 --- a/tests/mock_artwork_queue_repo.go +++ b/tests/mock_artwork_queue_repo.go @@ -13,6 +13,8 @@ type MockArtworkQueueRepo struct { Err error // ItemArtworkSource, when set, backs EnqueueStaleAbsent with real item_artwork state. ItemArtworkSource *MockArtworkRepo + // ExistingIDs, keyed by item_kind, backs PurgeDangling; a nil per-kind map keeps that kind. + ExistingIDs map[string]map[string]bool } func CreateMockArtworkQueueRepo() *MockArtworkQueueRepo { @@ -90,6 +92,35 @@ func (m *MockArtworkQueueRepo) Delete(kind, id, imageType string) error { return nil } +func (m *MockArtworkQueueRepo) DeleteIfUnchanged(kind, id, imageType string, 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(retryAt) { + delete(m.Data, k) + } + return nil +} + +func (m *MockArtworkQueueRepo) PurgeDangling() (int64, error) { + if m.Err != nil { + return 0, m.Err + } + var purged int64 + for k, it := range m.Data { + existing := m.ExistingIDs[it.ItemKind] + if existing == nil { + continue + } + if !existing[it.ItemID] { + delete(m.Data, k) + purged++ + } + } + return purged, nil +} + func (m *MockArtworkQueueRepo) Count() (int64, error) { if m.Err != nil { return 0, m.Err