From eecb7434d3b5eb34931f98e573e9b62f0d7ed836 Mon Sep 17 00:00:00 2001 From: Deluan Date: Sun, 26 Jul 2026 21:38:59 -0400 Subject: [PATCH] refactor(artwork): make acquisition a processor with its own receiver workerDeps was a parameter bag threaded into two free functions that nothing outside the worker calls. It becomes the processor type, with processItem and persist as acquire and persist methods on it, and Worker holds one collaborator instead of reaching through a bag. Kept as a separate type rather than folding onto Worker: acquisition takes a queue item and returns bytes, while Worker.process settles the queue row around it. That boundary is what keeps retry policy out of the image pipeline, and what lets the acquisition specs build a three-field value instead of a Worker with drain pools, gates, a broker and a real on-disk cache. --- core/artwork/processor.go | 34 ++++++++++++++-------- core/artwork/processor_test.go | 48 ++++++++++++++++---------------- core/artwork/resolve.go | 2 +- core/artwork/resolve_test.go | 2 +- core/artwork/worker.go | 37 ++++++++++-------------- core/artwork/worker_soak_test.go | 6 ++-- 6 files changed, 65 insertions(+), 64 deletions(-) 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.