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

fix(scrobbler): tolerate out-of-order playback reports (#5793)

Browse Source 
* fix(scrobbler): tolerate out-of-order playback reports

Clients may fire reportPlayback requests concurrently (Feishin sends
'starting' and 'playing' in parallel, and the previous track's 'stopped'
races the next track's start), so reports can be processed out of order.
Two cases corrupted the now-playing session: a late 'starting' for the
track already playing downgraded the session state, freezing position
estimation at 0:00 until the next report; and a late 'stopped' for the
previous track removed the new track's session and dispatched a playback
report mislabeled with the new track's metadata, causing presence-style
plugins (e.g. Discord Rich Presence) to clear or show stale state.

ReportPlayback now ignores a 'starting' report when the session already
has the same track in playing state, and ignores a 'stopped' report for
a track other than the current session's - skipping both the session
removal and the plugin dispatch, while still counting the play and
dispatching external scrobbles for the stopped track.

Reported in https://github.com/jeffvli/feishin/issues/2131

* fix(scrobbler): serialize session writes to close starting/playing race

The out-of-order 'starting' guard checked the session cache before the
media-file load, leaving a window where a concurrent 'playing' report on a
fresh session could write between the check and the write, and still be
overwritten back to 'starting'. Session check-then-write sections are now
serialized by a mutex, with the guard re-checked after the load. Also
tightens the guard comment to say 'playing session', matching the condition.

Found by Codex review on #5793.

* fix(scrobbler): fully exit ReportPlayback when ignoring out-of-order reports

The out-of-order guards used 'break', which only exits the switch, so the
post-switch NowPlaying block still ran for an ignored 'starting' report and
enqueued a NowPlaying dispatch with the stale report's position - potentially
overwriting a pending correct-position entry, since the queue is keyed by
client. Return nil instead, so ignored reports have no side effects.

Found by Gemini review on #5793.
This commit is contained in:
Deluan Quintão 2026-07-16 20:47:47 -04:00 committed by GitHub
parent 756df9decf
commit 27f0210392
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 167 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
core/scrobbler/play_tracker.go
View File

@ -89,6 +89,7 @@ type playTracker struct {
ds model.DataStore
broker events.Broker
playMap cache.SimpleCache[string, PlaybackSession]
sessionsMu sync.Mutex // serializes playMap check-then-write across concurrent reports
builtinScrobblers map[string]Scrobbler
pluginScrobblers map[string]Scrobbler
pluginLoader PluginLoader
@ -249,6 +250,12 @@ func (p *playTracker) getActiveScrobblers() map[string]Scrobbler {
return combined
}
// hasPlayingSession reports whether clientId's current session is already playing mediaId.
func (p *playTracker) hasPlayingSession(clientId, mediaId string) bool {
cur, err := p.playMap.Get(clientId)
return err == nil && cur.MediaFile.ID == mediaId && cur.State == StatePlaying
}
func remainingTTL(durationSec float32, positionMs int64, rate float64) time.Duration {
if rate <= 0 {
rate = 1.0
@ -268,6 +275,12 @@ func (p *playTracker) ReportPlayback(ctx context.Context, params ReportPlaybackP
switch params.State {
case StateStarting:
// Clients may send starting/playing unordered; a late "starting" must not downgrade
// a playing session, or position estimation freezes until the next report.
if p.hasPlayingSession(clientId, params.MediaId) {
log.Trace(ctx, "Ignoring out-of-order starting report for playing session", "clientId", clientId, "mediaId", params.MediaId)
return nil
}
mf, err := p.ds.MediaFile(ctx).GetWithParticipants(params.MediaId)
if err != nil {
return err
@ -284,7 +297,15 @@ func (p *playTracker) ReportPlayback(ctx context.Context, params ReportPlaybackP
PlaybackRate: params.PlaybackRate,
LastReport: now,
}
p.sessionsMu.Lock()
// re-check: a concurrent "playing" report may have created the session during the load above
if p.hasPlayingSession(clientId, params.MediaId) {
p.sessionsMu.Unlock()
log.Trace(ctx, "Ignoring out-of-order starting report for playing session", "clientId", clientId, "mediaId", params.MediaId)
return nil
}
err = p.playMap.AddWithTTL(clientId, info, remainingTTL(mf.Duration, params.PositionMs, params.PlaybackRate))
p.sessionsMu.Unlock()
if err != nil {
log.Warn(ctx, "Error adding PlaybackSession to cache", "clientId", clientId, "mediaId", params.MediaId, "state", params.State, err)
}
@ -315,7 +336,9 @@ func (p *playTracker) ReportPlayback(ctx context.Context, params ReportPlaybackP
ttl = remainingTTL(info.MediaFile.Duration, params.PositionMs, params.PlaybackRate)
}
log.Trace(ctx, "Updating PlaybackSession in cache", "clientId", clientId, "mediaId", params.MediaId, "state", params.State, "positionMs", params.PositionMs, "playbackRate", params.PlaybackRate, "ttl", ttl)
p.sessionsMu.Lock()
err := p.playMap.AddWithTTL(clientId, info, ttl)
p.sessionsMu.Unlock()
if err != nil {
log.Warn(ctx, "Error updating PlaybackSession in cache", "clientId", clientId, "mediaId", params.MediaId, "state", params.State, err)
}
@ -339,6 +362,17 @@ func (p *playTracker) ReportPlayback(ctx context.Context, params ReportPlaybackP
p.dispatchScrobble(ctx, mf, now)
}
}
p.sessionsMu.Lock()
info, getErr := p.playMap.Get(clientId)
// A late stop for a previous track must not end the current session nor reach
// playback reporters, or presence-style plugins would clear the active track.
if getErr == nil && info.MediaFile.ID != params.MediaId {
p.sessionsMu.Unlock()
log.Trace(ctx, "Ignoring out-of-order stopped report for different track", "clientId", clientId, "stoppedMediaId", params.MediaId, "currentMediaId", info.MediaFile.ID)
return nil
}
p.playMap.Remove(clientId)
p.sessionsMu.Unlock()
stoppedInfo := PlaybackSession{
UserId: user.ID,
Username: user.UserName,
@ -349,7 +383,7 @@ func (p *playTracker) ReportPlayback(ctx context.Context, params ReportPlaybackP
PlaybackRate: params.PlaybackRate,
LastReport: now,
}
if info, getErr := p.playMap.Get(clientId); getErr == nil {
if getErr == nil {
stoppedInfo.MediaFile = info.MediaFile
stoppedInfo.Start = info.Start
} else {
@ -364,7 +398,6 @@ func (p *playTracker) ReportPlayback(ctx context.Context, params ReportPlaybackP
stoppedInfo.MediaFile = *mf
}
p.enqueuePlaybackReport(ctx, stoppedInfo)
p.playMap.Remove(clientId)
}
if conf.Server.EnableNowPlaying {

134
core/scrobbler/play_tracker_test.go
View File

@ -3,6 +3,7 @@ package scrobbler
import (
"context"
"errors"
"fmt"
"net/http"
"sync"
"sync/atomic"
@ -45,6 +46,17 @@ func (m *mockPluginLoader) LoadScrobbler(name string) (Scrobbler, bool) {
return s, ok
}
// slowMediaFileRepo widens the window between a report's session check and its
// write, making check-then-write races reproducible.
type slowMediaFileRepo struct {
model.MediaFileRepository
}
func (s *slowMediaFileRepo) GetWithParticipants(id string) (*model.MediaFile, error) {
time.Sleep(5 * time.Millisecond)
return s.MediaFileRepository.GetWithParticipants(id)
}
var _ = Describe("PlayTracker", func() {
var ctx context.Context
var ds model.DataStore
@ -376,18 +388,23 @@ var _ = Describe("PlayTracker", func() {
Expect(playing).To(BeEmpty())
})
It("starting replaces existing entry for same player", func() {
It("starting replaces existing entry when switching tracks on same player", func() {
track2 := track
track2.ID = "456"
_ = ds.MediaFile(ctx).Put(&track2)
err := tracker.ReportPlayback(ctx, ReportPlaybackParams{
MediaId: "123", PositionMs: 50000, State: "playing", PlaybackRate: 1.0, ClientId: defaultClientId,
})
Expect(err).ToNot(HaveOccurred())
err = tracker.ReportPlayback(ctx, ReportPlaybackParams{
MediaId: "123", PositionMs: 0, State: "starting", PlaybackRate: 1.0, ClientId: defaultClientId,
MediaId: "456", PositionMs: 0, State: "starting", PlaybackRate: 1.0, ClientId: defaultClientId,
})
Expect(err).ToNot(HaveOccurred())
playing, err := tracker.GetNowPlaying(ctx)
Expect(err).ToNot(HaveOccurred())
Expect(playing).To(HaveLen(1))
Expect(playing[0].MediaFile.ID).To(Equal("456"))
Expect(playing[0].State).To(Equal("starting"))
Expect(playing[0].PositionMs).To(Equal(int64(0)))
})
@ -696,6 +713,119 @@ var _ = Describe("PlayTracker", func() {
})
})
Describe("resilience (out-of-order reports)", func() {
BeforeEach(func() {
track2 := track
track2.ID = "456"
_ = ds.MediaFile(ctx).Put(&track2)
})
It("does not downgrade an actively playing session when a late starting report arrives for the same track", func() {
err := tracker.ReportPlayback(ctx, ReportPlaybackParams{
MediaId: "123", PositionMs: 1000, State: "playing", PlaybackRate: 1.0, ClientId: defaultClientId,
})
Expect(err).ToNot(HaveOccurred())
err = tracker.ReportPlayback(ctx, ReportPlaybackParams{
MediaId: "123", PositionMs: 0, State: "starting", PlaybackRate: 1.0, ClientId: defaultClientId,
})
Expect(err).ToNot(HaveOccurred())
playing, err := tracker.GetNowPlaying(ctx)
Expect(err).ToNot(HaveOccurred())
Expect(playing).To(HaveLen(1))
Expect(playing[0].State).To(Equal("playing"))
Expect(playing[0].PositionMs).To(BeNumerically(">=", int64(1000)))
})
It("keeps the current session when a stopped report arrives for a different track", func() {
err := tracker.ReportPlayback(ctx, ReportPlaybackParams{
MediaId: "456", PositionMs: 0, State: "playing", PlaybackRate: 1.0, ClientId: defaultClientId,
})
Expect(err).ToNot(HaveOccurred())
err = tracker.ReportPlayback(ctx, ReportPlaybackParams{
MediaId: "123", PositionMs: 90000, State: "stopped", PlaybackRate: 1.0, ClientId: defaultClientId,
})
Expect(err).ToNot(HaveOccurred())
playing, err := tracker.GetNowPlaying(ctx)
Expect(err).ToNot(HaveOccurred())
Expect(playing).To(HaveLen(1))
Expect(playing[0].MediaFile.ID).To(Equal("456"))
Expect(playing[0].State).To(Equal("playing"))
})
It("still auto-scrobbles the stopped track when the current session is for a different track", func() {
err := tracker.ReportPlayback(ctx, ReportPlaybackParams{
MediaId: "456", PositionMs: 0, State: "playing", PlaybackRate: 1.0, ClientId: defaultClientId,
})
Expect(err).ToNot(HaveOccurred())
err = tracker.ReportPlayback(ctx, ReportPlaybackParams{
MediaId: "123", PositionMs: 90000, State: "stopped", PlaybackRate: 1.0, ClientId: defaultClientId,
})
Expect(err).ToNot(HaveOccurred())
Expect(track.PlayCount).To(Equal(int64(1)))
})
It("does not dispatch NowPlaying from an ignored out-of-order starting report", func() {
err := tracker.ReportPlayback(ctx, ReportPlaybackParams{
MediaId: "123", PositionMs: 60000, State: "playing", PlaybackRate: 1.0, ClientId: defaultClientId,
})
Expect(err).ToNot(HaveOccurred())
Eventually(func() bool { return fake.GetNowPlayingCalled() }).Should(BeTrue())
fake.nowPlayingCalled.Store(false)
err = tracker.ReportPlayback(ctx, ReportPlaybackParams{
MediaId: "123", PositionMs: 0, State: "starting", PlaybackRate: 1.0, ClientId: defaultClientId,
})
Expect(err).ToNot(HaveOccurred())
Consistently(func() bool { return fake.GetNowPlayingCalled() }).Should(BeFalse())
})
It("never lets a concurrent starting report downgrade the playing session", func() {
ds.(*tests.MockDataStore).MockedMediaFile = &slowMediaFileRepo{MediaFileRepository: ds.MediaFile(ctx)}
for i := range 20 {
raceClientId := fmt.Sprintf("race-client-%d", i)
var wg sync.WaitGroup
wg.Add(2)
go func() {
defer wg.Done()
defer GinkgoRecover()
_ = tracker.ReportPlayback(ctx, ReportPlaybackParams{
MediaId: "123", PositionMs: 0, State: "starting", PlaybackRate: 1.0, ClientId: raceClientId,
})
}()
go func() {
defer wg.Done()
defer GinkgoRecover()
_ = tracker.ReportPlayback(ctx, ReportPlaybackParams{
MediaId: "123", PositionMs: 0, State: "playing", PlaybackRate: 1.0, ClientId: raceClientId,
})
}()
wg.Wait()
info, err := tracker.playMap.Get(raceClientId)
Expect(err).ToNot(HaveOccurred())
Expect(info.State).To(Equal("playing"), "iteration %d", i)
}
})
It("does NOT forward a stopped report for a different track to playback reporters", func() {
err := tracker.ReportPlayback(ctx, ReportPlaybackParams{
MediaId: "456", PositionMs: 0, State: "playing", PlaybackRate: 1.0, ClientId: defaultClientId,
})
Expect(err).ToNot(HaveOccurred())
Eventually(func() bool { return fake.PlaybackReportCalled.Load() }).Should(BeTrue())
fake.PlaybackReportCalled.Store(false)
fake.LastPlaybackReport.Store(nil)
err = tracker.ReportPlayback(ctx, ReportPlaybackParams{
MediaId: "123", PositionMs: 100000, State: "stopped", PlaybackRate: 1.0, ClientId: defaultClientId,
})
Expect(err).ToNot(HaveOccurred())
Consistently(func() bool { return fake.PlaybackReportCalled.Load() }).Should(BeFalse())
})
})
Describe("external scrobbler dispatch", func() {
It("dispatches NowPlaying on starting", func() {
fake.nowPlayingCalled.Store(false)