diff --git a/core/artwork/processor.go b/core/artwork/processor.go index dc3e73707..f179370fc 100644 --- a/core/artwork/processor.go +++ b/core/artwork/processor.go @@ -9,6 +9,7 @@ import ( "image/draw" _ "image/gif" // the only artwork format with no other importer in this package "io" + "sync" "time" "github.com/navidrome/navidrome/core/agents" @@ -44,15 +45,17 @@ const maxImageBytes = 20 << 20 // huge canvas that image.Decode would expand into gigabytes (decompression bomb). const maxImagePixels = 64 << 20 -// workerDeps are the collaborators processItem needs; gate is set by NewWorker in -// production and nil only in tests, where resolveItem falls back to a plain passthrough. +// workerDeps are the collaborators processItem needs; gate and pruneLock are set by NewWorker +// in production and nil only in tests, where resolveItem falls back to a plain passthrough and +// nothing prunes. type workerDeps struct { - ds model.DataStore - store *ImageStore - agents *agents.Agents - ffmpeg ffmpeg.FFmpeg - cache cache.FileCache - gate gateFunc + ds model.DataStore + store *ImageStore + agents *agents.Agents + ffmpeg ffmpeg.FFmpeg + cache cache.FileCache + gate gateFunc + pruneLock sync.Locker } // acquired is what processItem persisted, handed back so the caller can warm the resize @@ -112,14 +115,34 @@ func processItem(ctx context.Context, deps *workerDeps, item model.ArtworkQueueI } art.SizeBytes = int64(len(data)) - sourcePath, refMtime, err := placeBytes(deps.store, art, res, data) + ia, err := persist(deps, repo, item, hash, art, res, data) if err != nil { - log.Warn(ctx, "artwork: failed to write image store", "kind", item.ItemKind, "id", item.ItemID, err) + log.Warn(ctx, "artwork: failed to persist resolved image", "kind", item.ItemKind, "id", item.ItemID, err) return outcomeFailed, nil } + got := &acquired{ia: ia, mime: art.Mime, data: data} + if res.extError { + return outcomeFoundStale, got + } + return outcomeFound, got +} + +// persist places the bytes and commits the rows referencing them. Only this window excludes +// Prune, which reclaims store files no row points at; resolution stays outside so a slow fetch +// cannot hold prune off. +func persist(deps *workerDeps, repo model.ArtworkRepository, item model.ArtworkQueueItem, hash string, + art *model.Artwork, res resolution, data []byte, +) (*model.ItemArtwork, error) { + if deps.pruneLock != nil { + deps.pruneLock.Lock() + defer deps.pruneLock.Unlock() + } + sourcePath, refMtime, err := placeBytes(deps.store, art, res, data) + if err != nil { + return nil, fmt.Errorf("writing image store: %w", err) + } if err := repo.PutImage(art); err != nil { - log.Warn(ctx, "artwork: failed to persist artwork image", "kind", item.ItemKind, "id", item.ItemID, err) - return outcomeFailed, nil + return nil, fmt.Errorf("persisting artwork image: %w", err) } ia := &model.ItemArtwork{ ItemKind: item.ItemKind, @@ -133,14 +156,9 @@ func processItem(ctx context.Context, deps *workerDeps, item model.ArtworkQueueI } // PutItemArtwork stamps UpdatedAt on ia, so what it holds now matches the persisted row. if err := repo.PutItemArtwork(ia); err != nil { - log.Warn(ctx, "artwork: failed to persist item artwork state", "kind", item.ItemKind, "id", item.ItemID, err) - return outcomeFailed, nil + return nil, fmt.Errorf("persisting item artwork state: %w", err) } - got := &acquired{ia: ia, mime: art.Mime, data: data} - if res.extError { - return outcomeFoundStale, got - } - return outcomeFound, got + return ia, nil } // writeAbsent records a known-absent state: every local/external source answered definitively "no". diff --git a/core/artwork/processor_test.go b/core/artwork/processor_test.go index adf11a75a..c6fb3321d 100644 --- a/core/artwork/processor_test.go +++ b/core/artwork/processor_test.go @@ -101,6 +101,43 @@ var _ = Describe("processItem", func() { Expect(os.IsNotExist(err)).To(BeTrue(), "folder-backed art must not be duplicated into the store") }) + Describe("prune lock scope", func() { + var lock *countingLocker + + BeforeEach(func() { + lock = &countingLocker{} + deps.pruneLock = lock + }) + + It("holds it once across the write window", func() { + folderRepo.result = []model.Folder{{ + Path: "tests/fixtures/artist/an-album", + ImageFiles: []string{"cover.jpg"}, + }} + ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{ + {ID: "alL1", Name: "Album", FolderIDs: []string{"f1"}}, + }) + + out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "alL1"}) + Expect(out).To(Equal(outcomeFound)) + Expect(lock.locks).To(BeNumerically(">", 0), "the write window must exclude prune") + Expect(lock.held()).To(BeFalse(), "the window must close before processItem returns") + }) + + // Resolution can reach the network under its own timeout, so holding the lock across it + // would let one slow provider block prune, and every drain queued behind prune's writer. + It("never takes it while only resolving", func() { + folderRepo.result = nil + ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{ + {ID: "alL2", Name: "Album", FolderIDs: []string{"f1"}}, + }) + + out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "alL2"}) + Expect(out).To(Equal(outcomeAbsent)) + Expect(lock.locks).To(BeZero()) + }) + }) + It("found-embedded: writes a store file and computes a non-empty blurhash from a real fixture", func() { ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{ {ID: "al2", Name: "Album", EmbedArtPath: "tests/fixtures/artist/an-album/test.mp3", FolderIDs: []string{"f1"}}, @@ -413,3 +450,14 @@ var _ = Describe("processItem", func() { Expect(err).To(MatchError(model.ErrNotFound)) }) }) + +// countingLocker stands in for the worker's prune read-lock, recording how often and how long +// processItem holds it. +type countingLocker struct { + locks int + unlocks int +} + +func (l *countingLocker) Lock() { l.locks++ } +func (l *countingLocker) Unlock() { l.unlocks++ } +func (l *countingLocker) held() bool { return l.locks != l.unlocks } diff --git a/core/artwork/worker.go b/core/artwork/worker.go index 03f2aca86..eb935d781 100644 --- a/core/artwork/worker.go +++ b/core/artwork/worker.go @@ -50,7 +50,7 @@ type drainPool 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 +// and circuit-broken independently, and prune is serialized against the store-write window // via pruneMu. type Worker struct { deps workerDeps @@ -61,21 +61,18 @@ type Worker struct { gatesMu sync.Mutex gates map[string]*extGate - - mu sync.Mutex - inFlight map[string]struct{} } func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg ffmpeg.FFmpeg, broker events.Broker, imgCache cache.FileCache) *Worker { w := &Worker{ - deps: workerDeps{ds: ds, store: store, agents: ag, ffmpeg: ffmpeg, cache: imgCache}, - broker: broker, - pools: newDrainPools(), - runCtx: context.Background(), - gates: map[string]*extGate{}, - inFlight: map[string]struct{}{}, + deps: workerDeps{ds: ds, store: store, agents: ag, ffmpeg: ffmpeg, cache: imgCache}, + broker: broker, + pools: newDrainPools(), + runCtx: context.Background(), + gates: map[string]*extGate{}, } w.deps.gate = w.gate + w.deps.pruneLock = w.pruneMu.RLocker() return w } @@ -175,11 +172,10 @@ func (w *Worker) drain(ctx context.Context, concurrency int, kinds ...string) (i // 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), kinds...) + items, err := w.deps.ds.ArtworkQueue(ctx).DequeueBatch(max(16, 4*concurrency), kinds...) if err != nil { return 0, err } - items := w.claim(batch) if len(items) == 0 { return 0, nil } @@ -190,30 +186,22 @@ 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 i, item := range items { + for _, 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 } - wg.Add(1) - go func(it model.ArtworkQueueItem) { - defer wg.Done() + wg.Go(func() { defer func() { <-sem }() - defer w.release(it) - out, got := w.process(ctx, it) + out, got := w.process(ctx, item) // Refresh clients on any visible state change: found/foundStale (new art) and absent // (removed art — clients must drop a previously-served immutable cover). foundStale // also wrote a served state row. if out == outcomeFound || out == outcomeFoundStale || out == outcomeAbsent { refreshMu.Lock() - refresh = append(refresh, it) + refresh = append(refresh, item) refreshMu.Unlock() } // Precache only actual images. Post-outcome only: the queue row was already settled @@ -221,7 +209,7 @@ func (w *Worker) drain(ctx context.Context, concurrency int, kinds ...string) (i if got != nil { w.precache(ctx, got) } - }(item) + }) } wg.Wait() w.broadcastRefresh(ctx, refresh) @@ -265,9 +253,7 @@ func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) (outc if item.ImageType == "" { item.ImageType = model.ImageTypePrimary } - w.pruneMu.RLock() out, got := processItem(ctx, &w.deps, item) - w.pruneMu.RUnlock() queue := w.deps.ds.ArtworkQueue(ctx) switch out { @@ -337,33 +323,6 @@ func (w *Worker) precache(ctx context.Context, got *acquired) { _ = stream.Close() } -// claim reserves items not already in flight, so a row appearing twice within a single -// batch is processed once. -func (w *Worker) claim(batch []model.ArtworkQueueItem) []model.ArtworkQueueItem { - w.mu.Lock() - defer w.mu.Unlock() - var out []model.ArtworkQueueItem - for _, it := range batch { - k := queueKey(it) - if _, busy := w.inFlight[k]; busy { - continue - } - w.inFlight[k] = struct{}{} - out = append(out, it) - } - return out -} - -func (w *Worker) release(it model.ArtworkQueueItem) { - w.mu.Lock() - delete(w.inFlight, queueKey(it)) - w.mu.Unlock() -} - -func queueKey(it model.ArtworkQueueItem) string { - return it.ItemKind + "|" + it.ItemID + "|" + it.ImageType -} - // gate wraps a named external step with that agent's own rate limiter and circuit // breaker, matching gateFunc so it can be injected via workerDeps.gate. func (w *Worker) gate(name string, f func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) { diff --git a/core/artwork/worker_test.go b/core/artwork/worker_test.go index af76cefc3..5271592ba 100644 --- a/core/artwork/worker_test.go +++ b/core/artwork/worker_test.go @@ -673,9 +673,9 @@ 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() { + // A cancelled drain leaves its undispatched rows untouched in the queue, so a later + // drain picks them up unchanged. + It("leaves undispatched items queued 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"}}) @@ -689,11 +689,10 @@ var _ = Describe("Worker", func() { _, 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") + for i := range 8 { + id := fmt.Sprintf("alc%d", i) + Expect(findQueued(queueRepo, "al", id)).ToNot(BeNil(), "row "+id+" must survive a cancelled drain") + } }) // The pool is fed from one dequeue per pass: a batch sized to the pool would make a