diff --git a/core/scrobbler/play_tracker.go b/core/scrobbler/play_tracker.go index a2d7e8639..e21db42d2 100644 --- a/core/scrobbler/play_tracker.go +++ b/core/scrobbler/play_tracker.go @@ -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 { diff --git a/core/scrobbler/play_tracker_test.go b/core/scrobbler/play_tracker_test.go index 831b0ce0d..f49d9a0bf 100644 --- a/core/scrobbler/play_tracker_test.go +++ b/core/scrobbler/play_tracker_test.go @@ -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)