From 05ba549843b337c200775fabadd8f4adbf79d8cc Mon Sep 17 00:00:00 2001 From: Deluan Date: Wed, 22 Jul 2026 20:53:22 -0400 Subject: [PATCH] feat(artwork): broadcast refresh events when artwork lands --- cmd/wire_gen.go | 2 +- core/artwork/worker.go | 49 ++++++++++++++++-- core/artwork/worker_test.go | 83 ++++++++++++++++++++++++++++-- core/artwork/worker_timing_test.go | 2 +- tests/mock_artwork_queue_repo.go | 21 ++++++++ tests/mock_artwork_repo.go | 27 ++++++++++ 6 files changed, 174 insertions(+), 10 deletions(-) diff --git a/cmd/wire_gen.go b/cmd/wire_gen.go index 6a500df3c..8f11b2efc 100644 --- a/cmd/wire_gen.go +++ b/cmd/wire_gen.go @@ -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 } diff --git a/core/artwork/worker.go b/core/artwork/worker.go index 9365cf50c..3261f35a7 100644 --- a/core/artwork/worker.go +++ b/core/artwork/worker.go @@ -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 diff --git a/core/artwork/worker_test.go b/core/artwork/worker_test.go index cee64987f..86a9d37a2 100644 --- a/core/artwork/worker_test.go +++ b/core/artwork/worker_test.go @@ -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) diff --git a/core/artwork/worker_timing_test.go b/core/artwork/worker_timing_test.go index 30c815c50..159a53081 100644 --- a/core/artwork/worker_timing_test.go +++ b/core/artwork/worker_timing_test.go @@ -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 { diff --git a/tests/mock_artwork_queue_repo.go b/tests/mock_artwork_queue_repo.go index 421e0dd15..500fdf4fd 100644 --- a/tests/mock_artwork_queue_repo.go +++ b/tests/mock_artwork_queue_repo.go @@ -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 } diff --git a/tests/mock_artwork_repo.go b/tests/mock_artwork_repo.go index 9e9f84272..aa873e7c6 100644 --- a/tests/mock_artwork_repo.go +++ b/tests/mock_artwork_repo.go @@ -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 }