perf(artwork): drain local and external artwork in separate pools

A first backfill enqueues artists before albums at a single priority, so
the drain took them in that order. Artists resolve through a
rate-limited agent, and gate() waits for its permit while holding a
worker slot, so the whole pool sat asleep in the limiter with every
album queued behind it.

Measured on a 96k-track library (29,115 artists to 6,949 albums, 4:1):
zero albums resolved in seven minutes, and roughly 3.3 hours before the
first album cover would have appeared. Splitting the drain gives each
class its own slots: albums now finish in under eight minutes while
artists trickle at the same 2/s they were always limited to.

The two budgets are carved out of MaxOpenConns so a second pool cannot
take connections the scanner and the UI need. Dequeue filters by kind,
and the drain index leads with item_kind so each pool seeks to its own
work instead of scanning past the other's backlog.

core/artwork/agent_images_test.go

@@ -28,11 +28,16 @@ type fakeImageAgent struct {
 	albumCalls    int
 	gotArtistName string
 	gotAlbumName  string
+	// block, when set, holds every lookup until closed, standing in for a slow/rate-limited agent.
+	block chan struct{}
 }
 
 func (f *fakeImageAgent) AgentName() string { return f.name }
 
 func (f *fakeImageAgent) GetArtistImages(_ context.Context, _, name, _ string) ([]agents.ExternalImage, error) {
+	if f.block != nil {
+		<-f.block
+	}
 	f.artistCalls++
 	f.gotArtistName = name
 	return f.imgs, f.err

core/artwork/worker.go

@@ -40,6 +40,15 @@ type extGate struct {
 	breaker *breaker
 }
 
+// drainPool drains one class of work with its own slot budget, so a kind whose resolution
+// blocks cannot occupy slots another kind needs.
+type drainPool struct {
+	name        string
+	kinds       []string
+	concurrency int
+	wake        chan struct{}
+}
+
 // Worker drains the artwork queue through processItem: each external agent is rate-limited
 // and circuit-broken independently, and prune is serialized against in-flight acquisitions
 // via pruneMu.
@@ -47,7 +56,7 @@ type Worker struct {
 	deps    workerDeps
 	broker  events.Broker
 	pruneMu sync.RWMutex
-	wake    chan struct{}
+	pools   []*drainPool
 	runCtx  context.Context
 
 	gatesMu sync.Mutex
@@ -61,7 +70,7 @@ func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg
 	w := &Worker{
 		deps:     workerDeps{ds: ds, store: store, agents: ag, ffmpeg: ffmpeg, cache: imgCache},
 		broker:   broker,
-		wake:     make(chan struct{}, 1),
+		pools:    newDrainPools(),
 		runCtx:   context.Background(),
 		gates:    map[string]*extGate{},
 		inFlight: map[string]struct{}{},
@@ -70,29 +79,63 @@ func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg
 	return w
 }
 
+// newDrainPools splits the drain by what bounds it: gate() waits for its rate-limit permit
+// while holding a slot, so a sleeping lookup would otherwise crowd out a cover sitting on disk.
+func newDrainPools() []*drainPool {
+	budget := conf.MaxOpenConns() // floored at 4, so both remainders below stay positive
+	local := min(max(1, conf.Server.ArtworkWorkerConcurrency), budget-1)
+	// More external slots than the rate allows would only sleep in the limiter.
+	external := min(max(2, 2*conf.Server.ArtworkExternalMaxRPS), budget-local)
+	return []*drainPool{
+		{name: "local", kinds: localDrainKinds, concurrency: local, wake: make(chan struct{}, 1)},
+		{name: "external", kinds: externalDrainKinds, concurrency: external, wake: make(chan struct{}, 1)},
+	}
+}
+
+// Kind is a proxy for cost: an album whose chain reaches "external" still costs a local slot,
+// but only the few with no local art do.
+var (
+	externalDrainKinds = []string{model.KindArtistArtwork.Prefix()}
+	localDrainKinds    = []string{
+		model.KindAlbumArtwork.Prefix(),
+		model.KindPlaylistArtwork.Prefix(),
+		model.KindRadioArtwork.Prefix(),
+		model.KindMediaFileArtwork.Prefix(),
+	}
+)
+
 // Run blocks draining the queue until ctx is cancelled. It exits cleanly with no
 // leaked goroutines: each drain waits for its batch before the loop can return.
 func (w *Worker) Run(ctx context.Context) error {
 	w.runCtx = ctx
-	concurrency := max(1, conf.Server.ArtworkWorkerConcurrency)
+	var wg sync.WaitGroup
+	for _, p := range w.pools {
+		wg.Go(func() { w.runPool(ctx, p) })
+	}
+	wg.Wait()
+	return nil
+}
+
+// runPool drains one pool's kinds until ctx is cancelled.
+func (w *Worker) runPool(ctx context.Context, p *drainPool) {
 	ticker := time.NewTicker(workerPollInterval)
 	defer ticker.Stop()
 	for {
-		n, err := w.drain(ctx, concurrency)
+		n, err := w.drain(ctx, p.concurrency, p.kinds...)
 		if err != nil && ctx.Err() == nil {
-			log.Warn(ctx, "artwork: worker drain failed", err)
+			log.Warn(ctx, "artwork: worker drain failed", "pool", p.name, err)
 		}
 		if ctx.Err() != nil {
-			return nil
+			return
 		}
 		if n > 0 {
-			continue // keep draining while the queue has ready work
+			continue // keep draining while this pool has ready work
 		}
 		select {
 		case <-ctx.Done():
-			return nil
+			return
 		case <-ticker.C:
-		case <-w.wake:
+		case <-p.wake:
 		}
 	}
 }
@@ -110,9 +153,13 @@ func (w *Worker) Bump(kind, id string) {
 		log.Warn("artwork: could not bump queue item", "kind", kind, "id", id, err)
 		return
 	}
-	select {
-	case w.wake <- struct{}{}:
-	default:
+	// Waking all beats routing by kind: a spurious wake costs one empty dequeue, while an
+	// unrouted kind would never wake at all.
+	for _, p := range w.pools {
+		select {
+		case p.wake <- struct{}{}:
+		default:
+		}
 	}
 }
 
@@ -124,11 +171,11 @@ func (w *Worker) RunPrune(ctx context.Context) error {
 	return Prune(ctx, w.deps.ds, w.deps.store)
 }
 
-func (w *Worker) drain(ctx context.Context, concurrency int) (int, error) {
+func (w *Worker) drain(ctx context.Context, concurrency int, kinds ...string) (int, error) {
 	// Dequeue well past the worker pool so a slow item (an external lookup burning its
 	// timeout) never idles the other slots: the pool stays fed until the batch runs out.
 	// DequeueBatch does not mark rows taken, so this is one query per pass, not per slot.
-	batch, err := w.deps.ds.ArtworkQueue(ctx).DequeueBatch(max(16, 4*concurrency))
+	batch, err := w.deps.ds.ArtworkQueue(ctx).DequeueBatch(max(16, 4*concurrency), kinds...)
 	if err != nil {
 		return 0, err
 	}

core/artwork/worker_test.go

@@ -55,8 +55,8 @@ type reenqueueOnDequeue struct {
 	done bool
 }
 
-func (r *reenqueueOnDequeue) DequeueBatch(n int) ([]model.ArtworkQueueItem, error) {
-	items, err := r.MockArtworkQueueRepo.DequeueBatch(n)
+func (r *reenqueueOnDequeue) DequeueBatch(n int, kinds ...string) ([]model.ArtworkQueueItem, error) {
+	items, err := r.MockArtworkQueueRepo.DequeueBatch(n, kinds...)
 	if !r.done && len(items) > 0 {
 		r.done = true
 		for k, it := range r.Data {
@@ -603,6 +603,68 @@ var _ = Describe("Worker", func() {
 		})
 	})
 
+	Describe("drain pools", func() {
+		// A kind in neither pool is never dequeued, with nothing to catch it at compile time.
+		It("covers every kind the worker can process, exactly once", func() {
+			var pooled []string
+			for _, p := range newDrainPools() {
+				pooled = append(pooled, p.kinds...)
+			}
+			for kind := range artworkKindToResource {
+				Expect(pooled).To(ContainElement(kind.Prefix()), "kind %q belongs to no drain pool", kind.Prefix())
+			}
+			Expect(pooled).To(HaveLen(len(artworkKindToResource)), "a kind is claimed by more than one pool")
+		})
+
+		// A first backfill enqueues artists before albums, and artists resolve through a
+		// rate-limited agent that holds its slot while waiting. Sharing one pool let ~29k
+		// sleeping lookups sit in front of every album for hours.
+		It("resolves albums while artists are stuck on a slow agent", func() {
+			conf.Server.CoverArtPriority = "cover.jpg"
+			conf.Server.ArtistArtPriority = "external"
+			folderRepo.result = []model.Folder{{
+				Path:       "tests/fixtures/artist/an-album",
+				ImageFiles: []string{"cover.jpg"},
+			}}
+			ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{{ID: "alx", Name: "Album", FolderIDs: []string{"f1"}}})
+			ds.MockedArtist = tests.CreateMockArtistRepo()
+
+			// Every artist lookup blocks until released, standing in for the rate limiter.
+			block := make(chan struct{})
+			DeferCleanup(func() { close(block) })
+			artists := model.Artists{}
+			for i := range 8 {
+				artists = append(artists, model.Artist{ID: fmt.Sprintf("arx%d", i), Name: "A"})
+			}
+			ds.MockedArtist.(*tests.MockArtistRepo).SetData(artists)
+			imageAgents(&fakeImageAgent{name: "slowAgent", block: block})
+
+			// Artists first, exactly as Backfill orders them.
+			for _, a := range artists {
+				Expect(queueRepo.Enqueue(model.ArtworkQueueItem{
+					ItemKind: "ar", ItemID: a.ID, Priority: model.ArtworkPriorityBackfill,
+				})).To(Succeed())
+			}
+			Expect(queueRepo.Enqueue(model.ArtworkQueueItem{
+				ItemKind: "al", ItemID: "alx", Priority: model.ArtworkPriorityBackfill,
+			})).To(Succeed())
+
+			runCtx, cancel := context.WithCancel(ctx)
+			DeferCleanup(cancel)
+			go func() { _ = w.Run(runCtx) }()
+
+			// The album must land while every artist is still parked in the agent.
+			Eventually(func() bool {
+				ia, err := artRepo.GetItemArtwork(model.KindAlbumArtwork, "alx", model.ImageTypePrimary)
+				return err == nil && ia.Hash != ""
+			}, 5*time.Second, 50*time.Millisecond).Should(BeTrue(),
+				"a blocked external pool must not hold up local artwork")
+
+			_, err := artRepo.GetItemArtwork(model.KindArtistArtwork, "arx0", model.ImageTypePrimary)
+			Expect(err).To(MatchError(model.ErrNotFound), "artists are still blocked, as intended")
+		})
+	})
+
 	Describe("batching", func() {
 		// The pool is fed from one dequeue per pass: a batch sized to the pool would make a
 		// single slow item idle the other slots for as long as it runs.

db/migrations/20260722023032_add_artwork_tables.sql

@@ -34,7 +34,9 @@ CREATE TABLE artwork_queue (
   PRIMARY KEY (item_kind, item_id, image_type)
 ) WITHOUT ROWID;
 -- Ordered to match DequeueBatch (priority DESC, enqueued_at) so drains stop after n rows; retry_at makes it covering.
-CREATE INDEX ix_artwork_queue_drain ON artwork_queue(priority DESC, enqueued_at, retry_at);
+-- item_kind leads: each drain pool dequeues only its own kinds, so it must seek straight to
+-- them rather than scan past another pool's backlog.
+CREATE INDEX ix_artwork_queue_drain ON artwork_queue(item_kind, priority DESC, enqueued_at, retry_at);
 
 -- +goose Down
 DROP TABLE artwork_queue;

model/artwork.go

@@ -99,7 +99,9 @@ type ArtworkQueueRepository interface {
 	// request-triggered read-through never resets a failed resolution's backoff.
 	EnqueueBump(items ...ArtworkQueueItem) error
 	// DequeueBatch returns up to n items with retry_at <= now, priority desc, enqueued_at asc.
-	DequeueBatch(n int) ([]ArtworkQueueItem, error)
+	// 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

persistence/artwork_queue_repository.go

@@ -59,11 +59,14 @@ func (r *artworkQueueRepository) enqueue(conflict string, items []model.ArtworkQ
 	return nil
 }
 
-func (r *artworkQueueRepository) DequeueBatch(n int) ([]model.ArtworkQueueItem, error) {
+func (r *artworkQueueRepository) DequeueBatch(n int, kinds ...string) ([]model.ArtworkQueueItem, error) {
 	sel := Select("*").From(r.tableName).
 		Where(LtOrEq{"retry_at": time.Now()}).
 		OrderBy("priority DESC", "enqueued_at ASC").
 		Limit(uint64(n))
+	if len(kinds) > 0 {
+		sel = sel.Where(Eq{"item_kind": kinds})
+	}
 	var res []model.ArtworkQueueItem
 	err := r.queryAll(sel, &res)
 	return res, err

tests/mock_artwork_queue_repo.go

@@ -1,6 +1,7 @@
 package tests
 
 import (
+	"slices"
 	"sort"
 	"sync"
 	"time"
@@ -53,7 +54,7 @@ func (m *MockArtworkQueueRepo) Enqueue(items ...model.ArtworkQueueItem) error {
 	return nil
 }
 
-func (m *MockArtworkQueueRepo) DequeueBatch(n int) ([]model.ArtworkQueueItem, error) {
+func (m *MockArtworkQueueRepo) DequeueBatch(n int, kinds ...string) ([]model.ArtworkQueueItem, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	if m.Err != nil {
@@ -62,7 +63,7 @@ func (m *MockArtworkQueueRepo) DequeueBatch(n int) ([]model.ArtworkQueueItem, er
 	var res []model.ArtworkQueueItem
 	now := time.Now()
 	for _, it := range m.Data {
-		if !it.RetryAt.After(now) {
+		if !it.RetryAt.After(now) && (len(kinds) == 0 || slices.Contains(kinds, it.ItemKind)) {
 			res = append(res, it)
 		}
 	}