mirror of
https://github.com/navidrome/navidrome.git
synced 2026-08-01 07:21:17 +00:00
feat(artwork): worker fetches agent images directly with per-agent rate limits and breakers
This commit is contained in:
Deluan
parent
b7f94f6727
commit
190c291e61
@ -244,10 +244,8 @@ func CreateArtworkWorker() *artwork.Worker {
|
||||
metricsMetrics := metrics.GetPrometheusInstance(dataStore)
|
||||
manager := plugins.GetManager(dataStore, broker, metricsMetrics)
|
||||
agentsAgents := agents.GetAgents(dataStore, manager)
|
||||
matcherMatcher := matcher.New(dataStore)
|
||||
provider := external.NewProvider(dataStore, agentsAgents, matcherMatcher)
|
||||
fFmpeg := ffmpeg.New()
|
||||
worker := artwork.NewWorker(dataStore, imageStore, provider, fFmpeg)
|
||||
worker := artwork.NewWorker(dataStore, imageStore, agentsAgents, fFmpeg)
|
||||
return worker
|
||||
}
|
||||
|
||||
|
||||
97
core/artwork/agent_images.go
Normal file
97
core/artwork/agent_images.go
Normal file
@ -0,0 +1,97 @@
|
||||
package artwork
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net/url"
|
||||
|
||||
"github.com/navidrome/navidrome/core/agents"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
)
|
||||
|
||||
// gateFunc gates one named external fetch (rate limit + circuit breaker per name).
|
||||
// resolveItem defaults to passthroughGate; the worker injects the per-agent gate.
|
||||
type gateFunc = func(name string, f func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error)
|
||||
|
||||
func passthroughGate(_ string, f func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) {
|
||||
return f()
|
||||
}
|
||||
|
||||
// bestImageURL returns the largest-Size image URL, skipping empty or unparseable
|
||||
// URLs; nil when none qualifies.
|
||||
func bestImageURL(imgs []agents.ExternalImage) *url.URL {
|
||||
var best *agents.ExternalImage
|
||||
for i := range imgs {
|
||||
if imgs[i].URL == "" {
|
||||
continue
|
||||
}
|
||||
if best == nil || imgs[i].Size > best.Size {
|
||||
best = &imgs[i]
|
||||
}
|
||||
}
|
||||
if best == nil {
|
||||
return nil
|
||||
}
|
||||
u, err := url.Parse(best.URL)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
return u
|
||||
}
|
||||
|
||||
// fetchArtistImage tries each enabled artist-image agent in order, each under its own gate.
|
||||
// Returns the winning reader + agent name; extErr is true only when NO agent succeeded and
|
||||
// at least one failed transiently (a later success beats an earlier agent error).
|
||||
func fetchArtistImage(ctx context.Context, ag *agents.Agents, gate gateFunc, ar model.Artist) (r io.ReadCloser, agentName string, extErr bool) {
|
||||
for _, a := range ag.ArtistImageAgents() {
|
||||
reader, _, err := gate(a.Name, func() (io.ReadCloser, string, error) {
|
||||
imgs, err := a.Retriever.GetArtistImages(ctx, ar.ID, ar.Name, ar.MbzArtistID)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
u := bestImageURL(imgs)
|
||||
if u == nil {
|
||||
return nil, "", agents.ErrNotFound
|
||||
}
|
||||
return fromURL(ctx, u)
|
||||
})
|
||||
if reader != nil {
|
||||
return reader, a.Name, false
|
||||
}
|
||||
if isTransientExternal(err) {
|
||||
extErr = true // includes errBreakerOpen and download failures: retry via the next agent
|
||||
}
|
||||
}
|
||||
return nil, "", extErr
|
||||
}
|
||||
|
||||
// fetchAlbumImage is the album counterpart of fetchArtistImage.
|
||||
func fetchAlbumImage(ctx context.Context, ag *agents.Agents, gate gateFunc, al model.Album) (r io.ReadCloser, agentName string, extErr bool) {
|
||||
for _, a := range ag.AlbumImageAgents() {
|
||||
reader, _, err := gate(a.Name, func() (io.ReadCloser, string, error) {
|
||||
imgs, err := a.Retriever.GetAlbumImages(ctx, al.Name, al.AlbumArtist, al.MbzAlbumID)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
u := bestImageURL(imgs)
|
||||
if u == nil {
|
||||
return nil, "", agents.ErrNotFound
|
||||
}
|
||||
return fromURL(ctx, u)
|
||||
})
|
||||
if reader != nil {
|
||||
return reader, a.Name, false
|
||||
}
|
||||
if isTransientExternal(err) {
|
||||
extErr = true
|
||||
}
|
||||
}
|
||||
return nil, "", extErr
|
||||
}
|
||||
|
||||
// isTransientExternal reports whether an external step failed in a way worth retrying;
|
||||
// a not-found (from either package) is a definitive answer, not a fault.
|
||||
func isTransientExternal(err error) bool {
|
||||
return err != nil && !errors.Is(err, agents.ErrNotFound) && !errors.Is(err, model.ErrNotFound)
|
||||
}
|
||||
180
core/artwork/agent_images_test.go
Normal file
180
core/artwork/agent_images_test.go
Normal file
@ -0,0 +1,180 @@
|
||||
package artwork
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
|
||||
"github.com/navidrome/navidrome/conf"
|
||||
"github.com/navidrome/navidrome/conf/configtest"
|
||||
"github.com/navidrome/navidrome/core/agents"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
"github.com/navidrome/navidrome/tests"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
// fakeImageAgent is a built-in agent stub implementing both image retrievers; it
|
||||
// records call counts so per-agent ordering and short-circuiting can be asserted.
|
||||
type fakeImageAgent struct {
|
||||
name string
|
||||
imgs []agents.ExternalImage
|
||||
err error
|
||||
artistCalls int
|
||||
albumCalls int
|
||||
}
|
||||
|
||||
func (f *fakeImageAgent) AgentName() string { return f.name }
|
||||
|
||||
func (f *fakeImageAgent) GetArtistImages(context.Context, string, string, string) ([]agents.ExternalImage, error) {
|
||||
f.artistCalls++
|
||||
return f.imgs, f.err
|
||||
}
|
||||
|
||||
func (f *fakeImageAgent) GetAlbumImages(context.Context, string, string, string) ([]agents.ExternalImage, error) {
|
||||
f.albumCalls++
|
||||
return f.imgs, f.err
|
||||
}
|
||||
|
||||
// imageAgents registers the fakes as built-in agents (ignoring the DataStore) and
|
||||
// enables them in order, returning the process-wide Agents. Because the fakes ignore
|
||||
// ds, reusing the GetAgents singleton across tests is safe.
|
||||
func imageAgents(fakes ...*fakeImageAgent) *agents.Agents {
|
||||
names := make([]string, 0, len(fakes))
|
||||
for _, f := range fakes {
|
||||
fake := f
|
||||
agents.Register(fake.name, func(model.DataStore) agents.Interface { return fake })
|
||||
names = append(names, fake.name)
|
||||
}
|
||||
conf.Server.Agents = strings.Join(names, ",")
|
||||
return agents.GetAgents(&tests.MockDataStore{}, nil)
|
||||
}
|
||||
|
||||
var _ = Describe("agent images", func() {
|
||||
var (
|
||||
ctx context.Context
|
||||
srv *httptest.Server
|
||||
)
|
||||
|
||||
BeforeEach(func() {
|
||||
DeferCleanup(configtest.SetupConfig())
|
||||
ctx = context.Background()
|
||||
srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
_, _ = w.Write([]byte("image-bytes"))
|
||||
}))
|
||||
DeferCleanup(srv.Close)
|
||||
})
|
||||
|
||||
img := func(path string, size int) agents.ExternalImage {
|
||||
return agents.ExternalImage{URL: srv.URL + path, Size: size}
|
||||
}
|
||||
|
||||
Describe("bestImageURL", func() {
|
||||
It("picks the largest-Size URL and skips empty ones", func() {
|
||||
u := bestImageURL([]agents.ExternalImage{
|
||||
{URL: "http://x/small", Size: 10},
|
||||
{URL: "", Size: 9999},
|
||||
{URL: "http://x/big", Size: 100},
|
||||
})
|
||||
Expect(u).ToNot(BeNil())
|
||||
Expect(u.String()).To(Equal("http://x/big"))
|
||||
})
|
||||
|
||||
It("returns nil when there is no non-empty URL", func() {
|
||||
Expect(bestImageURL(nil)).To(BeNil())
|
||||
Expect(bestImageURL([]agents.ExternalImage{{URL: "", Size: 5}})).To(BeNil())
|
||||
})
|
||||
})
|
||||
|
||||
Describe("fetchArtistImage", func() {
|
||||
It("returns the first agent's image and its name", func() {
|
||||
a := &fakeImageAgent{name: "agentA", imgs: []agents.ExternalImage{img("/a", 100)}}
|
||||
ag := imageAgents(a)
|
||||
|
||||
r, name, extErr := fetchArtistImage(ctx, ag, passthroughGate, model.Artist{ID: "ar1", Name: "Artist"})
|
||||
Expect(r).ToNot(BeNil())
|
||||
defer r.Close()
|
||||
Expect(name).To(Equal("agentA"))
|
||||
Expect(extErr).To(BeFalse())
|
||||
})
|
||||
|
||||
It("falls through to a later agent, and its success beats the earlier error", func() {
|
||||
a := &fakeImageAgent{name: "agentA", err: errBreakerOpen}
|
||||
b := &fakeImageAgent{name: "agentB", imgs: []agents.ExternalImage{img("/b", 50)}}
|
||||
ag := imageAgents(a, b)
|
||||
|
||||
r, name, extErr := fetchArtistImage(ctx, ag, passthroughGate, model.Artist{ID: "ar1"})
|
||||
Expect(r).ToNot(BeNil())
|
||||
defer r.Close()
|
||||
Expect(name).To(Equal("agentB"))
|
||||
Expect(extErr).To(BeFalse(), "a later hit clears an earlier agent's error")
|
||||
Expect(a.artistCalls).To(Equal(1))
|
||||
Expect(b.artistCalls).To(Equal(1))
|
||||
})
|
||||
|
||||
It("reports a clean miss when every agent finds nothing", func() {
|
||||
a := &fakeImageAgent{name: "agentA"} // no images, no error -> not found
|
||||
b := &fakeImageAgent{name: "agentB", err: agents.ErrNotFound}
|
||||
ag := imageAgents(a, b)
|
||||
|
||||
r, name, extErr := fetchArtistImage(ctx, ag, passthroughGate, model.Artist{ID: "ar1"})
|
||||
Expect(r).To(BeNil())
|
||||
Expect(name).To(BeEmpty())
|
||||
Expect(extErr).To(BeFalse(), "not-found is definitive, never a transient failure")
|
||||
})
|
||||
|
||||
It("reports extErr when one agent fails transiently and the rest find nothing", func() {
|
||||
a := &fakeImageAgent{name: "agentA", err: agents.ErrNotFound}
|
||||
b := &fakeImageAgent{name: "agentB", err: context.DeadlineExceeded}
|
||||
ag := imageAgents(a, b)
|
||||
|
||||
r, _, extErr := fetchArtistImage(ctx, ag, passthroughGate, model.Artist{ID: "ar1"})
|
||||
Expect(r).To(BeNil())
|
||||
Expect(extErr).To(BeTrue())
|
||||
})
|
||||
})
|
||||
|
||||
Describe("fetchAlbumImage", func() {
|
||||
It("returns the winning agent's image and name", func() {
|
||||
a := &fakeImageAgent{name: "agentA", imgs: []agents.ExternalImage{img("/a", 100)}}
|
||||
ag := imageAgents(a)
|
||||
|
||||
r, name, extErr := fetchAlbumImage(ctx, ag, passthroughGate, model.Album{Name: "Album", AlbumArtist: "Artist"})
|
||||
Expect(r).ToNot(BeNil())
|
||||
defer r.Close()
|
||||
Expect(name).To(Equal("agentA"))
|
||||
Expect(extErr).To(BeFalse())
|
||||
Expect(a.albumCalls).To(Equal(1))
|
||||
})
|
||||
|
||||
It("reports extErr when the only agent fails transiently", func() {
|
||||
a := &fakeImageAgent{name: "agentA", err: context.DeadlineExceeded}
|
||||
ag := imageAgents(a)
|
||||
|
||||
r, _, extErr := fetchAlbumImage(ctx, ag, passthroughGate, model.Album{Name: "Album"})
|
||||
Expect(r).To(BeNil())
|
||||
Expect(extErr).To(BeTrue())
|
||||
})
|
||||
})
|
||||
|
||||
Describe("gate naming", func() {
|
||||
It("invokes the gate once per agent, keyed by agent name", func() {
|
||||
a := &fakeImageAgent{name: "agentA", err: context.DeadlineExceeded}
|
||||
b := &fakeImageAgent{name: "agentB", imgs: []agents.ExternalImage{img("/b", 1)}}
|
||||
ag := imageAgents(a, b)
|
||||
|
||||
var gatedNames []string
|
||||
gate := func(name string, f func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) {
|
||||
gatedNames = append(gatedNames, name)
|
||||
return f()
|
||||
}
|
||||
|
||||
r, _, _ := fetchArtistImage(ctx, ag, gate, model.Artist{ID: "ar1"})
|
||||
Expect(r).ToNot(BeNil())
|
||||
defer r.Close()
|
||||
Expect(gatedNames).To(Equal([]string{"agentA", "agentB"}))
|
||||
})
|
||||
})
|
||||
})
|
||||
@ -10,8 +10,8 @@ import (
|
||||
"io"
|
||||
"time"
|
||||
|
||||
"github.com/navidrome/navidrome/core/agents"
|
||||
"github.com/navidrome/navidrome/core/artwork/blurhash"
|
||||
"github.com/navidrome/navidrome/core/external"
|
||||
"github.com/navidrome/navidrome/core/ffmpeg"
|
||||
"github.com/navidrome/navidrome/log"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
@ -42,14 +42,14 @@ 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; extGate is set by NewWorker in
|
||||
// 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.
|
||||
type workerDeps struct {
|
||||
ds model.DataStore
|
||||
store *ImageStore
|
||||
prov external.Provider
|
||||
ffmpeg ffmpeg.FFmpeg
|
||||
extGate extGateFunc
|
||||
ds model.DataStore
|
||||
store *ImageStore
|
||||
agents *agents.Agents
|
||||
ffmpeg ffmpeg.FFmpeg
|
||||
gate gateFunc
|
||||
}
|
||||
|
||||
// processItem resolves one queue item end to end: find an image, hash/decode/
|
||||
@ -57,7 +57,7 @@ type workerDeps struct {
|
||||
func processItem(ctx context.Context, deps *workerDeps, item model.ArtworkQueueItem) outcome {
|
||||
repo := deps.ds.Artwork(ctx)
|
||||
|
||||
res, err := resolveItem(ctx, deps.ds, deps.prov, deps.ffmpeg, item, deps.extGate)
|
||||
res, err := resolveItem(ctx, deps.ds, deps.agents, deps.ffmpeg, item, deps.gate)
|
||||
if err != nil {
|
||||
log.Warn(ctx, "artwork: could not resolve item", "kind", item.ItemKind, "id", item.ItemID, err)
|
||||
return outcomeFailed
|
||||
|
||||
@ -5,13 +5,15 @@ import (
|
||||
"encoding/binary"
|
||||
"errors"
|
||||
"hash/crc32"
|
||||
"net/url"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"github.com/navidrome/navidrome/conf"
|
||||
"github.com/navidrome/navidrome/conf/configtest"
|
||||
"github.com/navidrome/navidrome/core/agents"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
"github.com/navidrome/navidrome/tests"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
@ -40,7 +42,7 @@ var _ = Describe("processItem", func() {
|
||||
folderRepo *fakeFolderRepo
|
||||
libRepo *tests.MockLibraryRepo
|
||||
ffm *tests.MockFFmpeg
|
||||
prov *fakeExternalProvider
|
||||
ag *agents.Agents
|
||||
store *ImageStore
|
||||
artRepo *tests.MockArtworkRepo
|
||||
repoRoot string
|
||||
@ -58,7 +60,7 @@ var _ = Describe("processItem", func() {
|
||||
libRepo = &tests.MockLibraryRepo{}
|
||||
libRepo.SetData(model.Libraries{{ID: 0, Path: testFileLibPath(repoRoot)}})
|
||||
ffm = tests.NewMockFFmpeg("")
|
||||
prov = &fakeExternalProvider{}
|
||||
ag = agents.GetAgents(&tests.MockDataStore{}, nil)
|
||||
artRepo = tests.CreateMockArtworkRepo()
|
||||
ds = &tests.MockDataStore{
|
||||
MockedFolder: folderRepo,
|
||||
@ -67,7 +69,7 @@ var _ = Describe("processItem", func() {
|
||||
}
|
||||
ds.MockedAlbum = tests.CreateMockAlbumRepo()
|
||||
store = NewImageStore(GinkgoT().TempDir())
|
||||
deps = &workerDeps{ds: ds, store: store, prov: prov, ffmpeg: ffm}
|
||||
deps = &workerDeps{ds: ds, store: store, agents: ag, ffmpeg: ffm}
|
||||
|
||||
conf.Server.CoverArtPriority = "cover.jpg, embedded"
|
||||
})
|
||||
@ -141,9 +143,7 @@ var _ = Describe("processItem", func() {
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
|
||||
{ID: "al4", Name: "Album"},
|
||||
})
|
||||
prov.albumImage = func(context.Context, string) (*url.URL, error) {
|
||||
return nil, errors.New("agent timed out")
|
||||
}
|
||||
imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("agent timed out")})
|
||||
|
||||
out := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al4"})
|
||||
Expect(out).To(Equal(outcomeFailed))
|
||||
@ -161,9 +161,7 @@ var _ = Describe("processItem", func() {
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
|
||||
{ID: "alstale", Name: "Album", FolderIDs: []string{"f1"}},
|
||||
})
|
||||
prov.albumImage = func(context.Context, string) (*url.URL, error) {
|
||||
return nil, errors.New("agent timed out")
|
||||
}
|
||||
imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("agent timed out")})
|
||||
|
||||
out := processItem(ctx, deps, model.ArtworkQueueItem{ItemKind: "al", ItemID: "alstale"})
|
||||
Expect(out).To(Equal(outcomeFoundStale))
|
||||
@ -174,6 +172,34 @@ var _ = Describe("processItem", func() {
|
||||
Expect(ia.Source).To(Equal("folder"))
|
||||
})
|
||||
|
||||
It("found-external: persists source as external:<agentName> and stores the fetched bytes", func() {
|
||||
conf.Server.CoverArtPriority = "external"
|
||||
imgBytes, err := os.ReadFile(filepath.Join(repoRoot, "tests/fixtures/artist/an-album/cover.jpg"))
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
_, _ = w.Write(imgBytes)
|
||||
}))
|
||||
DeferCleanup(srv.Close)
|
||||
|
||||
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"})
|
||||
Expect(out).To(Equal(outcomeFound))
|
||||
|
||||
ia, err := artRepo.GetItemArtwork("al", "alext", model.ImageTypePrimary)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(ia.Source).To(Equal("external:deezerFake"))
|
||||
Expect(ia.Hash).ToNot(BeEmpty())
|
||||
|
||||
// External art is content-addressed into the store, not file-backed.
|
||||
art, err := artRepo.GetImage(ia.Hash)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
rc, err := store.Open(ia.Hash, art.Mime)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
rc.Close()
|
||||
})
|
||||
|
||||
It("dedup: a second item with identical bytes skips decode and reuses the artwork row", func() {
|
||||
folderRepo.result = []model.Folder{{
|
||||
Path: "tests/fixtures/artist/an-album",
|
||||
|
||||
@ -19,7 +19,7 @@ import (
|
||||
"github.com/Masterminds/squirrel"
|
||||
"github.com/navidrome/navidrome/conf"
|
||||
"github.com/navidrome/navidrome/consts"
|
||||
"github.com/navidrome/navidrome/core/external"
|
||||
"github.com/navidrome/navidrome/core/agents"
|
||||
"github.com/navidrome/navidrome/core/ffmpeg"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
)
|
||||
@ -35,26 +35,18 @@ type resolution struct {
|
||||
extError bool
|
||||
}
|
||||
|
||||
// extGateFunc is an alias for the external-step wrapper the worker injects (rate
|
||||
// limiter + circuit breaker); resolveItem defaults to a plain passthrough.
|
||||
type extGateFunc = func(func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error)
|
||||
|
||||
func passthroughExtGate(f func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) {
|
||||
return f()
|
||||
}
|
||||
|
||||
// resolveItem walks the kind's priority chain and returns the first hit.
|
||||
func resolveItem(ctx context.Context, ds model.DataStore, prov external.Provider, ffmpeg ffmpeg.FFmpeg, item model.ArtworkQueueItem, extGate extGateFunc) (resolution, error) {
|
||||
if extGate == nil {
|
||||
extGate = passthroughExtGate
|
||||
func resolveItem(ctx context.Context, ds model.DataStore, ag *agents.Agents, ffmpeg ffmpeg.FFmpeg, item model.ArtworkQueueItem, gate gateFunc) (resolution, error) {
|
||||
if gate == nil {
|
||||
gate = passthroughGate
|
||||
}
|
||||
switch item.ItemKind {
|
||||
case "al":
|
||||
return resolveAlbum(ctx, ds, prov, ffmpeg, item.ItemID, extGate)
|
||||
return resolveAlbum(ctx, ds, ag, ffmpeg, item.ItemID, gate)
|
||||
case "ar":
|
||||
return resolveArtist(ctx, ds, prov, ffmpeg, item.ItemID, extGate)
|
||||
return resolveArtist(ctx, ds, ag, ffmpeg, item.ItemID, gate)
|
||||
case "pl":
|
||||
return resolvePlaylist(ctx, ds, prov, ffmpeg, item.ItemID, extGate)
|
||||
return resolvePlaylist(ctx, ds, ag, ffmpeg, item.ItemID, gate)
|
||||
case "ra":
|
||||
return resolveRadio(ctx, ds, item.ItemID)
|
||||
default:
|
||||
@ -64,7 +56,7 @@ func resolveItem(ctx context.Context, ds model.DataStore, prov external.Provider
|
||||
|
||||
// resolveAlbum ports the folder/embedded/external selection from
|
||||
// reader_album.go, walking conf.Server.CoverArtPriority.
|
||||
func resolveAlbum(ctx context.Context, ds model.DataStore, prov external.Provider, ffm ffmpeg.FFmpeg, albumID string, extGate extGateFunc) (resolution, error) {
|
||||
func resolveAlbum(ctx context.Context, ds model.DataStore, ag *agents.Agents, ffm ffmpeg.FFmpeg, albumID string, gate gateFunc) (resolution, error) {
|
||||
al, err := ds.Album(ctx).Get(albumID)
|
||||
if err != nil {
|
||||
return resolution{}, err
|
||||
@ -88,8 +80,8 @@ func resolveAlbum(ctx context.Context, ds model.DataStore, prov external.Provide
|
||||
return res, nil
|
||||
}
|
||||
case pattern == "external":
|
||||
if res, ok, isErr := resolveExternalStep(extGate, fromAlbumExternalSource(ctx, *al, prov)); ok {
|
||||
return res, nil
|
||||
if r, name, isErr := fetchAlbumImage(ctx, ag, gate, *al); r != nil {
|
||||
return resolution{reader: r, source: "external:" + name}, nil
|
||||
} else if isErr {
|
||||
extErr = true
|
||||
}
|
||||
@ -105,7 +97,7 @@ func resolveAlbum(ctx context.Context, ds model.DataStore, prov external.Provide
|
||||
|
||||
// resolveArtist ports the upload/folder/external selection from
|
||||
// reader_artist.go: upload always wins, then conf.Server.ArtistArtPriority.
|
||||
func resolveArtist(ctx context.Context, ds model.DataStore, prov external.Provider, ffm ffmpeg.FFmpeg, artistID string, extGate extGateFunc) (resolution, error) {
|
||||
func resolveArtist(ctx context.Context, ds model.DataStore, ag *agents.Agents, ffm ffmpeg.FFmpeg, artistID string, gate gateFunc) (resolution, error) {
|
||||
ar, err := ds.Artist(ctx).Get(artistID)
|
||||
if err != nil {
|
||||
return resolution{}, err
|
||||
@ -145,8 +137,8 @@ func resolveArtist(ctx context.Context, ds model.DataStore, prov external.Provid
|
||||
pattern = strings.TrimSpace(pattern)
|
||||
switch {
|
||||
case pattern == "external":
|
||||
if res, ok, isErr := resolveExternalStep(extGate, fromArtistExternalResult(ctx, *ar, prov)); ok {
|
||||
return res, nil
|
||||
if r, name, isErr := fetchArtistImage(ctx, ag, gate, *ar); r != nil {
|
||||
return resolution{reader: r, source: "external:" + name}, nil
|
||||
} else if isErr {
|
||||
extErr = true
|
||||
}
|
||||
@ -178,7 +170,7 @@ func resolveArtist(ctx context.Context, ds model.DataStore, prov external.Provid
|
||||
|
||||
// resolvePlaylist ports reader_playlist.go's chain: uploaded image, sidecar,
|
||||
// ExternalImageURL, then the generated 2x2 grid sourced through resolveAlbum.
|
||||
func resolvePlaylist(ctx context.Context, ds model.DataStore, prov external.Provider, ffm ffmpeg.FFmpeg, playlistID string, extGate extGateFunc) (resolution, error) {
|
||||
func resolvePlaylist(ctx context.Context, ds model.DataStore, ag *agents.Agents, ffm ffmpeg.FFmpeg, playlistID string, gate gateFunc) (resolution, error) {
|
||||
pl, err := ds.Playlist(ctx).Get(playlistID)
|
||||
if err != nil {
|
||||
return resolution{}, err
|
||||
@ -191,7 +183,7 @@ func resolvePlaylist(ctx context.Context, ds model.DataStore, prov external.Prov
|
||||
if res, ok := resolveLocalFile(findPlaylistSidecarPath(ctx, pl.Path), "folder"); ok {
|
||||
return res, nil
|
||||
}
|
||||
if res, ok, isErr := resolveExternalStep(extGate, fromPlaylistExternalSource(ctx, *pl)); ok {
|
||||
if res, ok, isErr := resolveExternalStep(gate, "m3u", fromPlaylistExternalSource(ctx, *pl)); ok {
|
||||
return res, nil
|
||||
} else if isErr {
|
||||
extErr = true
|
||||
@ -205,7 +197,7 @@ func resolvePlaylist(ctx context.Context, ds model.DataStore, prov external.Prov
|
||||
var tiles []image.Image
|
||||
var tileErr error // first internal (non-external) tile failure, e.g. album deleted mid-flight
|
||||
for _, albumID := range albumIDs {
|
||||
res, err := resolveAlbum(ctx, ds, prov, ffm, albumID, extGate)
|
||||
res, err := resolveAlbum(ctx, ds, ag, ffm, albumID, gate)
|
||||
if err != nil {
|
||||
if tileErr == nil {
|
||||
tileErr = err
|
||||
@ -259,11 +251,11 @@ func resolveRadio(ctx context.Context, ds model.DataStore, radioID string) (reso
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// resolveExternalStep runs an external sourceFunc through extGate, shared by
|
||||
// resolveAlbum and resolveArtist. ok reports a hit; extErr reports a
|
||||
// non-not-found error (a not-found is a definitive "no", not a failure).
|
||||
func resolveExternalStep(extGate extGateFunc, sf func() (io.ReadCloser, string, error)) (res resolution, ok bool, extErr bool) {
|
||||
r, path, err := extGate(sf)
|
||||
// resolveExternalStep runs a single external sourceFunc through the named gate; used by
|
||||
// the playlist ExternalImageURL step. ok reports a hit; extErr reports a non-not-found
|
||||
// error (a not-found is a definitive "no", not a failure).
|
||||
func resolveExternalStep(gate gateFunc, name string, sf sourceFunc) (res resolution, ok bool, extErr bool) {
|
||||
r, path, err := gate(name, sf)
|
||||
if r != nil {
|
||||
return resolution{reader: r, source: "external", sourcePath: path}, true, false
|
||||
}
|
||||
|
||||
@ -8,45 +8,18 @@ import (
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
"github.com/navidrome/navidrome/conf"
|
||||
"github.com/navidrome/navidrome/conf/configtest"
|
||||
"github.com/navidrome/navidrome/core/external"
|
||||
"github.com/navidrome/navidrome/core/agents"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
"github.com/navidrome/navidrome/tests"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
// fakeExternalProvider is a minimal external.Provider stub for resolve_test.go;
|
||||
// only AlbumImage/ArtistImage are exercised by the resolvers.
|
||||
type fakeExternalProvider struct {
|
||||
external.Provider
|
||||
albumImage func(ctx context.Context, id string) (*url.URL, error)
|
||||
artistImage func(ctx context.Context, id string) (*url.URL, error)
|
||||
}
|
||||
|
||||
func (f *fakeExternalProvider) AlbumImage(ctx context.Context, id string) (*url.URL, error) {
|
||||
if f.albumImage != nil {
|
||||
return f.albumImage(ctx, id)
|
||||
}
|
||||
return nil, model.ErrNotFound
|
||||
}
|
||||
|
||||
func (f *fakeExternalProvider) ArtistImage(ctx context.Context, id string) (*url.URL, error) {
|
||||
if f.artistImage != nil {
|
||||
return f.artistImage(ctx, id)
|
||||
}
|
||||
return nil, model.ErrNotFound
|
||||
}
|
||||
|
||||
func (f *fakeExternalProvider) ArtistImageResult(ctx context.Context, id string) (*url.URL, error) {
|
||||
return f.ArtistImage(ctx, id)
|
||||
}
|
||||
|
||||
var _ = Describe("resolveItem", func() {
|
||||
var (
|
||||
ctx context.Context
|
||||
@ -54,7 +27,7 @@ var _ = Describe("resolveItem", func() {
|
||||
folderRepo *fakeFolderRepo
|
||||
libRepo *tests.MockLibraryRepo
|
||||
ffm *tests.MockFFmpeg
|
||||
prov *fakeExternalProvider
|
||||
ag *agents.Agents
|
||||
repoRoot string
|
||||
)
|
||||
|
||||
@ -69,7 +42,7 @@ var _ = Describe("resolveItem", func() {
|
||||
libRepo = &tests.MockLibraryRepo{}
|
||||
libRepo.SetData(model.Libraries{{ID: 0, Path: testFileLibPath(repoRoot)}})
|
||||
ffm = tests.NewMockFFmpeg("")
|
||||
prov = &fakeExternalProvider{}
|
||||
ag = agents.GetAgents(&tests.MockDataStore{}, nil)
|
||||
ds = &tests.MockDataStore{
|
||||
MockedFolder: folderRepo,
|
||||
MockedLibrary: libRepo,
|
||||
@ -78,7 +51,7 @@ var _ = Describe("resolveItem", func() {
|
||||
|
||||
Describe("kind dispatch", func() {
|
||||
It("returns an error for kinds the worker never enqueues", func() {
|
||||
_, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "mf", ItemID: "x"}, nil)
|
||||
_, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "mf", ItemID: "x"}, nil)
|
||||
Expect(err).To(HaveOccurred())
|
||||
})
|
||||
})
|
||||
@ -98,7 +71,7 @@ var _ = Describe("resolveItem", func() {
|
||||
{ID: "al1", Name: "Album", FolderIDs: []string{"f1"}},
|
||||
})
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al1"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al1"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).ToNot(BeNil())
|
||||
defer res.reader.Close()
|
||||
@ -114,7 +87,7 @@ var _ = Describe("resolveItem", func() {
|
||||
{ID: "al2", Name: "Album", EmbedArtPath: "tests/fixtures/artist/an-album/test.mp3", FolderIDs: []string{"f1"}},
|
||||
})
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al2"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al2"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).ToNot(BeNil())
|
||||
defer res.reader.Close()
|
||||
@ -128,11 +101,9 @@ var _ = Describe("resolveItem", func() {
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
|
||||
{ID: "al3", Name: "Album"},
|
||||
})
|
||||
prov.albumImage = func(context.Context, string) (*url.URL, error) {
|
||||
return nil, errors.New("agent timed out")
|
||||
}
|
||||
imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("agent timed out")})
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al3"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al3"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).To(BeNil())
|
||||
Expect(res.extError).To(BeTrue())
|
||||
@ -143,9 +114,9 @@ var _ = Describe("resolveItem", func() {
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
|
||||
{ID: "al4", Name: "Album"},
|
||||
})
|
||||
// prov.albumImage left nil -> fakeExternalProvider returns model.ErrNotFound
|
||||
// no image agents enabled -> the external step is a definitive not-found
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al4"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al4"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).To(BeNil())
|
||||
Expect(res.extError).To(BeFalse())
|
||||
@ -160,11 +131,9 @@ var _ = Describe("resolveItem", func() {
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
|
||||
{ID: "al6", Name: "Album", FolderIDs: []string{"f1"}},
|
||||
})
|
||||
prov.albumImage = func(context.Context, string) (*url.URL, error) {
|
||||
return nil, errors.New("agent timed out")
|
||||
}
|
||||
imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("agent timed out")})
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al6"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al6"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).ToNot(BeNil())
|
||||
defer res.reader.Close()
|
||||
@ -181,9 +150,9 @@ var _ = Describe("resolveItem", func() {
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
|
||||
{ID: "al7", Name: "Album", FolderIDs: []string{"f1"}},
|
||||
})
|
||||
// prov.albumImage left nil -> fakeExternalProvider returns model.ErrNotFound
|
||||
// no image agents enabled -> the external step is a definitive not-found
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al7"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al7"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).ToNot(BeNil())
|
||||
defer res.reader.Close()
|
||||
@ -191,24 +160,22 @@ var _ = Describe("resolveItem", func() {
|
||||
Expect(res.extError).To(BeFalse())
|
||||
})
|
||||
|
||||
It("routes the external step through a custom extGate", func() {
|
||||
It("routes the external step through the injected gate, keyed by agent name", func() {
|
||||
conf.Server.CoverArtPriority = "external"
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
|
||||
{ID: "al5", Name: "Album"},
|
||||
})
|
||||
prov.albumImage = func(context.Context, string) (*url.URL, error) {
|
||||
return nil, errors.New("boom")
|
||||
}
|
||||
var extGateCalls int
|
||||
extGate := func(f func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) {
|
||||
extGateCalls++
|
||||
imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("boom")})
|
||||
var gatedNames []string
|
||||
gate := func(name string, f func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) {
|
||||
gatedNames = append(gatedNames, name)
|
||||
return f()
|
||||
}
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al5"}, extGate)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "al", ItemID: "al5"}, gate)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.extError).To(BeTrue())
|
||||
Expect(extGateCalls).To(Equal(1))
|
||||
Expect(gatedNames).To(Equal([]string{"failAgent"}))
|
||||
})
|
||||
})
|
||||
|
||||
@ -224,7 +191,7 @@ var _ = Describe("resolveItem", func() {
|
||||
artistRepo.SetData(model.Artists{{ID: "ar1", Name: "Artist", UploadedImage: "ar1_test.jpg"}})
|
||||
ds.MockedArtist = artistRepo
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "ar", ItemID: "ar1"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "ar", ItemID: "ar1"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).ToNot(BeNil())
|
||||
defer res.reader.Close()
|
||||
@ -247,7 +214,7 @@ var _ = Describe("resolveItem", func() {
|
||||
{ID: "al9", Name: "Album", LibraryID: 0, FolderIDs: []string{"f1"}},
|
||||
}
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "ar", ItemID: "ar2"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "ar", ItemID: "ar2"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).ToNot(BeNil())
|
||||
defer res.reader.Close()
|
||||
@ -260,11 +227,9 @@ var _ = Describe("resolveItem", func() {
|
||||
artistRepo := tests.CreateMockArtistRepo()
|
||||
artistRepo.SetData(model.Artists{{ID: "ar3", Name: "Artist"}})
|
||||
ds.MockedArtist = artistRepo
|
||||
prov.artistImage = func(context.Context, string) (*url.URL, error) {
|
||||
return nil, errors.New("agent timed out")
|
||||
}
|
||||
imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("agent timed out")})
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "ar", ItemID: "ar3"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "ar", ItemID: "ar3"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).To(BeNil())
|
||||
Expect(res.extError).To(BeTrue())
|
||||
@ -275,32 +240,30 @@ var _ = Describe("resolveItem", func() {
|
||||
artistRepo := tests.CreateMockArtistRepo()
|
||||
artistRepo.SetData(model.Artists{{ID: "ar4", Name: "Artist"}})
|
||||
ds.MockedArtist = artistRepo
|
||||
// prov.artistImage left nil -> fakeExternalProvider returns model.ErrNotFound
|
||||
// no image agents enabled -> the external step is a definitive not-found
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "ar", ItemID: "ar4"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "ar", ItemID: "ar4"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).To(BeNil())
|
||||
Expect(res.extError).To(BeFalse())
|
||||
})
|
||||
|
||||
It("routes the external step through a custom extGate", func() {
|
||||
It("routes the external step through the injected gate, keyed by agent name", func() {
|
||||
conf.Server.ArtistArtPriority = "external"
|
||||
artistRepo := tests.CreateMockArtistRepo()
|
||||
artistRepo.SetData(model.Artists{{ID: "ar5", Name: "Artist"}})
|
||||
ds.MockedArtist = artistRepo
|
||||
prov.artistImage = func(context.Context, string) (*url.URL, error) {
|
||||
return nil, errors.New("boom")
|
||||
}
|
||||
var extGateCalls int
|
||||
extGate := func(f func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) {
|
||||
extGateCalls++
|
||||
imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("boom")})
|
||||
var gatedNames []string
|
||||
gate := func(name string, f func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) {
|
||||
gatedNames = append(gatedNames, name)
|
||||
return f()
|
||||
}
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "ar", ItemID: "ar5"}, extGate)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "ar", ItemID: "ar5"}, gate)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.extError).To(BeTrue())
|
||||
Expect(extGateCalls).To(Equal(1))
|
||||
Expect(gatedNames).To(Equal([]string{"failAgent"}))
|
||||
})
|
||||
})
|
||||
|
||||
@ -313,7 +276,7 @@ var _ = Describe("resolveItem", func() {
|
||||
radioRepo.Data = map[string]*model.Radio{"ra1": {ID: "ra1", Name: "Radio"}}
|
||||
ds.MockedRadio = radioRepo
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "ra", ItemID: "ra1"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "ra", ItemID: "ra1"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res).To(Equal(resolution{}))
|
||||
})
|
||||
@ -329,7 +292,7 @@ var _ = Describe("resolveItem", func() {
|
||||
radioRepo.Data = map[string]*model.Radio{"ra2": {ID: "ra2", Name: "Radio", UploadedImage: "ra2_test.jpg"}}
|
||||
ds.MockedRadio = radioRepo
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "ra", ItemID: "ra2"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "ra", ItemID: "ra2"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).ToNot(BeNil())
|
||||
defer res.reader.Close()
|
||||
@ -361,7 +324,7 @@ var _ = Describe("resolveItem", func() {
|
||||
plRepo.TracksRepo = &tests.MockPlaylistTrackRepo{AlbumIDs: albumIDs}
|
||||
ds.MockedPlaylist = plRepo
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "pl1"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "pl1"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).ToNot(BeNil())
|
||||
defer res.reader.Close()
|
||||
@ -393,7 +356,7 @@ var _ = Describe("resolveItem", func() {
|
||||
plRepo.TracksRepo = &tests.MockPlaylistTrackRepo{AlbumIDs: []string{"t1", "t2"}}
|
||||
ds.MockedPlaylist = plRepo
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "plu"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "plu"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).ToNot(BeNil())
|
||||
defer res.reader.Close()
|
||||
@ -411,7 +374,7 @@ var _ = Describe("resolveItem", func() {
|
||||
plRepo.TracksRepo = &tests.MockPlaylistTrackRepo{AlbumIDs: []string{"t1", "t2"}}
|
||||
ds.MockedPlaylist = plRepo
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "pls"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "pls"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).ToNot(BeNil())
|
||||
defer res.reader.Close()
|
||||
@ -428,17 +391,17 @@ var _ = Describe("resolveItem", func() {
|
||||
plRepo.TracksRepo = &tests.MockPlaylistTrackRepo{AlbumIDs: []string{"t1"}}
|
||||
ds.MockedPlaylist = plRepo
|
||||
|
||||
var extGateCalls int
|
||||
extGate := func(f func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) {
|
||||
extGateCalls++
|
||||
var gatedNames []string
|
||||
gate := func(name string, _ func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) {
|
||||
gatedNames = append(gatedNames, name)
|
||||
return nil, "", errors.New("network down")
|
||||
}
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "ple"}, extGate)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "ple"}, gate)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).To(BeNil())
|
||||
Expect(res.extError).To(BeTrue())
|
||||
Expect(extGateCalls).To(Equal(1))
|
||||
Expect(gatedNames).To(Equal([]string{"m3u"}), "the playlist URL fetch is gated under \"m3u\"")
|
||||
})
|
||||
|
||||
It("treats a missing local ExternalImageURL as a definitive miss, not extError", func() {
|
||||
@ -449,7 +412,7 @@ var _ = Describe("resolveItem", func() {
|
||||
plRepo.TracksRepo = &tests.MockPlaylistTrackRepo{AlbumIDs: []string{"t1"}}
|
||||
ds.MockedPlaylist = plRepo
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "plm"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "plm"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).To(BeNil())
|
||||
Expect(res.extError).To(BeFalse())
|
||||
@ -467,7 +430,7 @@ var _ = Describe("resolveItem", func() {
|
||||
plRepo.TracksRepo = &tests.MockPlaylistTrackRepo{AlbumIDs: []string{"t1"}}
|
||||
ds.MockedPlaylist = plRepo
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "pl404"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "pl404"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).ToNot(BeNil())
|
||||
defer res.reader.Close()
|
||||
@ -488,7 +451,7 @@ var _ = Describe("resolveItem", func() {
|
||||
plRepo.TracksRepo = &tests.MockPlaylistTrackRepo{AlbumIDs: []string{"t1"}}
|
||||
ds.MockedPlaylist = plRepo
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "pl500"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "pl500"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).To(BeNil())
|
||||
Expect(res.extError).To(BeTrue())
|
||||
@ -504,7 +467,7 @@ var _ = Describe("resolveItem", func() {
|
||||
ds.MockedPlaylist = plRepo
|
||||
folderRepo.result = nil
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "pl2"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "pl2"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).To(BeNil())
|
||||
Expect(res.source).To(BeEmpty())
|
||||
@ -523,7 +486,7 @@ var _ = Describe("resolveItem", func() {
|
||||
plRepo.TracksRepo = &tests.MockPlaylistTrackRepo{AlbumIDs: []string{"t1"}}
|
||||
ds.MockedPlaylist = plRepo
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "plbomb"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "plbomb"}, nil)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(res.reader).To(BeNil())
|
||||
Expect(res.source).To(BeEmpty())
|
||||
@ -537,7 +500,7 @@ var _ = Describe("resolveItem", func() {
|
||||
plRepo.TracksRepo = &tests.MockPlaylistTrackRepo{AlbumIDs: []string{"missing1", "missing2"}}
|
||||
ds.MockedPlaylist = plRepo
|
||||
|
||||
res, err := resolveItem(ctx, ds, prov, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "pl3"}, nil)
|
||||
res, err := resolveItem(ctx, ds, ag, ffm, model.ArtworkQueueItem{ItemKind: "pl", ItemID: "pl3"}, nil)
|
||||
Expect(err).To(HaveOccurred())
|
||||
Expect(res).To(Equal(resolution{}))
|
||||
})
|
||||
|
||||
@ -198,18 +198,6 @@ func fromArtistExternalSource(ctx context.Context, ar model.Artist, provider ext
|
||||
}
|
||||
}
|
||||
|
||||
// fromArtistExternalResult is the worker's artist external step: via ArtistImageResult a
|
||||
// transient agent failure surfaces as an error (extError) rather than settling as absent.
|
||||
func fromArtistExternalResult(ctx context.Context, ar model.Artist, provider external.Provider) sourceFunc {
|
||||
return func() (io.ReadCloser, string, error) {
|
||||
imageUrl, err := provider.ArtistImageResult(ctx, ar.ID)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
return fromURL(ctx, imageUrl)
|
||||
}
|
||||
}
|
||||
|
||||
func fromAlbumExternalSource(ctx context.Context, al model.Album, provider external.Provider) sourceFunc {
|
||||
return func() (io.ReadCloser, string, error) {
|
||||
imageUrl, err := provider.AlbumImage(ctx, al.ID)
|
||||
|
||||
@ -10,9 +10,9 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/navidrome/navidrome/conf"
|
||||
"github.com/navidrome/navidrome/core/external"
|
||||
"github.com/navidrome/navidrome/core/ffmpeg"
|
||||
"github.com/navidrome/navidrome/core/agents"
|
||||
"github.com/navidrome/navidrome/core/auth"
|
||||
"github.com/navidrome/navidrome/core/ffmpeg"
|
||||
"github.com/navidrome/navidrome/log"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
"golang.org/x/time/rate"
|
||||
@ -28,35 +28,38 @@ const (
|
||||
|
||||
var errBreakerOpen = errors.New("artwork: external circuit breaker open")
|
||||
|
||||
// Worker drains the artwork queue through processItem: the external step is rate-limited
|
||||
// and circuit-broken, and prune is serialized against in-flight acquisitions via pruneMu.
|
||||
type Worker struct {
|
||||
deps workerDeps
|
||||
// extGate is one agent's rate limiter + circuit breaker; each external agent gets its
|
||||
// own so a provider whose API or CDN is down backs off in isolation from the others.
|
||||
type extGate struct {
|
||||
limiter *rate.Limiter
|
||||
breaker *breaker
|
||||
}
|
||||
|
||||
// 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.
|
||||
type Worker struct {
|
||||
deps workerDeps
|
||||
pruneMu sync.RWMutex
|
||||
wake chan struct{}
|
||||
runCtx context.Context
|
||||
|
||||
gatesMu sync.Mutex
|
||||
gates map[string]*extGate
|
||||
|
||||
mu sync.Mutex
|
||||
inFlight map[string]struct{}
|
||||
}
|
||||
|
||||
func NewWorker(ds model.DataStore, store *ImageStore, prov external.Provider, ffmpeg ffmpeg.FFmpeg) *Worker {
|
||||
rps := conf.Server.ArtworkExternalMaxRPS
|
||||
limit := rate.Inf // 0 or negative disables the external throttle
|
||||
if rps > 0 {
|
||||
limit = rate.Limit(rps)
|
||||
}
|
||||
func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg ffmpeg.FFmpeg) *Worker {
|
||||
w := &Worker{
|
||||
deps: workerDeps{ds: ds, store: store, prov: prov, ffmpeg: ffmpeg},
|
||||
limiter: rate.NewLimiter(limit, max(1, rps)),
|
||||
breaker: newBreaker(),
|
||||
deps: workerDeps{ds: ds, store: store, agents: ag, ffmpeg: ffmpeg},
|
||||
wake: make(chan struct{}, 1),
|
||||
runCtx: context.Background(),
|
||||
gates: map[string]*extGate{},
|
||||
inFlight: map[string]struct{}{},
|
||||
}
|
||||
w.deps.extGate = w.gate
|
||||
w.deps.gate = w.gate
|
||||
return w
|
||||
}
|
||||
|
||||
@ -195,20 +198,39 @@ func queueKey(it model.ArtworkQueueItem) string {
|
||||
return it.ItemKind + "|" + it.ItemID + "|" + it.ImageType
|
||||
}
|
||||
|
||||
// gate wraps the external step with the rate limiter and circuit breaker, matching
|
||||
// extGateFunc so it can be injected via workerDeps.extGate.
|
||||
func (w *Worker) gate(f func() (io.ReadCloser, string, error)) (io.ReadCloser, string, error) {
|
||||
if !w.breaker.allow() {
|
||||
// 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) {
|
||||
g := w.gateFor(name)
|
||||
if !g.breaker.allow() {
|
||||
return nil, "", errBreakerOpen
|
||||
}
|
||||
if err := w.limiter.Wait(w.runCtx); err != nil {
|
||||
if err := g.limiter.Wait(w.runCtx); err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
r, path, err := f()
|
||||
w.breaker.record(err)
|
||||
g.breaker.record(err)
|
||||
return r, path, err
|
||||
}
|
||||
|
||||
// gateFor lazily creates the per-name gate on first use, each with its own limiter at
|
||||
// ArtworkExternalMaxRPS and its own breaker.
|
||||
func (w *Worker) gateFor(name string) *extGate {
|
||||
w.gatesMu.Lock()
|
||||
defer w.gatesMu.Unlock()
|
||||
if g, ok := w.gates[name]; ok {
|
||||
return g
|
||||
}
|
||||
rps := conf.Server.ArtworkExternalMaxRPS
|
||||
limit := rate.Inf
|
||||
if rps > 0 {
|
||||
limit = rate.Limit(rps)
|
||||
}
|
||||
g := &extGate{limiter: rate.NewLimiter(limit, max(1, rps)), breaker: newBreaker()}
|
||||
w.gates[name] = g
|
||||
return g
|
||||
}
|
||||
|
||||
// backoffFor returns min(5m×4^n, 48h) scaled by (1+jitter), with jitter in [-0.2, 0.2].
|
||||
func backoffFor(attempts int, jitter float64) time.Duration {
|
||||
d := math.Min(float64(backoffBase)*math.Pow(4, float64(attempts)), float64(backoffCap))
|
||||
|
||||
@ -10,6 +10,7 @@ import (
|
||||
|
||||
"github.com/navidrome/navidrome/conf"
|
||||
"github.com/navidrome/navidrome/conf/configtest"
|
||||
"github.com/navidrome/navidrome/core/agents"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
"github.com/navidrome/navidrome/tests"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
@ -39,7 +40,7 @@ var _ = Describe("Worker soak", func() {
|
||||
ImageFiles: []string{"cover.jpg"},
|
||||
}}}
|
||||
ffm := tests.NewMockFFmpeg("")
|
||||
prov := &fakeExternalProvider{}
|
||||
ag := agents.GetAgents(&tests.MockDataStore{}, nil)
|
||||
artRepo := tests.CreateMockArtworkRepo()
|
||||
albumRepo := tests.CreateMockAlbumRepo()
|
||||
albumRepo.SetData(model.Albums{
|
||||
@ -53,7 +54,7 @@ var _ = Describe("Worker soak", func() {
|
||||
MockedAlbum: albumRepo,
|
||||
}
|
||||
store := NewImageStore(GinkgoT().TempDir())
|
||||
deps := &workerDeps{ds: ds, store: store, prov: prov, ffmpeg: ffm}
|
||||
deps := &workerDeps{ds: ds, store: store, agents: ag, ffmpeg: ffm}
|
||||
conf.Server.CoverArtPriority = "cover.jpg, embedded"
|
||||
|
||||
// Dangling refs (al/ra ids the repos don't know about) mirror an entity
|
||||
|
||||
@ -4,12 +4,12 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net/url"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"github.com/navidrome/navidrome/conf"
|
||||
"github.com/navidrome/navidrome/conf/configtest"
|
||||
"github.com/navidrome/navidrome/core/agents"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
"github.com/navidrome/navidrome/tests"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
@ -54,7 +54,7 @@ var _ = Describe("Worker", func() {
|
||||
folderRepo *fakeFolderRepo
|
||||
libRepo *tests.MockLibraryRepo
|
||||
ffm *tests.MockFFmpeg
|
||||
prov *fakeExternalProvider
|
||||
ag *agents.Agents
|
||||
store *ImageStore
|
||||
artRepo *tests.MockArtworkRepo
|
||||
queueRepo *tests.MockArtworkQueueRepo
|
||||
@ -73,7 +73,7 @@ var _ = Describe("Worker", func() {
|
||||
libRepo = &tests.MockLibraryRepo{}
|
||||
libRepo.SetData(model.Libraries{{ID: 0, Path: testFileLibPath(repoRoot)}})
|
||||
ffm = tests.NewMockFFmpeg("")
|
||||
prov = &fakeExternalProvider{}
|
||||
ag = agents.GetAgents(&tests.MockDataStore{}, nil)
|
||||
artRepo = tests.CreateMockArtworkRepo()
|
||||
queueRepo = tests.CreateMockArtworkQueueRepo()
|
||||
ds = &tests.MockDataStore{
|
||||
@ -86,7 +86,7 @@ var _ = Describe("Worker", func() {
|
||||
store = NewImageStore(GinkgoT().TempDir())
|
||||
conf.Server.CoverArtPriority = "cover.jpg, embedded"
|
||||
conf.Server.ArtworkExternalMaxRPS = 1000 // keep the limiter out of the way of behavior tests
|
||||
w = NewWorker(ds, store, prov, ffm)
|
||||
w = NewWorker(ds, store, ag, ffm)
|
||||
})
|
||||
|
||||
Describe("drain", func() {
|
||||
@ -118,9 +118,7 @@ var _ = Describe("Worker", func() {
|
||||
It("reschedules a failed item via MarkFailed with a backed-off retry_at", func() {
|
||||
conf.Server.CoverArtPriority = "external"
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{{ID: "al4", Name: "Album"}})
|
||||
prov.albumImage = func(context.Context, string) (*url.URL, error) {
|
||||
return nil, errors.New("agent timed out")
|
||||
}
|
||||
imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("agent timed out")})
|
||||
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "al", ItemID: "al4"})).To(Succeed())
|
||||
|
||||
n, err := w.drain(ctx, 2)
|
||||
@ -145,9 +143,7 @@ var _ = Describe("Worker", func() {
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
|
||||
{ID: "alstale", Name: "Album", FolderIDs: []string{"f1"}},
|
||||
})
|
||||
prov.albumImage = func(context.Context, string) (*url.URL, error) {
|
||||
return nil, errors.New("agent timed out")
|
||||
}
|
||||
imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("agent timed out")})
|
||||
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "al", ItemID: "alstale"})).To(Succeed())
|
||||
|
||||
n, err := w.drain(ctx, 2)
|
||||
@ -174,7 +170,7 @@ var _ = Describe("Worker", func() {
|
||||
})
|
||||
racing := &reenqueueOnDequeue{MockArtworkQueueRepo: queueRepo}
|
||||
ds.MockedArtworkQueue = racing
|
||||
w = NewWorker(ds, store, prov, ffm)
|
||||
w = NewWorker(ds, store, ag, ffm)
|
||||
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{
|
||||
ItemKind: "al", ItemID: "al7", Priority: model.ArtworkPriorityScan,
|
||||
})).To(Succeed())
|
||||
@ -193,12 +189,10 @@ var _ = Describe("Worker", func() {
|
||||
It("keeps a fresh re-enqueue ahead of a stale failure backoff", func() {
|
||||
conf.Server.CoverArtPriority = "external"
|
||||
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{{ID: "al8", Name: "Album"}})
|
||||
prov.albumImage = func(context.Context, string) (*url.URL, error) {
|
||||
return nil, errors.New("agent timed out")
|
||||
}
|
||||
imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("agent timed out")})
|
||||
racing := &reenqueueOnDequeue{MockArtworkQueueRepo: queueRepo}
|
||||
ds.MockedArtworkQueue = racing
|
||||
w = NewWorker(ds, store, prov, ffm)
|
||||
w = NewWorker(ds, store, ag, ffm)
|
||||
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "al", ItemID: "al8"})).To(Succeed())
|
||||
dequeued := findQueued(queueRepo, "al", "al8").RetryAt
|
||||
|
||||
@ -221,7 +215,7 @@ var _ = Describe("Worker", func() {
|
||||
private: model.Playlist{ID: "plPriv", OwnerID: "admin"},
|
||||
tracks: &tests.MockPlaylistTrackRepo{},
|
||||
}
|
||||
w = NewWorker(vds, store, prov, ffm)
|
||||
w = NewWorker(vds, store, ag, ffm)
|
||||
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "pl", ItemID: "plPriv"})).To(Succeed())
|
||||
|
||||
n, err := w.drain(ctx, 1)
|
||||
@ -259,13 +253,13 @@ var _ = Describe("Worker", func() {
|
||||
return nil, "", errors.New("boom")
|
||||
}
|
||||
for range 5 {
|
||||
_, _, err := w.gate(failing)
|
||||
_, _, err := w.gate("A", failing)
|
||||
Expect(err).To(HaveOccurred())
|
||||
}
|
||||
Expect(calls).To(Equal(5))
|
||||
|
||||
_, _, err := w.gate(failing)
|
||||
Expect(err).To(HaveOccurred())
|
||||
_, _, err := w.gate("A", failing)
|
||||
Expect(err).To(MatchError(errBreakerOpen))
|
||||
Expect(calls).To(Equal(5), "an open breaker must not call the external step")
|
||||
})
|
||||
|
||||
@ -273,9 +267,9 @@ var _ = Describe("Worker", func() {
|
||||
failing := func() (io.ReadCloser, string, error) { return nil, "", errors.New("boom") }
|
||||
ok := func() (io.ReadCloser, string, error) { return io.NopCloser(nil), "p", nil }
|
||||
for range 4 {
|
||||
_, _, _ = w.gate(failing)
|
||||
_, _, _ = w.gate("A", failing)
|
||||
}
|
||||
_, _, err := w.gate(ok)
|
||||
_, _, err := w.gate("A", ok)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
var calls int
|
||||
@ -284,10 +278,30 @@ var _ = Describe("Worker", func() {
|
||||
return nil, "", errors.New("boom")
|
||||
}
|
||||
for range 5 {
|
||||
_, _, _ = w.gate(counting)
|
||||
_, _, _ = w.gate("A", counting)
|
||||
}
|
||||
Expect(calls).To(Equal(5), "the breaker should have re-closed after the success")
|
||||
})
|
||||
|
||||
It("isolates each agent's breaker: one open gate does not block another", func() {
|
||||
failing := func() (io.ReadCloser, string, error) { return nil, "", errors.New("boom") }
|
||||
for range breakerThreshold {
|
||||
_, _, _ = w.gate("A", failing)
|
||||
}
|
||||
_, _, err := w.gate("A", failing)
|
||||
Expect(err).To(MatchError(errBreakerOpen), "agent A's breaker is open")
|
||||
|
||||
var bCalls int
|
||||
bStep := func() (io.ReadCloser, string, error) {
|
||||
bCalls++
|
||||
return io.NopCloser(nil), "p", nil
|
||||
}
|
||||
for range breakerThreshold + 2 {
|
||||
_, _, err := w.gate("B", bStep)
|
||||
Expect(err).ToNot(HaveOccurred(), "agent B keeps being called while A is open")
|
||||
}
|
||||
Expect(bCalls).To(Equal(breakerThreshold + 2))
|
||||
})
|
||||
})
|
||||
|
||||
Describe("RunPrune", func() {
|
||||
@ -313,7 +327,7 @@ var _ = Describe("Worker", func() {
|
||||
DeferCleanup(func() { goleak.VerifyNone(GinkgoT(), ignore) })
|
||||
|
||||
localDS := &tests.MockDataStore{MockedArtworkQueue: tests.CreateMockArtworkQueueRepo()}
|
||||
lw := NewWorker(localDS, NewImageStore(GinkgoT().TempDir()), &fakeExternalProvider{}, tests.NewMockFFmpeg(""))
|
||||
lw := NewWorker(localDS, NewImageStore(GinkgoT().TempDir()), agents.GetAgents(localDS, nil), tests.NewMockFFmpeg(""))
|
||||
|
||||
runCtx, cancel := context.WithCancel(ctx)
|
||||
done := make(chan error, 1)
|
||||
|
||||
@ -2,10 +2,13 @@ package artwork
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"io"
|
||||
"testing"
|
||||
"testing/synctest"
|
||||
"time"
|
||||
|
||||
"github.com/navidrome/navidrome/core/agents"
|
||||
"github.com/navidrome/navidrome/tests"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
@ -37,3 +40,39 @@ func TestArtworkBreakerHalfOpen(t *testing.T) {
|
||||
g.Expect(b.allow()).To(BeTrue())
|
||||
})
|
||||
}
|
||||
|
||||
// Drives the worker's per-name gate map with the fake clock: one agent's open breaker
|
||||
// must neither block another agent nor short-circuit the other's probe recovery.
|
||||
func TestArtworkGatePerAgentBreakerIsolation(t *testing.T) {
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
g := NewWithT(t)
|
||||
w := NewWorker(&tests.MockDataStore{}, NewImageStore(t.TempDir()),
|
||||
agents.GetAgents(&tests.MockDataStore{}, nil), tests.NewMockFFmpeg(""))
|
||||
|
||||
fail := func() (io.ReadCloser, string, error) { return nil, "", errors.New("boom") }
|
||||
for range breakerThreshold {
|
||||
_, _, _ = w.gate("A", fail)
|
||||
}
|
||||
_, _, err := w.gate("A", fail)
|
||||
g.Expect(err).To(MatchError(errBreakerOpen), "A opens after consecutive errors")
|
||||
|
||||
// B has its own breaker, untouched by A being open.
|
||||
var bCalls int
|
||||
bStep := func() (io.ReadCloser, string, error) { bCalls++; return nil, "", errors.New("boom") }
|
||||
for range breakerThreshold - 1 {
|
||||
_, _, err := w.gate("B", bStep)
|
||||
g.Expect(err).To(MatchError("boom"))
|
||||
}
|
||||
g.Expect(bCalls).To(Equal(breakerThreshold-1), "B keeps being called while A is open")
|
||||
|
||||
// After the probe window, A admits exactly one probe again.
|
||||
time.Sleep(breakerProbeAfter)
|
||||
var aCalls int
|
||||
aFail := func() (io.ReadCloser, string, error) { aCalls++; return nil, "", errors.New("boom") }
|
||||
_, _, _ = w.gate("A", aFail)
|
||||
g.Expect(aCalls).To(Equal(1), "A grants a single probe after the interval")
|
||||
_, _, err = w.gate("A", aFail)
|
||||
g.Expect(err).To(MatchError(errBreakerOpen), "the probe failed, so A stays open")
|
||||
g.Expect(aCalls).To(Equal(1))
|
||||
})
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user