nik/navidrome
1
0
Fork 0
mirror of https://github.com/navidrome/navidrome.git synced 2026-08-01 07:21:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(artwork): broadcast refresh events when artwork lands

Browse Source
This commit is contained in:
Deluan 2026-07-22 20:53:22 -04:00
parent 5179691811
commit 05ba549843
6 changed files with 174 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmd/wire_gen.go
View File

@ -245,7 +245,7 @@ func CreateArtworkWorker() *artwork.Worker {
manager := plugins.GetManager(dataStore, broker, metricsMetrics)
agentsAgents := agents.GetAgents(dataStore, manager)
fFmpeg := ffmpeg.New()
worker := artwork.NewWorker(dataStore, imageStore, agentsAgents, fFmpeg)
worker := artwork.NewWorker(dataStore, imageStore, agentsAgents, fFmpeg, broker)
return worker
}

49
core/artwork/worker.go
View File

@ -15,6 +15,7 @@ import (
"github.com/navidrome/navidrome/core/ffmpeg"
"github.com/navidrome/navidrome/log"
"github.com/navidrome/navidrome/model"
"github.com/navidrome/navidrome/server/events"
"golang.org/x/time/rate"
)
@ -40,6 +41,7 @@ type extGate struct {
// via pruneMu.
type Worker struct {
deps workerDeps
broker events.Broker
pruneMu sync.RWMutex
wake chan struct{}
runCtx context.Context
@ -51,9 +53,10 @@ type Worker struct {
inFlight map[string]struct{}
}
func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg ffmpeg.FFmpeg) *Worker {
func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg ffmpeg.FFmpeg, broker events.Broker) *Worker {
w := &Worker{
deps: workerDeps{ds: ds, store: store, agents: ag, ffmpeg: ffmpeg},
broker: broker,
wake: make(chan struct{}, 1),
runCtx: context.Background(),
gates: map[string]*extGate{},
@ -131,6 +134,8 @@ func (w *Worker) drain(ctx context.Context, concurrency int) (int, error) {
}
sem := make(chan struct{}, concurrency)
var wg sync.WaitGroup
var foundMu sync.Mutex
var found []model.ArtworkQueueItem
for _, item := range items {
sem <- struct{}{}
wg.Add(1)
@ -138,14 +143,51 @@ func (w *Worker) drain(ctx context.Context, concurrency int) (int, error) {
defer wg.Done()
defer func() { <-sem }()
defer w.release(it)
w.process(ctx, it)
if w.process(ctx, it) == outcomeFound {
foundMu.Lock()
found = append(found, it)
foundMu.Unlock()
}
}(item)
}
wg.Wait()
w.broadcastRefresh(ctx, found)
return len(items), nil
}
func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) {
// artworkKindToResource maps a queue item's kind to the UI resource name carried
// in the refresh event.
var artworkKindToResource = map[string]string{
"al": "album",
"ar": "artist",
"pl": "playlist",
"ra": "radio",
"mf": "song",
}
// broadcastRefresh emits one coalesced RefreshResource for the batch's newly-acquired
// artwork, so connected UIs re-fetch the affected records (and pick up the new coverArt id).
func (w *Worker) broadcastRefresh(ctx context.Context, found []model.ArtworkQueueItem) {
if len(found) == 0 {
return
}
event := &events.RefreshResource{}
byResource := map[string][]string{}
for _, it := range found {
if res, ok := artworkKindToResource[it.ItemKind]; ok {
byResource[res] = append(byResource[res], it.ItemID)
}
}
if len(byResource) == 0 {
return
}
for res, ids := range byResource {
event = event.With(res, ids...)
}
w.broker.SendBroadcastMessage(ctx, event)
}
func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) outcome {
if item.ImageType == "" {
item.ImageType = model.ImageTypePrimary
}
@ -169,6 +211,7 @@ func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) {
log.Warn(ctx, "artwork: could not reschedule failed queue item", "kind", item.ItemKind, "id", item.ItemID, err)
}
}
return out
}
// claim reserves items not already in flight, so a row appearing twice within a single

83
core/artwork/worker_test.go
View File

