diff --git a/core/artwork/processor.go b/core/artwork/processor.go index a62cf131c..33451ab90 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/artwork/blurhash" @@ -41,7 +42,7 @@ const maxImageBytes = 20 << 20 // huge canvas that image.Decode would expand into gigabytes (decompression bomb). const maxImagePixels = 64 << 20 -// acquired is what processItem persisted, handed back so the caller can warm the resize +// acquired is what a successful acquire persisted, handed back so the caller can warm the resize // cache without re-reading the rows and the file it just wrote. type acquired struct { ia *model.ItemArtwork @@ -49,12 +50,21 @@ type acquired struct { data []byte } -// processItem resolves one queue item end to end: find an image, hash/decode/ -// blurhash it, place its bytes, and persist the resulting state. -func processItem(ctx context.Context, deps *workerDeps, item model.ArtworkQueueItem) (outcome, *acquired) { - repo := deps.ds.Artwork(ctx) +// processor turns one queue item into stored artwork. It owns acquisition only — settling the +// queue row afterwards is the Worker's job. pruneLock is nil in tests, where nothing prunes. +type processor struct { + ds model.DataStore + store *ImageStore + resolver *resolver + pruneLock sync.Locker +} - res, err := deps.resolver.resolve(ctx, item) +// acquire resolves one queue item end to end: find an image, hash/decode/ +// blurhash it, place its bytes, and persist the resulting state. +func (p *processor) acquire(ctx context.Context, item model.ArtworkQueueItem) (outcome, *acquired) { + repo := p.ds.Artwork(ctx) + + res, err := p.resolver.resolve(ctx, item) if err != nil { log.Warn(ctx, "Artwork: Could not resolve item", "kind", item.ItemKind, "id", item.ItemID, err) return outcomeFailed, nil @@ -98,7 +108,7 @@ func processItem(ctx context.Context, deps *workerDeps, item model.ArtworkQueueI } art.SizeBytes = int64(len(data)) - ia, err := persist(deps, repo, item, art, res, data) + ia, err := p.persist(repo, item, art, res, data) if err != nil { log.Warn(ctx, "Artwork: Failed to persist resolved image", "kind", item.ItemKind, "id", item.ItemID, err) return outcomeFailed, nil @@ -113,14 +123,14 @@ func processItem(ctx context.Context, deps *workerDeps, item model.ArtworkQueueI // 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, +func (p *processor) persist(repo model.ArtworkRepository, item model.ArtworkQueueItem, art *model.Artwork, res resolution, data []byte, ) (*model.ItemArtwork, error) { - if deps.pruneLock != nil { - deps.pruneLock.Lock() - defer deps.pruneLock.Unlock() + if p.pruneLock != nil { + p.pruneLock.Lock() + defer p.pruneLock.Unlock() } - sourcePath, refMtime, err := placeBytes(deps.store, art, res, data) + sourcePath, refMtime, err := placeBytes(p.store, art, res, data) if err != nil { return nil, fmt.Errorf("writing image store: %w", err) } diff --git a/core/artwork/processor_test.go b/core/artwork/processor_test.go index d16179624..fcdd9de8a 100644 --- a/core/artwork/processor_test.go +++ b/core/artwork/processor_test.go @@ -37,7 +37,7 @@ func pngHeaderWithDims(w, h uint32) []byte { return binary.BigEndian.AppendUint32(out, crc32.ChecksumIEEE(chunk)) } -var _ = Describe("processItem", func() { +var _ = Describe("processor.acquire", func() { var ( ctx context.Context ds *tests.MockDataStore @@ -48,7 +48,7 @@ var _ = Describe("processItem", func() { store *ImageStore artRepo *tests.MockArtworkRepo repoRoot string - deps *workerDeps + proc *processor ) BeforeEach(func() { @@ -71,7 +71,7 @@ var _ = Describe("processItem", func() { } ds.MockedAlbum = tests.CreateMockAlbumRepo() store = NewImageStore(GinkgoT().TempDir()) - deps = &workerDeps{ds: ds, store: store, resolver: newResolver(ds, ag, ffm, nil)} + proc = &processor{ds: ds, store: store, resolver: newResolver(ds, ag, ffm, nil)} conf.Server.CoverArtPriority = "cover.jpg, embedded" }) @@ -85,7 +85,7 @@ var _ = Describe("processItem", func() { {ID: "al1", Name: "Album", FolderIDs: []string{"f1"}}, }) - out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al1"}) + out, _ := proc.acquire(ctx, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al1"}) Expect(out).To(Equal(outcomeFound)) ia, err := artRepo.GetItemArtwork(model.KindAlbumArtwork, "al1", model.ImageTypePrimary) @@ -106,7 +106,7 @@ var _ = Describe("processItem", func() { BeforeEach(func() { lock = &countingLocker{} - deps.pruneLock = lock + proc.pruneLock = lock }) It("holds it once across the write window", func() { @@ -118,10 +118,10 @@ var _ = Describe("processItem", func() { {ID: "alL1", Name: "Album", FolderIDs: []string{"f1"}}, }) - out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "alL1"}) + out, _ := proc.acquire(ctx, 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") + Expect(lock.held()).To(BeFalse(), "the window must close before acquire returns") }) // Resolution can reach the network under its own timeout, so holding the lock across it @@ -132,7 +132,7 @@ var _ = Describe("processItem", func() { {ID: "alL2", Name: "Album", FolderIDs: []string{"f1"}}, }) - out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "alL2"}) + out, _ := proc.acquire(ctx, model.ArtworkQueueItem{ItemKind: "al", ItemID: "alL2"}) Expect(out).To(Equal(outcomeAbsent)) Expect(lock.locks).To(BeZero()) }) @@ -144,7 +144,7 @@ var _ = Describe("processItem", func() { }) folderRepo.result = nil - out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al2"}) + out, _ := proc.acquire(ctx, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al2"}) Expect(out).To(Equal(outcomeFound)) ia, err := artRepo.GetItemArtwork(model.KindAlbumArtwork, "al2", model.ImageTypePrimary) @@ -167,7 +167,7 @@ var _ = Describe("processItem", func() { {ID: "al3", Name: "Album"}, }) - out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al3"}) + out, _ := proc.acquire(ctx, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al3"}) Expect(out).To(Equal(outcomeAbsent)) ia, err := artRepo.GetItemArtwork(model.KindAlbumArtwork, "al3", model.ImageTypePrimary) @@ -189,7 +189,7 @@ var _ = Describe("processItem", func() { {ID: "al-io", Name: "Album", FolderIDs: []string{"f1"}}, }) - out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al-io"}) + out, _ := proc.acquire(ctx, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al-io"}) Expect(out).To(Equal(outcomeFailed)) _, err := artRepo.GetItemArtwork(model.KindAlbumArtwork, "al-io", model.ImageTypePrimary) @@ -216,7 +216,7 @@ var _ = Describe("processItem", func() { DeferCleanup(func() { _ = os.Chmod(upload, 0o600) }) radioRepo.Data["ra-io"] = &model.Radio{ID: "ra-io", Name: "Station", UploadedImage: "ra-io.jpg"} - out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "ra", ItemID: "ra-io"}) + out, _ := proc.acquire(ctx, model.ArtworkQueueItem{ItemKind: "ra", ItemID: "ra-io"}) Expect(out).To(Equal(outcomeFailed)) _, err := artRepo.GetItemArtwork(model.KindRadioArtwork, "ra-io", model.ImageTypePrimary) @@ -230,7 +230,7 @@ var _ = Describe("processItem", func() { }) imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("agent timed out")}) - out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al4"}) + out, _ := proc.acquire(ctx, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al4"}) Expect(out).To(Equal(outcomeFailed)) _, err := artRepo.GetItemArtwork(model.KindAlbumArtwork, "al4", model.ImageTypePrimary) @@ -248,7 +248,7 @@ var _ = Describe("processItem", func() { }) imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("agent timed out")}) - out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "alstale"}) + out, _ := proc.acquire(ctx, model.ArtworkQueueItem{ItemKind: "al", ItemID: "alstale"}) Expect(out).To(Equal(outcomeFoundStale)) ia, err := artRepo.GetItemArtwork(model.KindAlbumArtwork, "alstale", model.ImageTypePrimary) @@ -269,7 +269,7 @@ var _ = Describe("processItem", func() { ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{{ID: "alext", Name: "Album"}}) imageAgents(&fakeImageAgent{name: "deezerFake", imgs: []agents.ExternalImage{{URL: srv.URL, Size: 500}}}) - out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "alext"}) + out, _ := proc.acquire(ctx, model.ArtworkQueueItem{ItemKind: "al", ItemID: "alext"}) Expect(out).To(Equal(outcomeFound)) ia, err := artRepo.GetItemArtwork(model.KindAlbumArtwork, "alext", model.ImageTypePrimary) @@ -295,7 +295,7 @@ var _ = Describe("processItem", func() { {ID: "al6", Name: "Album B", FolderIDs: []string{"f1"}}, }) - out1, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al5"}) + out1, _ := proc.acquire(ctx, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al5"}) Expect(out1).To(Equal(outcomeFound)) ia1, err := artRepo.GetItemArtwork(model.KindAlbumArtwork, "al5", model.ImageTypePrimary) Expect(err).ToNot(HaveOccurred()) @@ -306,7 +306,7 @@ var _ = Describe("processItem", func() { poisoned.BlurHash = "SENTINEL" artRepo.Data[ia1.Hash] = poisoned - out2, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al6"}) + out2, _ := proc.acquire(ctx, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al6"}) Expect(out2).To(Equal(outcomeFound)) ia2, err := artRepo.GetItemArtwork(model.KindAlbumArtwork, "al6", model.ImageTypePrimary) Expect(err).ToNot(HaveOccurred()) @@ -337,7 +337,7 @@ var _ = Describe("processItem", func() { }) folderRepo.result = []model.Folder{{Path: "album-a", ImageFiles: []string{"cover.jpg"}}} - outN, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "alA"}) + outN, _ := proc.acquire(ctx, model.ArtworkQueueItem{ItemKind: "al", ItemID: "alA"}) Expect(outN).To(Equal(outcomeFound)) iaA, err := artRepo.GetItemArtwork(model.KindAlbumArtwork, "alA", model.ImageTypePrimary) Expect(err).ToNot(HaveOccurred()) @@ -351,7 +351,7 @@ var _ = Describe("processItem", func() { artRepo.Data[iaA.Hash] = poisoned folderRepo.result = []model.Folder{{Path: "album-b", ImageFiles: []string{"cover.jpg"}}} - outN, _ = processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "alB"}) + outN, _ = proc.acquire(ctx, model.ArtworkQueueItem{ItemKind: "al", ItemID: "alB"}) Expect(outN).To(Equal(outcomeFound)) iaB, err := artRepo.GetItemArtwork(model.KindAlbumArtwork, "alB", model.ImageTypePrimary) Expect(err).ToNot(HaveOccurred()) @@ -383,7 +383,7 @@ var _ = Describe("processItem", func() { radioRepo.Data = map[string]*model.Radio{"ra1": {ID: "ra1", Name: "Radio", UploadedImage: "ra1_test.jpg"}} ds.MockedRadio = radioRepo - out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "ra", ItemID: "ra1"}) + out, _ := proc.acquire(ctx, model.ArtworkQueueItem{ItemKind: "ra", ItemID: "ra1"}) Expect(out).To(Equal(outcomeFailed)) _, err := artRepo.GetItemArtwork(model.KindRadioArtwork, "ra1", model.ImageTypePrimary) @@ -404,7 +404,7 @@ var _ = Describe("processItem", func() { radioRepo.Data = map[string]*model.Radio{"big": {ID: "big", Name: "Radio", UploadedImage: "big_test.jpg"}} ds.MockedRadio = radioRepo - out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "ra", ItemID: "big"}) + out, _ := proc.acquire(ctx, model.ArtworkQueueItem{ItemKind: "ra", ItemID: "big"}) Expect(out).To(Equal(outcomeFailed)) _, err = artRepo.GetItemArtwork(model.KindRadioArtwork, "big", model.ImageTypePrimary) @@ -441,9 +441,9 @@ var _ = Describe("processItem", func() { // A store root that is a plain file makes every MkdirAll under it fail. blockedRoot := filepath.Join(GinkgoT().TempDir(), "not-a-dir") Expect(os.WriteFile(blockedRoot, []byte("x"), 0600)).To(Succeed()) - deps.store = NewImageStore(blockedRoot) + proc.store = NewImageStore(blockedRoot) - out, _ := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al7"}) + out, _ := proc.acquire(ctx, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al7"}) Expect(out).To(Equal(outcomeFailed)) _, err := artRepo.GetItemArtwork(model.KindAlbumArtwork, "al7", model.ImageTypePrimary) @@ -452,7 +452,7 @@ var _ = Describe("processItem", func() { }) // countingLocker stands in for the worker's prune read-lock, recording how often and how long -// processItem holds it. +// acquire holds it. type countingLocker struct { locks int unlocks int diff --git a/core/artwork/resolve.go b/core/artwork/resolve.go index a621931d3..844232fa9 100644 --- a/core/artwork/resolve.go +++ b/core/artwork/resolve.go @@ -462,7 +462,7 @@ func mtimeViaFS(fsys fs.FS, name string) int64 { // decodeTile and assembleTiles mirror playlistArtworkReader's createTile/ // createTiledImage, reusing the same rect/fillCenter cropping helpers. -// decodeTile runs on every sampled album's resolved bytes before processItem's +// decodeTile runs on every sampled album's resolved bytes before the processor's // own maxImageBytes/maxImagePixels guards apply, so it enforces them itself too. func decodeTile(r io.ReadCloser) (image.Image, error) { data, err := readCapped(r) diff --git a/core/artwork/resolve_test.go b/core/artwork/resolve_test.go index 6c866aa5c..1459d7da8 100644 --- a/core/artwork/resolve_test.go +++ b/core/artwork/resolve_test.go @@ -620,7 +620,7 @@ var _ = Describe("resolveItem", func() { }) }) -// decodeTile runs on every sampled album's resolved bytes before processItem's +// decodeTile runs on every sampled album's resolved bytes before the processor's // own guards apply, so it must enforce the same caps independently. var _ = Describe("decodeTile", func() { It("rejects a decompression bomb before the full decode", func() { diff --git a/core/artwork/worker.go b/core/artwork/worker.go index 6881ac1a1..d113a6fe0 100644 --- a/core/artwork/worker.go +++ b/core/artwork/worker.go @@ -49,20 +49,11 @@ type drainPool struct { wake chan struct{} } -// workerDeps are the collaborators processItem needs. The resolver is built once by NewWorker -// rather than per item; pruneLock is nil only in tests, where nothing prunes. -type workerDeps struct { - ds model.DataStore - store *ImageStore - resolver *resolver - pruneLock sync.Locker -} - -// Worker drains the artwork queue through processItem: each external agent is rate-limited +// Worker drains the artwork queue through the processor: each external agent is rate-limited // and circuit-broken independently, and prune is serialized against the store-write window // via pruneMu. type Worker struct { - deps workerDeps + proc *processor cache cache.FileCache ffmpeg ffmpeg.FFmpeg broker events.Broker @@ -76,7 +67,7 @@ type Worker 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}, + proc: &processor{ds: ds, store: store}, cache: imgCache, ffmpeg: ffmpeg, broker: broker, @@ -84,8 +75,8 @@ func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg runCtx: context.Background(), gates: map[string]*extGate{}, } - w.deps.resolver = newResolver(ds, ag, ffmpeg, w.gate) - w.deps.pruneLock = w.pruneMu.RLocker() + w.proc.resolver = newResolver(ds, ag, ffmpeg, w.gate) + w.proc.pruneLock = w.pruneMu.RLocker() return w } @@ -159,7 +150,7 @@ func (w *Worker) Bump(kind, id string) { ImageType: model.ImageTypePrimary, Priority: model.ArtworkPriorityBump, } - if err := w.deps.ds.ArtworkQueue(context.Background()).Enqueue(item); err != nil { + if err := w.proc.ds.ArtworkQueue(context.Background()).Enqueue(item); err != nil { log.Warn("Artwork: Could not bump queue item", "kind", kind, "id", id, err) return } @@ -178,14 +169,14 @@ func (w *Worker) Bump(kind, id string) { func (w *Worker) RunPrune(ctx context.Context) error { w.pruneMu.Lock() defer w.pruneMu.Unlock() - return prune(ctx, w.deps.ds, w.deps.store) + return prune(ctx, w.proc.ds, w.proc.store) } 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. - items, err := w.deps.ds.ArtworkQueue(ctx).DequeueBatch(max(16, 4*concurrency), kinds...) + items, err := w.proc.ds.ArtworkQueue(ctx).DequeueBatch(max(16, 4*concurrency), kinds...) if err != nil { return 0, err } @@ -194,7 +185,7 @@ func (w *Worker) drain(ctx context.Context, concurrency int, kinds ...string) (i } // Resolved only once there is work, and per drain rather than per item: the worker needs an // admin identity for private playlists, and can start before any admin exists. - ctx = auth.WithAdminUser(ctx, w.deps.ds) + ctx = auth.WithAdminUser(ctx, w.proc.ds) sem := make(chan struct{}, concurrency) var wg sync.WaitGroup var refreshMu sync.Mutex @@ -266,9 +257,9 @@ func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) (outc if item.ImageType == "" { item.ImageType = model.ImageTypePrimary } - out, got := processItem(ctx, &w.deps, item) + out, got := w.proc.acquire(ctx, item) - queue := w.deps.ds.ArtworkQueue(ctx) + queue := w.proc.ds.ArtworkQueue(ctx) switch out { case outcomeFound, outcomeAbsent: // DeleteIfUnchanged, not Delete: a scan that re-enqueued this row mid-flight reset @@ -291,7 +282,7 @@ func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) (outc // being served is kept, since exhaustion means the source stayed unreachable rather // than that the entity lost its cover. if out == outcomeFailed && hasRecheckPath(item.ItemKind) && !w.hasResolvedArtwork(ctx, item) { - writeAbsent(ctx, w.deps.ds.Artwork(ctx), item) + writeAbsent(ctx, w.proc.ds.Artwork(ctx), item) } if err := queue.DeleteIfUnchanged(item.ItemKind, item.ItemID, item.ImageType, item.RetryAt); err != nil { log.Warn(ctx, "Artwork: Could not remove exhausted queue item", "kind", item.ItemKind, "id", item.ItemID, err) @@ -306,7 +297,7 @@ func (w *Worker) hasResolvedArtwork(ctx context.Context, item model.ArtworkQueue if !ok { return false } - ia, err := w.deps.ds.Artwork(ctx).GetItemArtwork(kind, item.ItemID, item.ImageType) + ia, err := w.proc.ds.Artwork(ctx).GetItemArtwork(kind, item.ItemID, item.ImageType) return err == nil && ia.Hash != "" } @@ -336,7 +327,7 @@ func (w *Worker) precache(ctx context.Context, got *acquired) { } // 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. +// breaker, matching gateFunc so it can be handed to the processor's resolver. func (w *Worker) gate(name string, f func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) { g := w.gateFor(name) if !g.breaker.allow() { diff --git a/core/artwork/worker_soak_test.go b/core/artwork/worker_soak_test.go index 5d06a19a0..e905664a4 100644 --- a/core/artwork/worker_soak_test.go +++ b/core/artwork/worker_soak_test.go @@ -22,7 +22,7 @@ import ( const soakCycles = 2200 var _ = Describe("Worker soak", func() { - // Runs processItem over many cycles across a mix of sources, asserting + // Runs acquisition over many cycles across a mix of sources, asserting // goroutines/heap plateau instead of growing unbounded (a leak guard). Skipped under -short. It("does not leak goroutines, heap, or fds over many acquisition cycles", func() { if testing.Short() { @@ -54,7 +54,7 @@ var _ = Describe("Worker soak", func() { MockedAlbum: albumRepo, } store := NewImageStore(GinkgoT().TempDir()) - deps := &workerDeps{ds: ds, store: store, resolver: newResolver(ds, ag, ffm, nil)} + proc := &processor{ds: ds, store: store, resolver: newResolver(ds, ag, ffm, nil)} conf.Server.CoverArtPriority = "cover.jpg, embedded" // Dangling refs (al/ra ids the repos don't know about) mirror an entity @@ -100,7 +100,7 @@ var _ = Describe("Worker soak", func() { start := time.Now() for i := range soakCycles { it := items[i%len(items)] - out, _ := processItem(context.Background(), deps, it) + out, _ := proc.acquire(context.Background(), it) // "Serve-adjacent" read-back: exercise the Phase 2 surfaces a caller would // use after acquisition, not the old serving pipeline.