fix(artwork): purge dangling queue rows and guard concurrent re-enqueues

Queue rows for deleted entities failed forever (Get -> ErrNotFound -> failed -> capped retries, unbounded). Add ArtworkQueueRepository.PurgeDangling, called from Prune next to the item_artwork purge. Separately, the found/absent path unconditionally deleted the dequeued row, erasing a concurrent scan re-enqueue; switch to DeleteIfUnchanged, which deletes only while retry_at still matches the dequeued value (verified retry_at is the column an Enqueue upsert resets).

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)

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

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

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

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

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)

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)

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