From 97a7b084955e93e164f955edfeb5a8a665e122fc Mon Sep 17 00:00:00 2001 From: Deluan Date: Sat, 25 Jul 2026 14:39:44 -0400 Subject: [PATCH] fix(artwork): return undispatched items when a drain is cancelled MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 7 ++++++- core/artwork/worker_test.go | 23 +++++++++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/core/artwork/worker.go b/core/artwork/worker.go index c2513baf0..3f280c034 100644 --- a/core/artwork/worker.go +++ b/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 } diff --git a/core/artwork/worker_test.go b/core/artwork/worker_test.go index 076b1cbc9..54ac5d09f 100644 --- a/core/artwork/worker_test.go +++ b/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() {