From dfd4796e4d2fd2e6df81ae6fd33ca029861b3356 Mon Sep 17 00:00:00 2001 From: fossisawesome-macbook-with-linux Date: Fri, 3 Jul 2026 15:53:16 -0400 Subject: [PATCH] fix(jukebox): address review feedback on mutex refactor - PlaybackDone now carries the finished *mpv.MpvTrack instead of a bare bool, so trackSwitcherGoroutine can detect and drop stale finish signals raised for a track that Skip/Clear/Set already replaced. Previously a stale signal could cause the goroutine to tear down and advance past the newly-started track. - getStatus() snapshots ActiveTrack/index/gain under the lock, then performs the blocking mpv IPC calls (Position/IsPlaying) after releasing it, so Status/Get polling no longer holds the device mutex during a socket round-trip to mpv. - Add/Set now resolve media file IDs (DB queries) before acquiring the lock, only holding it to mutate the queue. Co-Authored-By: Claude --- core/playback/device.go | 96 +++++++++++++++++++++++++---------- core/playback/mpv/mpv_test.go | 2 +- core/playback/mpv/track.go | 6 +-- 3 files changed, 73 insertions(+), 31 deletions(-) diff --git a/core/playback/device.go b/core/playback/device.go index 43214706e..5e08b5d48 100644 --- a/core/playback/device.go +++ b/core/playback/device.go @@ -32,7 +32,7 @@ type playbackDevice struct { DeviceName string PlaybackQueue *Queue Gain float32 - PlaybackDone chan bool + PlaybackDone chan *mpv.MpvTrack ActiveTrack Track startTrackSwitcher sync.Once } @@ -46,7 +46,9 @@ type DeviceStatus struct { const DefaultGain float32 = 1.0 -// getStatusLocked must be called with pd.mutex held. +// getStatusLocked must be called with pd.mutex held. It performs blocking IPC +// calls to mpv (Position/IsPlaying) while holding the lock; prefer getStatus +// for read-only call sites that don't already need the lock for anything else. func (pd *playbackDevice) getStatusLocked() DeviceStatus { pos := 0 if pd.ActiveTrack != nil { @@ -60,6 +62,30 @@ func (pd *playbackDevice) getStatusLocked() DeviceStatus { } } +// getStatus snapshots device state under the lock, then performs the blocking +// mpv IPC calls (Position/IsPlaying) after releasing it, so it doesn't block +// other concurrent operations on the device. +func (pd *playbackDevice) getStatus() DeviceStatus { + pd.mutex.Lock() + track := pd.ActiveTrack + index := pd.PlaybackQueue.Index + gain := pd.Gain + pd.mutex.Unlock() + + pos := 0 + playing := false + if track != nil { + pos = track.Position() + playing = track.IsPlaying() + } + return DeviceStatus{ + CurrentIndex: index, + Playing: playing, + Gain: gain, + Position: pos, + } +} + // NewPlaybackDevice creates a new playback device which implements all the basic Jukebox mode commands defined here: // http://www.subsonic.org/pages/api.jsp#jukeboxControl // Starts the trackSwitcher goroutine for the device. @@ -72,7 +98,7 @@ func NewPlaybackDevice(ctx context.Context, playbackServer PlaybackServer, name DeviceName: deviceName, Gain: DefaultGain, PlaybackQueue: NewQueue(), - PlaybackDone: make(chan bool), + PlaybackDone: make(chan *mpv.MpvTrack), } } @@ -83,26 +109,31 @@ func (pd *playbackDevice) String() string { func (pd *playbackDevice) Get(ctx context.Context) (model.MediaFiles, DeviceStatus, error) { log.Debug(ctx, "Processing Get action", "device", pd) pd.mutex.Lock() - defer pd.mutex.Unlock() - return pd.PlaybackQueue.Get(), pd.getStatusLocked(), nil + items := pd.PlaybackQueue.Get() + pd.mutex.Unlock() + return items, pd.getStatus(), nil } func (pd *playbackDevice) Status(ctx context.Context) (DeviceStatus, error) { log.Debug(ctx, fmt.Sprintf("processing Status action on: %s, queue: %s", pd, pd.PlaybackQueue)) - pd.mutex.Lock() - defer pd.mutex.Unlock() - return pd.getStatusLocked(), nil + return pd.getStatus(), nil } // Set is similar to a clear followed by a add, but will not change the currently playing track. func (pd *playbackDevice) Set(ctx context.Context, ids []string) (DeviceStatus, error) { log.Debug(ctx, "Processing Set action", "ids", ids, "device", pd) - pd.mutex.Lock() - defer pd.mutex.Unlock() + items, err := pd.fetchMediaFiles(ctx, ids) + if err != nil { + return DeviceStatus{}, err + } + pd.mutex.Lock() pd.clearLocked() - return pd.addLocked(ctx, ids) + pd.PlaybackQueue.Add(items) + pd.mutex.Unlock() + + return pd.getStatus(), nil } func (pd *playbackDevice) Start(ctx context.Context) (DeviceStatus, error) { @@ -198,29 +229,35 @@ func (pd *playbackDevice) Skip(ctx context.Context, index int, offset int) (Devi func (pd *playbackDevice) Add(ctx context.Context, ids []string) (DeviceStatus, error) { log.Debug(ctx, "Processing Add action", "ids", ids, "device", pd) - pd.mutex.Lock() - defer pd.mutex.Unlock() - return pd.addLocked(ctx, ids) -} - -func (pd *playbackDevice) addLocked(ctx context.Context, ids []string) (DeviceStatus, error) { if len(ids) < 1 { - return pd.getStatusLocked(), nil + return pd.getStatus(), nil } - items := model.MediaFiles{} + items, err := pd.fetchMediaFiles(ctx, ids) + if err != nil { + return DeviceStatus{}, err + } + pd.mutex.Lock() + pd.PlaybackQueue.Add(items) + pd.mutex.Unlock() + + return pd.getStatus(), nil +} + +// fetchMediaFiles resolves media file IDs to model.MediaFiles. It performs +// database queries and must not be called while pd.mutex is held. +func (pd *playbackDevice) fetchMediaFiles(ctx context.Context, ids []string) (model.MediaFiles, error) { + items := model.MediaFiles{} for _, id := range ids { mf, err := pd.ParentPlaybackServer.GetMediaFile(id) if err != nil { - return DeviceStatus{}, err + return nil, err } log.Debug(ctx, "Found mediafile: "+mf.Path) items = append(items, *mf) } - pd.PlaybackQueue.Add(items) - - return pd.getStatusLocked(), nil + return items, nil } func (pd *playbackDevice) Clear(ctx context.Context) (DeviceStatus, error) { @@ -296,13 +333,18 @@ func (pd *playbackDevice) trackSwitcherGoroutine() { log.Debug("Started trackSwitcher goroutine", "device", pd) for { select { - case <-pd.PlaybackDone: + case finishedTrack := <-pd.PlaybackDone: log.Debug("Track switching detected") pd.mutex.Lock() - if pd.ActiveTrack != nil { - pd.ActiveTrack.Close() - pd.ActiveTrack = nil + if pd.ActiveTrack != finishedTrack { + // The active track was already replaced (e.g. by Skip/Clear/Set) + // since this finish signal was sent. Ignore the stale signal. + log.Debug("Ignoring stale track-finished signal") + pd.mutex.Unlock() + continue } + pd.ActiveTrack.Close() + pd.ActiveTrack = nil if !pd.PlaybackQueue.IsAtLastElement() { pd.PlaybackQueue.IncreaseIndex() diff --git a/core/playback/mpv/mpv_test.go b/core/playback/mpv/mpv_test.go index 6754b39ac..6e5a0658a 100644 --- a/core/playback/mpv/mpv_test.go +++ b/core/playback/mpv/mpv_test.go @@ -349,7 +349,7 @@ var _ = Describe("MPV", func() { ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) defer cancel() - playbackDone := make(chan bool, 1) + playbackDone := make(chan *MpvTrack, 1) _, err := NewTrack(ctx, playbackDone, "auto", testMediaFile) Expect(err).To(HaveOccurred()) Expect(err.Error()).To(Equal("no mpv command arguments provided")) diff --git a/core/playback/mpv/track.go b/core/playback/mpv/track.go index 1038b9190..c5731ce6a 100644 --- a/core/playback/mpv/track.go +++ b/core/playback/mpv/track.go @@ -18,14 +18,14 @@ import ( type MpvTrack struct { MediaFile model.MediaFile - PlaybackDone chan bool + PlaybackDone chan<- *MpvTrack Conn *mpvipc.Connection IPCSocketName string Exe *Executor CloseCalled bool } -func NewTrack(ctx context.Context, playbackDoneChannel chan bool, deviceName string, mf model.MediaFile) (*MpvTrack, error) { +func NewTrack(ctx context.Context, playbackDoneChannel chan<- *MpvTrack, deviceName string, mf model.MediaFile) (*MpvTrack, error) { log.Debug("Loading track", "trackPath", mf.Path, "mediaType", mf.ContentType()) if _, err := mpvCommand(); err != nil { @@ -65,7 +65,7 @@ func NewTrack(ctx context.Context, playbackDoneChannel chan bool, deviceName str conn.WaitUntilClosed() log.Info("Hitting end-of-stream, signalling on channel") if !theTrack.CloseCalled { - playbackDoneChannel <- true + playbackDoneChannel <- theTrack } }()