diff --git a/core/artwork/worker.go b/core/artwork/worker.go index 62c6a7ea6..6122f88f2 100644 --- a/core/artwork/worker.go +++ b/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) diff --git a/core/artwork/worker_test.go b/core/artwork/worker_test.go index 95b2df0a6..a6c07b763 100644 --- a/core/artwork/worker_test.go +++ b/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 { diff --git a/core/artwork/worker_timing_test.go b/core/artwork/worker_timing_test.go index d4aa79d26..7896f6f09 100644 --- a/core/artwork/worker_timing_test.go +++ b/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()