From 0018a64115548fe29e14c6b123719fd98b194955 Mon Sep 17 00:00:00 2001
From: Deluan <deluan@navidrome.org>
Date: Sat, 25 Jul 2026 11:48:14 -0400
Subject: [PATCH] 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      | 12 ++++++++++--
 core/artwork/worker_test.go | 18 ++++++++++++++++++
 2 files changed, 28 insertions(+), 2 deletions(-)

diff --git a/core/artwork/worker.go b/core/artwork/worker.go
index 4adf53b5e..01c987603 100644
--- a/core/artwork/worker.go
+++ b/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()
diff --git a/core/artwork/worker_test.go b/core/artwork/worker_test.go
index 41032bdd4..684efba67 100644
--- a/core/artwork/worker_test.go
+++ b/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())