From 6823bfd43648aff85f6e5a4127c4b606ebbd4260 Mon Sep 17 00:00:00 2001 From: Deluan Date: Sat, 25 Jul 2026 13:32:07 -0400 Subject: [PATCH] perf(artwork): drain local and external artwork in separate pools A first backfill enqueues artists before albums at a single priority, so the drain took them in that order. Artists resolve through a rate-limited agent, and gate() waits for its permit while holding a worker slot, so the whole pool sat asleep in the limiter with every album queued behind it. Measured on a 96k-track library (29,115 artists to 6,949 albums, 4:1): zero albums resolved in seven minutes, and roughly 3.3 hours before the first album cover would have appeared. Splitting the drain gives each class its own slots: albums now finish in under eight minutes while artists trickle at the same 2/s they were always limited to. The two budgets are carved out of MaxOpenConns so a second pool cannot take connections the scanner and the UI need. Dequeue filters by kind, and the drain index leads with item_kind so each pool seeks to its own work instead of scanning past the other's backlog. --- core/artwork/agent_images_test.go | 5 ++ core/artwork/worker.go | 75 +++++++++++++++---- core/artwork/worker_test.go | 66 +++++++++++++++- .../20260722023032_add_artwork_tables.sql | 4 +- model/artwork.go | 4 +- persistence/artwork_queue_repository.go | 5 +- tests/mock_artwork_queue_repo.go | 5 +- 7 files changed, 143 insertions(+), 21 deletions(-) diff --git a/core/artwork/agent_images_test.go b/core/artwork/agent_images_test.go index ea58fe0ac..d8332d66a 100644 --- a/core/artwork/agent_images_test.go +++ b/core/artwork/agent_images_test.go @@ -28,11 +28,16 @@ type fakeImageAgent struct { albumCalls int gotArtistName string gotAlbumName string + // block, when set, holds every lookup until closed, standing in for a slow/rate-limited agent. + block chan struct{} } func (f *fakeImageAgent) AgentName() string { return f.name } func (f *fakeImageAgent) GetArtistImages(_ context.Context, _, name, _ string) ([]agents.ExternalImage, error) { + if f.block != nil { + <-f.block + } f.artistCalls++ f.gotArtistName = name return f.imgs, f.err diff --git a/core/artwork/worker.go b/core/artwork/worker.go index 01c987603..c2513baf0 100644 --- a/core/artwork/worker.go +++ b/core/artwork/worker.go @@ -40,6 +40,15 @@ type extGate struct { breaker *breaker } +// drainPool drains one class of work with its own slot budget, so a kind whose resolution +// blocks cannot occupy slots another kind needs. +type drainPool struct { + name string + kinds []string + concurrency int + wake chan struct{} +} + // Worker drains the artwork queue through processItem: each external agent is rate-limited // and circuit-broken independently, and prune is serialized against in-flight acquisitions // via pruneMu. @@ -47,7 +56,7 @@ type Worker struct { deps workerDeps broker events.Broker pruneMu sync.RWMutex - wake chan struct{} + pools []*drainPool runCtx context.Context gatesMu sync.Mutex @@ -61,7 +70,7 @@ func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg w := &Worker{ deps: workerDeps{ds: ds, store: store, agents: ag, ffmpeg: ffmpeg, cache: imgCache}, broker: broker, - wake: make(chan struct{}, 1), + pools: newDrainPools(), runCtx: context.Background(), gates: map[string]*extGate{}, inFlight: map[string]struct{}{}, @@ -70,29 +79,63 @@ func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg return w } +// newDrainPools splits the drain by what bounds it: gate() waits for its rate-limit permit +// while holding a slot, so a sleeping lookup would otherwise crowd out a cover sitting on disk. +func newDrainPools() []*drainPool { + budget := conf.MaxOpenConns() // floored at 4, so both remainders below stay positive + local := min(max(1, conf.Server.ArtworkWorkerConcurrency), budget-1) + // More external slots than the rate allows would only sleep in the limiter. + external := min(max(2, 2*conf.Server.ArtworkExternalMaxRPS), budget-local) + return []*drainPool{ + {name: "local", kinds: localDrainKinds, concurrency: local, wake: make(chan struct{}, 1)}, + {name: "external", kinds: externalDrainKinds, concurrency: external, wake: make(chan struct{}, 1)}, + } +} + +// Kind is a proxy for cost: an album whose chain reaches "external" still costs a local slot, +// but only the few with no local art do. +var ( + externalDrainKinds = []string{model.KindArtistArtwork.Prefix()} + localDrainKinds = []string{ + model.KindAlbumArtwork.Prefix(), + model.KindPlaylistArtwork.Prefix(), + model.KindRadioArtwork.Prefix(), + model.KindMediaFileArtwork.Prefix(), + } +) + // Run blocks draining the queue until ctx is cancelled. It exits cleanly with no // leaked goroutines: each drain waits for its batch before the loop can return. func (w *Worker) Run(ctx context.Context) error { w.runCtx = ctx - concurrency := max(1, conf.Server.ArtworkWorkerConcurrency) + var wg sync.WaitGroup + for _, p := range w.pools { + wg.Go(func() { w.runPool(ctx, p) }) + } + wg.Wait() + return nil +} + +// runPool drains one pool's kinds until ctx is cancelled. +func (w *Worker) runPool(ctx context.Context, p *drainPool) { ticker := time.NewTicker(workerPollInterval) defer ticker.Stop() for { - n, err := w.drain(ctx, concurrency) + n, err := w.drain(ctx, p.concurrency, p.kinds...) if err != nil && ctx.Err() == nil { - log.Warn(ctx, "artwork: worker drain failed", err) + log.Warn(ctx, "artwork: worker drain failed", "pool", p.name, err) } if ctx.Err() != nil { - return nil + return } if n > 0 { - continue // keep draining while the queue has ready work + continue // keep draining while this pool has ready work } select { case <-ctx.Done(): - return nil + return case <-ticker.C: - case <-w.wake: + case <-p.wake: } } } @@ -110,9 +153,13 @@ func (w *Worker) Bump(kind, id string) { log.Warn("artwork: could not bump queue item", "kind", kind, "id", id, err) return } - select { - case w.wake <- struct{}{}: - default: + // Waking all beats routing by kind: a spurious wake costs one empty dequeue, while an + // unrouted kind would never wake at all. + for _, p := range w.pools { + select { + case p.wake <- struct{}{}: + default: + } } } @@ -124,11 +171,11 @@ func (w *Worker) RunPrune(ctx context.Context) error { return Prune(ctx, w.deps.ds, w.deps.store) } -func (w *Worker) drain(ctx context.Context, concurrency int) (int, error) { +func (w *Worker) drain(ctx context.Context, concurrency int, kinds ...string) (int, error) { // 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)) + batch, err := w.deps.ds.ArtworkQueue(ctx).DequeueBatch(max(16, 4*concurrency), kinds...) if err != nil { return 0, err } diff --git a/core/artwork/worker_test.go b/core/artwork/worker_test.go index 684efba67..794f9190f 100644 --- a/core/artwork/worker_test.go +++ b/core/artwork/worker_test.go @@ -55,8 +55,8 @@ type reenqueueOnDequeue struct { done bool } -func (r *reenqueueOnDequeue) DequeueBatch(n int) ([]model.ArtworkQueueItem, error) { - items, err := r.MockArtworkQueueRepo.DequeueBatch(n) +func (r *reenqueueOnDequeue) DequeueBatch(n int, kinds ...string) ([]model.ArtworkQueueItem, error) { + items, err := r.MockArtworkQueueRepo.DequeueBatch(n, kinds...) if !r.done && len(items) > 0 { r.done = true for k, it := range r.Data { @@ -603,6 +603,68 @@ var _ = Describe("Worker", func() { }) }) + Describe("drain pools", func() { + // A kind in neither pool is never dequeued, with nothing to catch it at compile time. + It("covers every kind the worker can process, exactly once", func() { + var pooled []string + for _, p := range newDrainPools() { + pooled = append(pooled, p.kinds...) + } + for kind := range artworkKindToResource { + Expect(pooled).To(ContainElement(kind.Prefix()), "kind %q belongs to no drain pool", kind.Prefix()) + } + Expect(pooled).To(HaveLen(len(artworkKindToResource)), "a kind is claimed by more than one pool") + }) + + // A first backfill enqueues artists before albums, and artists resolve through a + // rate-limited agent that holds its slot while waiting. Sharing one pool let ~29k + // sleeping lookups sit in front of every album for hours. + It("resolves albums while artists are stuck on a slow agent", func() { + conf.Server.CoverArtPriority = "cover.jpg" + conf.Server.ArtistArtPriority = "external" + folderRepo.result = []model.Folder{{ + Path: "tests/fixtures/artist/an-album", + ImageFiles: []string{"cover.jpg"}, + }} + ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{{ID: "alx", Name: "Album", FolderIDs: []string{"f1"}}}) + ds.MockedArtist = tests.CreateMockArtistRepo() + + // Every artist lookup blocks until released, standing in for the rate limiter. + block := make(chan struct{}) + DeferCleanup(func() { close(block) }) + artists := model.Artists{} + for i := range 8 { + artists = append(artists, model.Artist{ID: fmt.Sprintf("arx%d", i), Name: "A"}) + } + ds.MockedArtist.(*tests.MockArtistRepo).SetData(artists) + imageAgents(&fakeImageAgent{name: "slowAgent", block: block}) + + // Artists first, exactly as Backfill orders them. + for _, a := range artists { + Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ + ItemKind: "ar", ItemID: a.ID, Priority: model.ArtworkPriorityBackfill, + })).To(Succeed()) + } + Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ + ItemKind: "al", ItemID: "alx", Priority: model.ArtworkPriorityBackfill, + })).To(Succeed()) + + runCtx, cancel := context.WithCancel(ctx) + DeferCleanup(cancel) + go func() { _ = w.Run(runCtx) }() + + // The album must land while every artist is still parked in the agent. + Eventually(func() bool { + ia, err := artRepo.GetItemArtwork(model.KindAlbumArtwork, "alx", model.ImageTypePrimary) + return err == nil && ia.Hash != "" + }, 5*time.Second, 50*time.Millisecond).Should(BeTrue(), + "a blocked external pool must not hold up local artwork") + + _, err := artRepo.GetItemArtwork(model.KindArtistArtwork, "arx0", model.ImageTypePrimary) + Expect(err).To(MatchError(model.ErrNotFound), "artists are still blocked, as intended") + }) + }) + 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. diff --git a/db/migrations/20260722023032_add_artwork_tables.sql b/db/migrations/20260722023032_add_artwork_tables.sql index bc400f90e..a7bd0897c 100644 --- a/db/migrations/20260722023032_add_artwork_tables.sql +++ b/db/migrations/20260722023032_add_artwork_tables.sql @@ -34,7 +34,9 @@ CREATE TABLE artwork_queue ( PRIMARY KEY (item_kind, item_id, image_type) ) WITHOUT ROWID; -- Ordered to match DequeueBatch (priority DESC, enqueued_at) so drains stop after n rows; retry_at makes it covering. -CREATE INDEX ix_artwork_queue_drain ON artwork_queue(priority DESC, enqueued_at, retry_at); +-- item_kind leads: each drain pool dequeues only its own kinds, so it must seek straight to +-- them rather than scan past another pool's backlog. +CREATE INDEX ix_artwork_queue_drain ON artwork_queue(item_kind, priority DESC, enqueued_at, retry_at); -- +goose Down DROP TABLE artwork_queue; diff --git a/model/artwork.go b/model/artwork.go index b2dae0024..72258b923 100644 --- a/model/artwork.go +++ b/model/artwork.go @@ -99,7 +99,9 @@ type ArtworkQueueRepository interface { // request-triggered read-through never resets a failed resolution's backoff. EnqueueBump(items ...ArtworkQueueItem) error // DequeueBatch returns up to n items with retry_at <= now, priority desc, enqueued_at asc. - DequeueBatch(n int) ([]ArtworkQueueItem, error) + // Restricted to the given item kinds when any are passed, so a drain pool sees only its own + // work and cannot be held up behind another kind's backlog. + DequeueBatch(n int, kinds ...string) ([]ArtworkQueueItem, error) // MarkFailed increments attempts and pushes retry_at into the future. MarkFailed(kind, id, imageType string, retryAt time.Time) error // MarkFailedIfUnchanged applies the failure backoff only while retry_at still matches diff --git a/persistence/artwork_queue_repository.go b/persistence/artwork_queue_repository.go index 4430efd5c..69af9c66a 100644 --- a/persistence/artwork_queue_repository.go +++ b/persistence/artwork_queue_repository.go @@ -59,11 +59,14 @@ func (r *artworkQueueRepository) enqueue(conflict string, items []model.ArtworkQ return nil } -func (r *artworkQueueRepository) DequeueBatch(n int) ([]model.ArtworkQueueItem, error) { +func (r *artworkQueueRepository) DequeueBatch(n int, kinds ...string) ([]model.ArtworkQueueItem, error) { sel := Select("*").From(r.tableName). Where(LtOrEq{"retry_at": time.Now()}). OrderBy("priority DESC", "enqueued_at ASC"). Limit(uint64(n)) + if len(kinds) > 0 { + sel = sel.Where(Eq{"item_kind": kinds}) + } var res []model.ArtworkQueueItem err := r.queryAll(sel, &res) return res, err diff --git a/tests/mock_artwork_queue_repo.go b/tests/mock_artwork_queue_repo.go index 896516e02..b85515dab 100644 --- a/tests/mock_artwork_queue_repo.go +++ b/tests/mock_artwork_queue_repo.go @@ -1,6 +1,7 @@ package tests import ( + "slices" "sort" "sync" "time" @@ -53,7 +54,7 @@ func (m *MockArtworkQueueRepo) Enqueue(items ...model.ArtworkQueueItem) error { return nil } -func (m *MockArtworkQueueRepo) DequeueBatch(n int) ([]model.ArtworkQueueItem, error) { +func (m *MockArtworkQueueRepo) DequeueBatch(n int, kinds ...string) ([]model.ArtworkQueueItem, error) { m.mu.Lock() defer m.mu.Unlock() if m.Err != nil { @@ -62,7 +63,7 @@ func (m *MockArtworkQueueRepo) DequeueBatch(n int) ([]model.ArtworkQueueItem, er var res []model.ArtworkQueueItem now := time.Now() for _, it := range m.Data { - if !it.RetryAt.After(now) { + if !it.RetryAt.After(now) && (len(kinds) == 0 || slices.Contains(kinds, it.ItemKind)) { res = append(res, it) } }