fix(artwork): let cancelled work unwind before the worker stops

On shutdown the drain returned while cancelled acquires were still running,
so Run could return, and the DB close, before they finished. The race
detector caught one of them reading config while the e2e suite restored it.

When the context is cancelled, process now waits up to shutdownGrace (5s)
for the cancelled acquire to return. Work that honors ctx unwinds in
milliseconds, so Run joins it; only a stuck item adds the grace period.

core/artwork/worker.go

@@ -30,6 +30,8 @@ const (
 	giveUpAfter = 12 * time.Hour
 	// itemTimeout must outlast a slow but healthy item: each image agent's call plus its download, in turn.
 	itemTimeout = 3 * time.Minute
+	// shutdownGrace lets cancelled work unwind before Run returns and the DB closes, without waiting on a stuck item.
+	shutdownGrace = 5 * time.Second
 )
 
 // drainPool drains one class of work with its own slot budget, so a blocking kind cannot
@@ -287,6 +289,10 @@ func (w *Worker) acquireWithTimeout(ctx context.Context, item model.ArtworkQueue
 	case r := <-done:
 		return r.out, r.got, r.retryIn
 	case <-ctx.Done():
+		select {
+		case <-done:
+		case <-time.After(shutdownGrace):
+		}
 		return outcomeFailed, nil, 0
 	case <-time.After(itemTimeout):
 		log.Error(ctx, "Artwork: Item timed out, moving on", "kind", item.ItemKind, "id", item.ItemID, "timeout", itemTimeout)

core/artwork/worker_test.go

@@ -838,11 +838,6 @@ var _ = Describe("Worker", func() {
 				cancel()
 				close(block) // unpark the blocked lookups so the pools can unwind
 				<-done
-				// Run returns without waiting on the lookups, so wait for them before config restores.
-				Eventually(func() (n int) {
-					w.busy.Range(func(_, _ any) bool { n++; return true })
-					return n
-				}).Should(BeZero())
 			})
 
 			Eventually(func() bool {

core/artwork/worker_timing_test.go

@@ -177,8 +177,32 @@ func TestArtworkDrainMovesOnFromAStuckItem(t *testing.T) {
 	})
 }
 
-// Shutdown must not wait on a stuck item, and must leave its row for the next run.
-func TestArtworkDrainStopsAtOnceOnShutdown(t *testing.T) {
+// Shutdown joins cancelled work that unwinds, so nothing touches the DB after Run returns.
+func TestArtworkDrainShutdownWaitsForWorkToUnwind(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		g := NewWithT(t)
+		w, queue, _, block := newStuckItemWorker(t)
+		before := *findQueued(queue, "ar", "ar1")
+		ctx, cancel := context.WithCancel(t.Context())
+		go func() {
+			time.Sleep(time.Second)
+			cancel()
+			time.Sleep(100 * time.Millisecond)
+			close(block)
+		}()
+
+		start := time.Now()
+		_, err := w.drain(ctx, 1)
+		g.Expect(err).ToNot(HaveOccurred())
+		g.Expect(time.Since(start)).To(Equal(time.Second+100*time.Millisecond), "the drain waits for the cancelled item to return")
+		_, running := w.busy.Load(itemKey{"ar", "ar1", model.ImageTypePrimary})
+		g.Expect(running).To(BeFalse())
+		g.Expect(*findQueued(queue, "ar", "ar1")).To(Equal(before), "the row is left for the next run")
+	})
+}
+
+// Shutdown must not wait on a stuck item for longer than the grace period.
+func TestArtworkDrainShutdownDoesNotWaitOnAStuckItem(t *testing.T) {
 	synctest.Test(t, func(t *testing.T) {
 		g := NewWithT(t)
 		w, queue, _, block := newStuckItemWorker(t)
@@ -192,8 +216,8 @@ func TestArtworkDrainStopsAtOnceOnShutdown(t *testing.T) {
 		start := time.Now()
 		_, err := w.drain(ctx, 1)
 		g.Expect(err).ToNot(HaveOccurred())
-		g.Expect(time.Since(start)).To(Equal(time.Second), "the drain returns as soon as the context is cancelled")
-		g.Expect(*findQueued(queue, "ar", "ar1")).To(Equal(before), "the row is left as it was")
+		g.Expect(time.Since(start)).To(Equal(time.Second+shutdownGrace), "the drain gives up on the stuck item after the grace period")
+		g.Expect(*findQueued(queue, "ar", "ar1")).To(Equal(before), "the row is left for the next run")
 
 		close(block)
 		synctest.Wait()