From 756df9decf0cfe55a1eb528f892f31367e466c0c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Deluan=20Quint=C3=A3o?= Date: Thu, 16 Jul 2026 20:10:35 -0400 Subject: [PATCH] fix: dedupe and cap concurrent lyrics plugin fetches (#5792) * fix: dedupe and cap concurrent lyrics plugin fetches Clients like Finamp prefetch lyrics for several queue tracks at once. The resulting burst of concurrent plugin calls can rate-limit the primary lyrics provider into a timeout, making the plugin fall back to a lower quality source and cache the bad result. SimpleCache.GetWithLoader now deduplicates concurrent loads of the same key via singleflight, with every waiter receiving the winner's result or error. The Jellyfin lyrics loader is detached from the request context so one cancelled request cannot fail the load for all waiters, and the lyrics adapter caps in-flight plugin calls at 2 per plugin, queueing the rest. As a side effect, the cached HTTP client used by the Last.fm, Deezer and ListenBrainz agents also collapses identical concurrent requests into a single upstream call. * fix: harden lyrics concurrency fixes per review Replace the stringified singleflight keys with a per-cache flight map keyed by the cache key type itself, eliminating potential key collisions for non-string keys, the nil-interface assertion panic, and the stringification overhead. Release the lyrics semaphore slot via defer so a panicking plugin call cannot leak it, and bound the detached lyrics load with a one-minute timeout so a hung plugin cannot pin its singleflight and semaphore slot indefinitely. --- plugins/lyrics_adapter.go | 10 +++ plugins/lyrics_adapter_test.go | 41 +++++++++++ plugins/manager_loader.go | 1 + plugins/manager_plugin.go | 1 + server/jellyfin/lyrics.go | 9 ++- server/jellyfin/lyrics_test.go | 31 +++++++-- utils/cache/simple_cache.go | 79 ++++++++++++++++------ utils/cache/simple_cache_test.go | 112 +++++++++++++++++++++++++++++++ 8 files changed, 257 insertions(+), 27 deletions(-) diff --git a/plugins/lyrics_adapter.go b/plugins/lyrics_adapter.go index 281f022fb..9e02115e7 100644 --- a/plugins/lyrics_adapter.go +++ b/plugins/lyrics_adapter.go @@ -14,6 +14,10 @@ const ( FuncLyricsGetLyrics = "nd_lyrics_get_lyrics" ) +// maxConcurrentLyricsCalls caps in-flight lyrics calls per plugin: clients prefetch +// lyrics for whole queues, and the resulting burst can rate-limit upstream providers. +const maxConcurrentLyricsCalls = 2 + func init() { registerCapability( CapabilityLyrics, @@ -34,6 +38,12 @@ type LyricsPlugin struct { // GetLyrics calls the plugin to fetch lyrics, then content-sniffs each response // via model.ParseLyrics (TTML/SRT/YAML/LRC/plain). func (l *LyricsPlugin) GetLyrics(ctx context.Context, mf *model.MediaFile) (model.LyricList, error) { + select { + case l.plugin.lyricsSem <- struct{}{}: + defer func() { <-l.plugin.lyricsSem }() + case <-ctx.Done(): + return nil, ctx.Err() + } req := capabilities.GetLyricsRequest{ Track: mediaFileToTrackInfo(l.plugin, mf), } diff --git a/plugins/lyrics_adapter_test.go b/plugins/lyrics_adapter_test.go index 6e82dbfab..d110665f5 100644 --- a/plugins/lyrics_adapter_test.go +++ b/plugins/lyrics_adapter_test.go @@ -3,6 +3,8 @@ package plugins import ( + "context" + "github.com/navidrome/navidrome/model" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -71,6 +73,45 @@ var _ = Describe("LyricsPlugin", Ordered, func() { Expect(result[0].Lang).To(Equal("xxx")) }) + It("blocks new calls while the per-plugin concurrency cap is saturated", func() { + sem := provider.plugin.lyricsSem + for range cap(sem) { + sem <- struct{}{} + } + + ctx := GinkgoT().Context() + track := &model.MediaFile{ID: "track-1", Title: "Test Song", Artist: "Test Artist"} + done := make(chan error, 1) + go func() { + _, err := provider.GetLyrics(ctx, track) + done <- err + }() + + Consistently(done, "500ms").ShouldNot(Receive()) + <-sem // free one slot; the pending call should now proceed + Eventually(done).Should(Receive(BeNil())) + for range cap(sem) - 1 { + <-sem + } + }) + + It("gives up waiting for a slot when the context is cancelled", func() { + sem := provider.plugin.lyricsSem + for range cap(sem) { + sem <- struct{}{} + } + defer func() { + for range cap(sem) { + <-sem + } + }() + + ctx, cancel := context.WithCancel(GinkgoT().Context()) + cancel() + _, err := provider.GetLyrics(ctx, &model.MediaFile{ID: "track-1"}) + Expect(err).To(MatchError(context.Canceled)) + }) + It("returns error when plugin returns error", func() { manager, _ := createTestManagerWithPlugins(map[string]map[string]string{ "test-lyrics": {"error": "service unavailable"}, diff --git a/plugins/manager_loader.go b/plugins/manager_loader.go index 757ededb5..675c85e26 100644 --- a/plugins/manager_loader.go +++ b/plugins/manager_loader.go @@ -421,6 +421,7 @@ func (m *Manager) loadPluginWithConfig(p *model.Plugin) error { allowedUserIDs: allowedUsers, allUsers: p.AllUsers, libraries: newLibraryAccess(allowedLibraries, p.AllLibraries), + lyricsSem: make(chan struct{}, maxConcurrentLyricsCalls), } m.mu.Unlock() loaded = true diff --git a/plugins/manager_plugin.go b/plugins/manager_plugin.go index f0c7c56d5..155663781 100644 --- a/plugins/manager_plugin.go +++ b/plugins/manager_plugin.go @@ -24,6 +24,7 @@ type plugin struct { allowedUserIDs []string // User IDs this plugin can access (from DB configuration) allUsers bool // If true, plugin can access all users libraries libraryAccess + lyricsSem chan struct{} // Caps concurrent lyrics calls (see LyricsPlugin.GetLyrics) } // instance creates a new plugin instance for the given context. diff --git a/server/jellyfin/lyrics.go b/server/jellyfin/lyrics.go index f6468ceeb..a9d77ba19 100644 --- a/server/jellyfin/lyrics.go +++ b/server/jellyfin/lyrics.go @@ -10,11 +10,18 @@ import ( "github.com/navidrome/navidrome/server/jellyfin/dto" ) +const lyricsLoadTimeout = time.Minute + // cachedLyrics resolves lyrics through the full source pipeline (embedded, sidecar, plugins), // caching results — including empty: clients poll per played track, so misses are the hot path. func (api *Router) cachedLyrics(ctx context.Context, mf *model.MediaFile) model.LyricList { + // The load is shared across requests (singleflight) and cached, so don't let one + // cancelled request abort it for everybody — detach it from the request's lifetime, + // keeping a bound so a hung plugin can't pin the fetch (and its plugin slot) forever. + loadCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), lyricsLoadTimeout) + defer cancel() list, err := api.lyricsCache.GetWithLoader(mf.ID, func(string) (model.LyricList, time.Duration, error) { - l, err := api.lyrics.GetLyrics(ctx, mf) + l, err := api.lyrics.GetLyrics(loadCtx, mf) return l, 0, err // 0 → cache DefaultTTL }) if err != nil { diff --git a/server/jellyfin/lyrics_test.go b/server/jellyfin/lyrics_test.go index f2c5993f6..402d14ed0 100644 --- a/server/jellyfin/lyrics_test.go +++ b/server/jellyfin/lyrics_test.go @@ -17,13 +17,18 @@ import ( // fakeLyricsService returns canned lyrics per media-file ID and counts calls. type fakeLyricsService struct { - lyrics map[string]model.LyricList - err error - calls int + lyrics map[string]model.LyricList + err error + calls int + hadDeadline bool } -func (f *fakeLyricsService) GetLyrics(_ context.Context, mf *model.MediaFile) (model.LyricList, error) { +func (f *fakeLyricsService) GetLyrics(ctx context.Context, mf *model.MediaFile) (model.LyricList, error) { f.calls++ + _, f.hadDeadline = ctx.Deadline() + if err := ctx.Err(); err != nil { + return nil, err + } if f.err != nil { return nil, f.err } @@ -129,4 +134,22 @@ var _ = Describe("getLyrics", func() { Expect(doRequest("s2").Code).To(Equal(http.StatusNotFound)) Expect(fake.calls).To(Equal(1)) }) + + It("completes and caches the fetch even when the request context is cancelled", func() { + fake.lyrics["s1"] = model.LyricList{ + {Kind: "main", Synced: true, Line: []model.Line{{Start: p(1000), Value: "hello"}}}, + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + list := api.cachedLyrics(ctx, &model.MediaFile{ID: "s1"}) + Expect(list).ToNot(BeEmpty()) + Expect(doRequest("s1").Code).To(Equal(http.StatusOK)) + Expect(fake.calls).To(Equal(1)) + }) + + It("bounds the detached fetch with a timeout", func() { + Expect(doRequest("s2").Code).To(Equal(http.StatusNotFound)) + Expect(fake.hadDeadline).To(BeTrue()) + }) }) diff --git a/utils/cache/simple_cache.go b/utils/cache/simple_cache.go index eb3c99995..494451c9e 100644 --- a/utils/cache/simple_cache.go +++ b/utils/cache/simple_cache.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "runtime" + "sync" "sync/atomic" "time" @@ -44,7 +45,8 @@ func NewSimpleCache[K comparable, V any](options ...Options) SimpleCache[K, V] { c := ttlcache.New[K, V](opts...) cache := &simpleCache[K, V]{ - data: c, + data: c, + loads: make(map[K]*flight[V]), } go cache.data.Start() @@ -61,6 +63,23 @@ const evictionTimeout = 1 * time.Hour type simpleCache[K comparable, V any] struct { data *ttlcache.Cache[K, V] evictionDeadline atomic.Pointer[time.Time] + loadsMu sync.Mutex + loads map[K]*flight[V] +} + +// flight tracks an in-progress load so concurrent misses of the same key share it. +type flight[V any] struct { + done chan struct{} + val V + err error +} + +func (f *flight[V]) result() (V, error) { + if f.err != nil { + var zero V + return zero, fmt.Errorf("cache error: loader returned %w", f.err) + } + return f.val, nil } func (c *simpleCache[K, V]) Add(key K, value V) error { @@ -90,31 +109,47 @@ func (c *simpleCache[K, V]) Get(key K) (V, error) { return item.Value(), nil } +// GetWithLoader loads misses via the loader, deduplicating concurrent loads of +// the same key: one loader call runs, and every waiter shares its result (or error). func (c *simpleCache[K, V]) GetWithLoader(key K, loader func(key K) (V, time.Duration, error)) (V, error) { - var err error - loaderWrapper := ttlcache.LoaderFunc[K, V]( - func(t *ttlcache.Cache[K, V], key K) *ttlcache.Item[K, V] { - c.evictExpired() - var value V - var ttl time.Duration - value, ttl, err = loader(key) - if err != nil { - return nil - } - return t.Set(key, value, ttl) - }, - ) - item := c.data.Get(key, ttlcache.WithLoader[K, V](loaderWrapper)) - if item == nil { - var zero V - if err != nil { - return zero, fmt.Errorf("cache error: loader returned %w", err) - } - return zero, errors.New("item not found") + if item := c.data.Get(key); item != nil { + return item.Value(), nil } - return item.Value(), nil + + c.loadsMu.Lock() + if f, ok := c.loads[key]; ok { + c.loadsMu.Unlock() + <-f.done + return f.result() + } + f := &flight[V]{done: make(chan struct{}), err: errLoaderPanicked} + c.loads[key] = f + c.loadsMu.Unlock() + + // Deregister even if the loader panics, so waiters get an error instead of + // blocking forever on a flight that will never complete. + defer func() { + close(f.done) + c.loadsMu.Lock() + delete(c.loads, key) + c.loadsMu.Unlock() + }() + + if item := c.data.Get(key); item != nil { // a flight may have completed since the miss + f.val, f.err = item.Value(), nil + } else { + c.evictExpired() + var ttl time.Duration + f.val, ttl, f.err = loader(key) + if f.err == nil { + c.data.Set(key, f.val, ttl) + } + } + return f.result() } +var errLoaderPanicked = errors.New("loader panicked") + func (c *simpleCache[K, V]) evictExpired() { if c.evictionDeadline.Load() == nil || c.evictionDeadline.Load().Before(time.Now()) { c.data.DeleteExpired() diff --git a/utils/cache/simple_cache_test.go b/utils/cache/simple_cache_test.go index 45ba2c966..1c4f5c9bb 100644 --- a/utils/cache/simple_cache_test.go +++ b/utils/cache/simple_cache_test.go @@ -3,6 +3,8 @@ package cache import ( "errors" "fmt" + "sync" + "sync/atomic" "time" . "github.com/onsi/ginkgo/v2" @@ -69,6 +71,116 @@ var _ = Describe("SimpleCache", func() { _, err := cache.GetWithLoader("key", loader) Expect(err).To(HaveOccurred()) }) + + It("suppresses concurrent loads for the same key", func() { + var calls atomic.Int32 + release := make(chan struct{}) + started := make(chan struct{}, 10) + loader := func(key string) (string, time.Duration, error) { + calls.Add(1) + started <- struct{}{} + <-release + return "shared", time.Minute, nil + } + + const n = 5 + var wg sync.WaitGroup + results := make([]string, n) + errs := make([]error, n) + for i := range n { + wg.Go(func() { + results[i], errs[i] = cache.GetWithLoader("key", loader) + }) + } + + Eventually(started).Should(Receive()) + Consistently(started).ShouldNot(Receive()) + close(release) + wg.Wait() + + Expect(calls.Load()).To(Equal(int32(1))) + for i := range n { + Expect(errs[i]).ToNot(HaveOccurred()) + Expect(results[i]).To(Equal("shared")) + } + }) + + It("returns the loader error to all concurrent callers", func() { + release := make(chan struct{}) + started := make(chan struct{}, 10) + loader := func(key string) (string, time.Duration, error) { + started <- struct{}{} + <-release + return "", 0, errors.New("load failed") + } + + const n = 3 + var wg sync.WaitGroup + errs := make([]error, n) + for i := range n { + wg.Go(func() { + _, errs[i] = cache.GetWithLoader("key", loader) + }) + } + + Eventually(started).Should(Receive()) + Consistently(started).ShouldNot(Receive()) + close(release) + wg.Wait() + + for i := range n { + Expect(errs[i]).To(MatchError(ContainSubstring("load failed"))) + } + }) + + It("supports interface value types with nil results", func() { + c := NewSimpleCache[string, any]() + v, err := c.GetWithLoader("key", func(string) (any, time.Duration, error) { + return nil, time.Minute, nil + }) + Expect(err).ToNot(HaveOccurred()) + Expect(v).To(BeNil()) + }) + + It("cleans up the in-flight registration when the loader panics", func() { + Expect(func() { + _, _ = cache.GetWithLoader("key", func(string) (string, time.Duration, error) { + panic("boom") + }) + }).To(PanicWith("boom")) + + // Without cleanup this would deadlock on the never-completed flight + v, err := cache.GetWithLoader("key", func(string) (string, time.Duration, error) { + return "ok", 0, nil + }) + Expect(err).ToNot(HaveOccurred()) + Expect(v).To(Equal("ok")) + }) + + It("loads different keys independently", func() { + release := make(chan struct{}) + started := make(chan struct{}, 10) + loader := func(key string) (string, time.Duration, error) { + started <- struct{}{} + <-release + return key + "=value", time.Minute, nil + } + + var wg sync.WaitGroup + for _, key := range []string{"key1", "key2"} { + wg.Go(func() { + value, err := cache.GetWithLoader(key, loader) + Expect(err).ToNot(HaveOccurred()) + Expect(value).To(Equal(key + "=value")) + }) + } + + // Both loaders must be in flight at once: distinct keys are not suppressed + Eventually(started).Should(Receive()) + Eventually(started).Should(Receive()) + close(release) + wg.Wait() + }) }) Describe("Keys and Values", func() {