mirror of
https://github.com/navidrome/navidrome.git
synced 2026-08-01 07:21:17 +00:00
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.
This commit is contained in:
Deluan
parent
f893690cb6
commit
eecb7434d3
@ -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)
|
||||
}
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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() {
|
||||
|
||||
@ -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() {
|
||||
|
||||
@ -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.
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user