perf(artwork): keep the worker pool fed across a drain

The pool was fed from a batch sized to the pool itself (2x concurrency)
with a WaitGroup barrier before the next dequeue, so one item burning
its external timeout idled every other slot until it finished. The
legacy cache warmer had no such barrier: it streamed through a pipeline
of 4.

Dequeuing well past the pool keeps the slots fed for the whole pass at
no extra cost, since DequeueBatch does not mark rows taken and was
already one query per pass. Acquiring a slot now also observes
cancellation, so a larger batch cannot delay shutdown.

core/artwork/worker.go

@@ -125,7 +125,10 @@ func (w *Worker) RunPrune(ctx context.Context) error {
 }
 
 func (w *Worker) drain(ctx context.Context, concurrency int) (int, error) {
-	batch, err := w.deps.ds.ArtworkQueue(ctx).DequeueBatch(2 * concurrency)
+	// 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))
 	if err != nil {
 		return 0, err
 	}
@@ -141,7 +144,12 @@ func (w *Worker) drain(ctx context.Context, concurrency int) (int, error) {
 	var refreshMu sync.Mutex
 	var refresh []model.ArtworkQueueItem
 	for _, item := range items {
-		sem <- struct{}{}
+		select {
+		case sem <- struct{}{}:
+		case <-ctx.Done():
+			wg.Wait()
+			return len(items), nil
+		}
 		wg.Add(1)
 		go func(it model.ArtworkQueueItem) {
 			defer wg.Done()

core/artwork/worker_test.go

@@ -3,6 +3,7 @@ package artwork
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"net/http"
 	"os"
@@ -602,6 +603,23 @@ var _ = Describe("Worker", func() {
 		})
 	})
 
+	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.
+		It("dequeues past the worker pool so one drain covers many items", func() {
+			for i := range 16 {
+				ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{{ID: fmt.Sprintf("alb%d", i), Name: "Album"}})
+				Expect(queueRepo.Enqueue(model.ArtworkQueueItem{
+					ItemKind: "al", ItemID: fmt.Sprintf("alb%d", i), Priority: model.ArtworkPriorityScan,
+				})).To(Succeed())
+			}
+
+			n, err := w.drain(ctx, 2)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(n).To(Equal(16), "a batch sized to the pool would have stopped at 4")
+		})
+	})
+
 	Describe("RunPrune", func() {
 		It("runs a prune under the worker mutex", func() {
 			Expect(w.RunPrune(ctx)).To(Succeed())