fix(artwork): return undispatched items when a drain is cancelled

claim() reserves the whole batch before dispatch, but the cancellation
path returned without releasing what it had not yet started, leaving
those items in the in-flight set permanently — no later drain could
claim them again.

Harmless until the batch grew past the pool size; now a cancel strands
up to a full batch. The e2e harness cancels mid-drain after every
acquire, so it surfaced there first: one spec timed out waiting for an
item that had been claimed and abandoned, and the suite went from 59s
to 87s on CI.

core/artwork/worker.go

@@ -190,10 +190,15 @@ func (w *Worker) drain(ctx context.Context, concurrency int, kinds ...string) (i
 	var wg sync.WaitGroup
 	var refreshMu sync.Mutex
 	var refresh []model.ArtworkQueueItem
-	for _, item := range items {
+	for i, item := range items {
 		select {
 		case sem <- struct{}{}:
 		case <-ctx.Done():
+			// claim() reserved the whole batch; anything not dispatched has to go back, or it
+			// stays in flight forever and no later drain can pick it up.
+			for _, undispatched := range items[i:] {
+				w.release(undispatched)
+			}
 			wg.Wait()
 			return len(items), nil
 		}

core/artwork/worker_test.go

@@ -672,6 +672,29 @@ var _ = Describe("Worker", func() {
 	})
 
 	Describe("batching", func() {
+		// claim() reserves the whole batch before dispatching, so a cancel mid-batch used to
+		// strand the rest in the in-flight set, where no later drain could ever claim them.
+		It("returns undispatched items to the pool when cancelled mid-batch", func() {
+			for i := range 8 {
+				id := fmt.Sprintf("alc%d", i)
+				ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{{ID: id, Name: "Album"}})
+				Expect(queueRepo.Enqueue(model.ArtworkQueueItem{
+					ItemKind: "al", ItemID: id, Priority: model.ArtworkPriorityScan,
+				})).To(Succeed())
+			}
+			cancelledCtx, cancel := context.WithCancel(ctx)
+			cancel()
+
+			_, err := w.drain(cancelledCtx, 1)
+			Expect(err).ToNot(HaveOccurred())
+
+			// Every item must be claimable again; a stranded one would be silently skipped.
+			w.mu.Lock()
+			stranded := len(w.inFlight)
+			w.mu.Unlock()
+			Expect(stranded).To(BeZero(), "a cancelled drain must not strand claimed items")
+		})
+
 		// 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() {