@ -4,13 +4,16 @@ import (
"context"
"errors"
"io"
"net/http"
"os"
"sync"
"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/server/events"
"github.com/navidrome/navidrome/tests"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
@ -38,6 +41,32 @@ func (r *reenqueueOnDequeue) DequeueBatch(n int) ([]model.ArtworkQueueItem, erro
return items, err
}
type fakeEventBroker struct {
http.Handler
mu sync.Mutex
events []events.Event
}
func (f *fakeEventBroker) SendMessage(_ context.Context, event events.Event) {
f.mu.Lock()
defer f.mu.Unlock()
f.events = append(f.events, event)
}
func (f *fakeEventBroker) SendBroadcastMessage(_ context.Context, event events.Event) {
f.mu.Lock()
defer f.mu.Unlock()
f.events = append(f.events, event)
}
func (f *fakeEventBroker) getEvents() []events.Event {
f.mu.Lock()
defer f.mu.Unlock()
return f.events
}
var _ events.Broker = (*fakeEventBroker)(nil)
func findQueued(q *tests.MockArtworkQueueRepo, kind, id string) *model.ArtworkQueueItem {
for _, it := range q.Data {
if it.ItemKind == kind && it.ItemID == id {
@ -58,6 +87,7 @@ var _ = Describe("Worker", func() {
store *ImageStore
artRepo *tests.MockArtworkRepo
queueRepo *tests.MockArtworkQueueRepo
broker *fakeEventBroker
repoRoot string
w *Worker
)
@ -86,7 +116,8 @@ 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, ag, ffm)
broker = &fakeEventBroker{}
w = NewWorker(ds, store, ag, ffm, broker)
})
Describe("drain", func() {
@ -203,7 +234,7 @@ var _ = Describe("Worker", func() {
})
racing := &reenqueueOnDequeue{MockArtworkQueueRepo: queueRepo}
ds.MockedArtworkQueue = racing
w = NewWorker(ds, store, ag, ffm)
w = NewWorker(ds, store, ag, ffm, broker)
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{
ItemKind: "al", ItemID: "al7", Priority: model.ArtworkPriorityScan,
})).To(Succeed())
@ -225,7 +256,7 @@ var _ = Describe("Worker", func() {
imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("agent timed out")})
racing := &reenqueueOnDequeue{MockArtworkQueueRepo: queueRepo}
ds.MockedArtworkQueue = racing
w = NewWorker(ds, store, ag, ffm)
w = NewWorker(ds, store, ag, ffm, broker)
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "al", ItemID: "al8"})).To(Succeed())
dequeued := findQueued(queueRepo, "al", "al8").RetryAt
@ -248,7 +279,7 @@ var _ = Describe("Worker", func() {
private: model.Playlist{ID: "plPriv", OwnerID: "admin"},
tracks: &tests.MockPlaylistTrackRepo{},
}
w = NewWorker(vds, store, ag, ffm)
w = NewWorker(vds, store, ag, ffm, broker)
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "pl", ItemID: "plPriv"})).To(Succeed())
n, err := w.drain(ctx, 1)
@ -267,6 +298,48 @@ var _ = Describe("Worker", func() {
Expect(err).ToNot(HaveOccurred())
Expect(n).To(BeZero())
})
It("broadcasts a single refresh event for the found items in a batch", func() {
folderRepo.result = []model.Folder{{
Path: "tests/fixtures/artist/an-album",
ImageFiles: []string{"cover.jpg"},
}}
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{
{ID: "al1", Name: "Album 1", FolderIDs: []string{"f1"}},
{ID: "al2", Name: "Album 2", FolderIDs: []string{"f1"}},
})
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "al", ItemID: "al1", Priority: model.ArtworkPriorityScan})).To(Succeed())
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "al", ItemID: "al2", Priority: model.ArtworkPriorityScan})).To(Succeed())
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "ar", ItemID: "ar1", Priority: model.ArtworkPriorityScan})).To(Succeed())
n, err := w.drain(ctx, 3)
Expect(err).ToNot(HaveOccurred())
Expect(n).To(Equal(3))
evts := broker.getEvents()
Expect(evts).To(HaveLen(1), "exactly one coalesced event per drain batch")
rr, ok := evts[0].(*events.RefreshResource)
Expect(ok).To(BeTrue())
data := rr.Data(rr)
Expect(data).To(ContainSubstring(`"album"`))
Expect(data).To(ContainSubstring("al1"))
Expect(data).To(ContainSubstring("al2"))
Expect(data).ToNot(ContainSubstring("artist"), "the absent artist must not be refreshed")
Expect(data).ToNot(ContainSubstring("ar1"))
})
It("does not broadcast when no item is found", func() {
conf.Server.CoverArtPriority = "external"
ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{{ID: "alx", Name: "Album"}})
imageAgents(&fakeImageAgent{name: "failAgent", err: errors.New("agent timed out")})
Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ItemKind: "al", ItemID: "alx"})).To(Succeed())
n, err := w.drain(ctx, 2)
Expect(err).ToNot(HaveOccurred())
Expect(n).To(Equal(1))
Expect(broker.getEvents()).To(BeEmpty(), "a drain with no found items sends no event")
})
})
Describe("Bump", func() {
@ -379,7 +452,7 @@ var _ = Describe("Worker", func() {
DeferCleanup(func() { goleak.VerifyNone(GinkgoT(), ignore) })
localDS := &tests.MockDataStore{MockedArtworkQueue: tests.CreateMockArtworkQueueRepo()}
lw := NewWorker(localDS, NewImageStore(GinkgoT().TempDir()), agents.GetAgents(localDS, nil), tests.NewMockFFmpeg(""))
lw := NewWorker(localDS, NewImageStore(GinkgoT().TempDir()), agents.GetAgents(localDS, nil), tests.NewMockFFmpeg(""), &fakeEventBroker{})
runCtx, cancel := context.WithCancel(ctx)
done := make(chan error, 1)

2
core/artwork/worker_timing_test.go
View File

@ -47,7 +47,7 @@ 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(""))
agents.GetAgents(&tests.MockDataStore{}, nil), tests.NewMockFFmpeg(""), &fakeEventBroker{})
fail := func() (io.ReadCloser, string, error) { return nil, "", errors.New("boom") }
for range breakerThreshold {

21
tests/mock_artwork_queue_repo.go
View File

@ -2,6 +2,7 @@ package tests
import (
"sort"
"sync"
"time"
"github.com/navidrome/navidrome/model"
@ -9,6 +10,8 @@ import (
type MockArtworkQueueRepo struct {
model.ArtworkQueueRepository
// mu guards Data so the worker's concurrent drain can hit this mock race-free.
mu sync.Mutex
Data map[string]model.ArtworkQueueItem // keyed by iaKey(kind, id, imageType)
Err error
// ItemArtworkSource, when set, backs EnqueueStaleAbsent with real item_artwork state.
@ -22,6 +25,8 @@ func CreateMockArtworkQueueRepo() *MockArtworkQueueRepo {
}
func (m *MockArtworkQueueRepo) Enqueue(items ...model.ArtworkQueueItem) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return m.Err
}
@ -47,6 +52,8 @@ func (m *MockArtworkQueueRepo) Enqueue(items ...model.ArtworkQueueItem) error {
}
func (m *MockArtworkQueueRepo) DequeueBatch(n int) ([]model.ArtworkQueueItem, error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return nil, m.Err
}
@ -70,6 +77,8 @@ func (m *MockArtworkQueueRepo) DequeueBatch(n int) ([]model.ArtworkQueueItem, er
}
func (m *MockArtworkQueueRepo) MarkFailed(kind, id, imageType string, retryAt time.Time) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return m.Err
}
@ -85,6 +94,8 @@ func (m *MockArtworkQueueRepo) MarkFailed(kind, id, imageType string, retryAt ti
}
func (m *MockArtworkQueueRepo) MarkFailedIfUnchanged(kind, id, imageType string, seenRetryAt, retryAt time.Time) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return m.Err
}
@ -98,6 +109,8 @@ func (m *MockArtworkQueueRepo) MarkFailedIfUnchanged(kind, id, imageType string,
}
func (m *MockArtworkQueueRepo) Delete(kind, id, imageType string) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return m.Err
}
@ -106,6 +119,8 @@ func (m *MockArtworkQueueRepo) Delete(kind, id, imageType string) error {
}
func (m *MockArtworkQueueRepo) DeleteIfUnchanged(kind, id, imageType string, retryAt time.Time) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return m.Err
}
@ -117,6 +132,8 @@ func (m *MockArtworkQueueRepo) DeleteIfUnchanged(kind, id, imageType string, ret
}
func (m *MockArtworkQueueRepo) PurgeDangling() (int64, error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return 0, m.Err
}
@ -135,6 +152,8 @@ func (m *MockArtworkQueueRepo) PurgeDangling() (int64, error) {
}
func (m *MockArtworkQueueRepo) Count() (int64, error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return 0, m.Err
}
@ -142,6 +161,8 @@ func (m *MockArtworkQueueRepo) Count() (int64, error) {
}
func (m *MockArtworkQueueRepo) EnqueueStaleAbsent(kind string, attemptedBefore time.Time) (int64, error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil || m.ItemArtworkSource == nil {
return 0, m.Err
}

27
tests/mock_artwork_repo.go
View File

@ -1,6 +1,7 @@
package tests
import (
"sync"
"time"
"github.com/navidrome/navidrome/model"
@ -8,6 +9,8 @@ import (
type MockArtworkRepo struct {
model.ArtworkRepository
// mu guards the maps so the worker's concurrent drain can hit this mock race-free.
mu sync.Mutex
Data map[string]model.Artwork
ItemData map[string]model.ItemArtwork // keyed by iaKey(kind, id, imageType)
OrphanHashes []string
@ -23,6 +26,8 @@ func CreateMockArtworkRepo() *MockArtworkRepo {
func iaKey(kind, id, imageType string) string { return kind + "|" + id + "|" + imageType }
func (m *MockArtworkRepo) GetImage(hash string) (*model.Artwork, error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return nil, m.Err
}
@ -33,6 +38,8 @@ func (m *MockArtworkRepo) GetImage(hash string) (*model.Artwork, error) {
}
func (m *MockArtworkRepo) PutImage(a *model.Artwork) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return m.Err
}
@ -43,6 +50,8 @@ func (m *MockArtworkRepo) PutImage(a *model.Artwork) error {
}
func (m *MockArtworkRepo) GetImages(hashes []string) (map[string]model.Artwork, error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return nil, m.Err
}
@ -56,10 +65,14 @@ func (m *MockArtworkRepo) GetImages(hashes []string) (map[string]model.Artwork,
}
func (m *MockArtworkRepo) GetOrphanHashes(createdBefore time.Time) ([]string, error) {
m.mu.Lock()
defer m.mu.Unlock()
return m.OrphanHashes, m.Err
}
func (m *MockArtworkRepo) GetAllMimes() (map[string]string, error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return nil, m.Err
}
@ -71,6 +84,8 @@ func (m *MockArtworkRepo) GetAllMimes() (map[string]string, error) {
}
func (m *MockArtworkRepo) PurgeDanglingItemArtwork() (int64, error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return 0, m.Err
}
@ -90,6 +105,8 @@ func (m *MockArtworkRepo) PurgeDanglingItemArtwork() (int64, error) {
}
func (m *MockArtworkRepo) DeleteOrphans(createdBefore time.Time, hashes []string) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return m.Err
}
@ -116,6 +133,8 @@ func (m *MockArtworkRepo) referenced(hash string) bool {
}
func (m *MockArtworkRepo) GetItemArtwork(kind, id, imageType string) (*model.ItemArtwork, error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return nil, m.Err
}
@ -126,6 +145,8 @@ func (m *MockArtworkRepo) GetItemArtwork(kind, id, imageType string) (*model.Ite
}
func (m *MockArtworkRepo) PutItemArtwork(ia *model.ItemArtwork) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return m.Err
}
@ -141,6 +162,8 @@ func (m *MockArtworkRepo) PutItemArtwork(ia *model.ItemArtwork) error {
}
func (m *MockArtworkRepo) DeleteForItem(kind, id string) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return m.Err
}
@ -153,6 +176,8 @@ func (m *MockArtworkRepo) DeleteForItem(kind, id string) error {
}
func (m *MockArtworkRepo) DeleteForItems(kind string, ids []string) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return m.Err
}
@ -169,6 +194,8 @@ func (m *MockArtworkRepo) DeleteForItems(kind string, ids []string) error {
}
func (m *MockArtworkRepo) GetInfoForItems(kind string, ids []string) (map[string]model.ItemArtworkInfo, error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.Err != nil {
return nil, m.Err
}