diff --git a/backend/player/persistwriter.go b/backend/player/persistwriter.go new file mode 100644 index 0000000..256a854 --- /dev/null +++ b/backend/player/persistwriter.go @@ -0,0 +1,61 @@ +package player + +// The player saves its state — volume, mute, the loaded track and its +// position — from inside every path that changes any of them, and every +// one of those paths holds p.mu. The write goes through SQLite's +// single writer connection (database.DB sets MaxOpenConns(1)), which a +// background pass can hold for seconds at a time. +// +// Inline, that made LoadFile block on an unrelated backfill while +// holding p.mu, so the queue's own SetQueue — which calls it under q.mu +// — froze the transport and the queue panel with it: the track changed, +// the state stayed paused, and Play() had not been reached yet. +// +// So the write is submitted and runs in submission order on one +// goroutine, off the lock. A job must not touch p; it gets a snapshot. +// SaveState (shutdown) is the one caller that waits. + +// persistQueueDepth is how many pending writes are buffered before a +// submission runs inline. Player state is written a few times per +// track, so the buffer is for a stalled writer, not for throughput. +const persistQueueDepth = 64 + +// writer returns the persistence goroutine's channel, starting it on +// first use. +func (p *Player) writer() chan func() { + p.persistOnce.Do(func() { + p.persistCh = make(chan func(), persistQueueDepth) + + go func() { + for job := range p.persistCh { + job() + } + }() + }) + + return p.persistCh +} + +// submitWrite queues a database write to run off the caller's lock. +// The caller may hold p.mu; the job may not touch p's state. +func (p *Player) submitWrite(job func()) { + select { + case p.writer() <- job: + default: + // The buffer is full, so the writer connection has been held + // for a long time. Running inline is the old behaviour and + // blocks the caller, but losing the state is worse: it is what + // the app reopens to. + p.logger.Warn("Player write buffer full, persisting inline") + job() + } +} + +// flushWrites blocks until every write submitted so far has run. +func (p *Player) flushWrites() { + done := make(chan struct{}) + + p.writer() <- func() { close(done) } + + <-done +} diff --git a/backend/player/player.go b/backend/player/player.go index 641d661..3c29718 100644 --- a/backend/player/player.go +++ b/backend/player/player.go @@ -70,6 +70,12 @@ type Player struct { // consumer can tell "the same second, again" from "a fresh // reading" and reset its interpolation on both. positionSeq uint64 + + // persistCh carries state writes to the single goroutine that runs + // them, so no path holds mu while waiting on SQLite's writer + // connection. See persistwriter.go. + persistOnce sync.Once + persistCh chan func() } // PositionInfo is the payload of the PlaybackPositionChanged event: @@ -1180,17 +1186,25 @@ func (p *Player) buildMediaMetadata( // State persistence // --------------------------------------------------------------- -// SaveState persists the current player state to the database. -// This is called during shutdown to capture the final state. +// SaveState persists the current player state to the database and +// waits for the write. This is called during shutdown to capture the +// final state, which is the one case that cannot be deferred. func (p *Player) SaveState() { p.mu.Lock() - defer p.mu.Unlock() - p.saveState() + p.mu.Unlock() + + p.flushWrites() } -// saveState is the internal helper that writes the current player -// state to the database. Must be called with p.mu held. +// saveState snapshots the current player state and hands it to the +// persistence goroutine. Must be called with p.mu held. +// +// It does not write here: every caller holds p.mu (LoadFile, Play, +// Pause, Seek, the volume paths), the write goes through SQLite's +// single writer connection, and a background pass can hold that for +// seconds — which is how loading a track came to block the whole +// player behind a durability write. See persistwriter.go. func (p *Player) saveState() { if p.db == nil { p.logger.Warn( @@ -1215,29 +1229,29 @@ func (p *Player) saveState() { positionSeconds := int64(p.displayPositionSecsLocked()) - err := p.db.Queries.UpdatePlayerState( - p.db.Ctx, - sqlcgen.UpdatePlayerStateParams{ - Volume: volume, - Muted: muted, - LastTrackPath: trackPath, - LastPositionSeconds: positionSeconds, - }, - ) - if err != nil { - p.logger.Error( - "Failed to save player state", "err", err, - ) - - return + params := sqlcgen.UpdatePlayerStateParams{ + Volume: volume, + Muted: muted, + LastTrackPath: trackPath, + LastPositionSeconds: positionSeconds, } - p.logger.Info("Player state saved", - "volume", volume, - "muted", muted, - "trackPath", trackPath, - "positionSeconds", positionSeconds, - ) + p.submitWrite(func() { + if err := p.db.Queries.UpdatePlayerState(p.db.Ctx, params); err != nil { + p.logger.Error( + "Failed to save player state", "err", err, + ) + + return + } + + p.logger.Info("Player state saved", + "volume", volume, + "muted", muted, + "trackPath", trackPath, + "positionSeconds", positionSeconds, + ) + }) } // --------------------------------------------------------------- @@ -1253,6 +1267,10 @@ func (p *Player) RestoreState() { } func (p *Player) restoreStateLocked() { + // Reads back what saveState wrote, so it waits for anything still + // in flight rather than restoring the state before last. + p.flushWrites() + defer profiling.TimeOp(p.logger, "player.RestoreState")() if p.db == nil { diff --git a/backend/queue/persistence.go b/backend/queue/persistence.go index 92e7da1..d09af7a 100644 --- a/backend/queue/persistence.go +++ b/backend/queue/persistence.go @@ -4,6 +4,7 @@ import ( "database/sql" "encoding/json" "fmt" + "slices" "strings" "yellowjacket/backend/coverart" @@ -15,13 +16,15 @@ import ( // No position shifting is needed because this is always an append. // The caller must hold q.mu. func (q *Queue) persistAddTrack(track Track) { - _, err := q.db.Queries.InsertQueueTrack(q.db.Ctx, sqlcgen.InsertQueueTrackParams{ - AudioFileID: track.AudioFileID, - Position: track.Position, + q.submitWrite(func() { + _, err := q.db.Queries.InsertQueueTrack(q.db.Ctx, sqlcgen.InsertQueueTrackParams{ + AudioFileID: track.AudioFileID, + Position: track.Position, + }) + if err != nil { + q.logger.Error("Failed to persist added track", "err", err) + } }) - if err != nil { - q.logger.Error("Failed to persist added track", "err", err) - } } // persistAddTracks inserts multiple tracks at the end of the queue @@ -33,6 +36,14 @@ func (q *Queue) persistAddTracks(tracks []Track) { return } + // Cloned because the caller's slice is usually a window onto + // q.tracks, which mutates the moment the lock is released. + snapshot := slices.Clone(tracks) + + q.submitWrite(func() { q.writeAddTracks(snapshot) }) +} + +func (q *Queue) writeAddTracks(tracks []Track) { tx, err := q.db.BeginTx() if err != nil { q.logger.Error("Failed to begin transaction", "err", err) @@ -84,6 +95,12 @@ func (q *Queue) persistInsertTracks(tracks []Track, insertPos int) { return } + snapshot := slices.Clone(tracks) + + q.submitWrite(func() { q.writeInsertTracks(snapshot, insertPos) }) +} + +func (q *Queue) writeInsertTracks(tracks []Track, insertPos int) { tx, err := q.db.BeginTx() if err != nil { q.logger.Error("Failed to begin transaction", "err", err) @@ -145,6 +162,10 @@ func (q *Queue) persistInsertTracks(tracks []Track, insertPos int) { // shifts subsequent positions down to close the gap. // The caller must hold q.mu. func (q *Queue) persistRemoveTrack(position int) { + q.submitWrite(func() { q.writeRemoveTrack(position) }) +} + +func (q *Queue) writeRemoveTrack(position int) { tx, err := q.db.BeginTx() if err != nil { q.logger.Error("Failed to begin transaction", "err", err) @@ -262,6 +283,12 @@ func (q *Queue) lookupChunk( // persistTracks writes the current queue tracks to the database atomically // using a transaction with batched multi-row inserts. func (q *Queue) persistTracks() { + snapshot := slices.Clone(q.tracks) + + q.submitWrite(func() { q.writeTracks(snapshot) }) +} + +func (q *Queue) writeTracks(tracks []Track) { tx, err := q.db.BeginTx() if err != nil { q.logger.Error("Failed to begin transaction", "err", err) @@ -296,13 +323,13 @@ func (q *Queue) persistTracks() { batchSize := maxSQLiteVars / varsPerRow - for i := 0; i < len(q.tracks); i += batchSize { + for i := 0; i < len(tracks); i += batchSize { end := i + batchSize - if end > len(q.tracks) { - end = len(q.tracks) + if end > len(tracks) { + end = len(tracks) } - batch := q.tracks[i:end] + batch := tracks[i:end] if insertErr := q.insertTrackBatch(tx, batch); insertErr != nil { q.logger.Error( @@ -369,30 +396,33 @@ func (q *Queue) persistState() { } } - err := q.db.Queries.UpdateQueueState( - q.db.Ctx, - sqlcgen.UpdateQueueStateParams{ - CurrentPosition: int64(q.currentIndex), - ShuffleMode: q.shuffleMode, - RepeatMode: string(q.repeatMode), - ShuffleOrder: shuffleOrderJSON, - SourceType: q.source.Type, - SourceID: q.source.ID, - SourceLabel: q.source.Label, - }, - ) - if err != nil { - q.logger.Error("Failed to persist queue state", "err", err) + params := sqlcgen.UpdateQueueStateParams{ + CurrentPosition: int64(q.currentIndex), + ShuffleMode: q.shuffleMode, + RepeatMode: string(q.repeatMode), + ShuffleOrder: shuffleOrderJSON, + SourceType: q.source.Type, + SourceID: q.source.ID, + SourceLabel: q.source.Label, } + + q.submitWrite(func() { + if err := q.db.Queries.UpdateQueueState(q.db.Ctx, params); err != nil { + q.logger.Error("Failed to persist queue state", "err", err) + } + }) } -// SaveState persists the queue state to the database. +// SaveState persists the queue state to the database. Unlike every +// other write here it waits for the writer: its callers are shutdown +// and the tests, both of which need the row to exist on return. func (q *Queue) SaveState() { q.mu.Lock() defer q.mu.Unlock() q.persistTracks() q.persistState() + q.flushWrites() q.logger.Info("Queue state saved", "trackCount", len(q.tracks), "currentIndex", q.currentIndex, @@ -408,6 +438,10 @@ func (q *Queue) RestoreState() { q.mu.Lock() defer q.mu.Unlock() + // The other read-back: at startup there is nothing pending, but a + // restore after any mutation must see it. + q.flushWrites() + // Restore queue metadata. state, err := q.db.ReadQueries.GetQueueState(q.db.Ctx) if err != nil { diff --git a/backend/queue/persistwriter.go b/backend/queue/persistwriter.go new file mode 100644 index 0000000..ce85515 --- /dev/null +++ b/backend/queue/persistwriter.go @@ -0,0 +1,66 @@ +package queue + +// Every queue write goes through SQLite's single writer connection +// (database.DB sets MaxOpenConns(1)), which a background pass can hold +// for seconds at a time — the discography backfill's per-row FTS +// maintenance is the measured example. Doing these writes inline meant +// SetQueue held q.mu across that wait, so the queue panel, the play +// button and every other bound method blocked behind a durability write +// nobody was waiting for: the track changed, the transport sat at +// paused, and the queue never arrived. +// +// So a write is *submitted*, not performed. Jobs run in submission +// order on one goroutine, each carrying its own snapshot, which is what +// keeps "clear and rewrite the queue" and "insert three tracks at 4" +// meaning the same thing they meant at the moment they were called. A +// job must therefore never touch q's fields — it has no lock and the +// state has moved on. +// +// SaveState is the one caller that still waits: shutdown is exactly the +// case where the write has to have happened before we return. + +// persistQueueDepth is how many pending writes are buffered before a +// submission runs inline. These are user-driven mutations, so the +// buffer exists for a stalled writer rather than for throughput. +const persistQueueDepth = 256 + +// writer returns the persistence goroutine's channel, starting it on +// first use. Lazy because a Queue is constructed in tests that never +// write, and an idle goroutine per instance is a cost with no reader. +func (q *Queue) writer() chan func() { + q.persistOnce.Do(func() { + q.persistCh = make(chan func(), persistQueueDepth) + + go func() { + for job := range q.persistCh { + job() + } + }() + }) + + return q.persistCh +} + +// submitWrite queues a database write to run off the caller's lock. +// The caller may hold q.mu; the job may not touch q. +func (q *Queue) submitWrite(job func()) { + select { + case q.writer() <- job: + default: + // The buffer is full, which means the writer connection has + // been held for a long time. Running inline is the old + // behaviour and blocks the caller — but a dropped write is a + // queue that restores wrong, which is worse. + q.logger.Warn("Queue write buffer full, persisting inline") + job() + } +} + +// flushWrites blocks until every write submitted so far has run. +func (q *Queue) flushWrites() { + done := make(chan struct{}) + + q.writer() <- func() { close(done) } + + <-done +} diff --git a/backend/queue/persistwriter_test.go b/backend/queue/persistwriter_test.go new file mode 100644 index 0000000..0b1d47a --- /dev/null +++ b/backend/queue/persistwriter_test.go @@ -0,0 +1,119 @@ +package queue + +import ( + "testing" + "time" +) + +// mustFinish fails the test if fn has not finished within d. +func mustFinish(t *testing.T, d time.Duration, what string, fn func()) { + t.Helper() + + done := make(chan struct{}) + + go func() { + fn() + close(done) + }() + + select { + case <-done: + case <-time.After(d): + t.Fatalf("%s did not finish within %s", what, d) + } +} + +// The bug this exists for: SQLite's writer is a single connection, a +// background pass can hold it for seconds, and the queue used to do its +// writes inline while holding q.mu. So starting an album left the +// track changed, the transport at paused and the queue panel empty — +// SetQueue was still waiting on a durability write, and every other +// bound method was waiting on SetQueue. +// +// A stalled writer must cost nothing but durability. +func TestStalledWriterDoesNotBlockTheQueue(t *testing.T) { + t.Parallel() + + q, db := setupTestQueue(t) + paths := seedAudioFiles(t, db, 5) + + // Occupy the persistence goroutine the way a held write connection + // does, and keep it occupied for the rest of the test. + release := make(chan struct{}) + defer close(release) + + q.submitWrite(func() { <-release }) + + mustFinish(t, 5*time.Second, "SetQueue", func() { + q.SetQueue(paths, 0, false, Source{Type: "album", ID: 1, Label: "Test"}) + }) + + mustFinish(t, 5*time.Second, "GetState", func() { + if got := len(q.GetState().Tracks); got != len(paths) { + t.Errorf("tracks = %d, want %d", got, len(paths)) + } + }) + + mustFinish(t, 5*time.Second, "Play", func() { q.Play() }) + + mustFinish(t, 5*time.Second, "AddTrack", func() { q.AddTrack(paths[0]) }) + + mustFinish(t, 5*time.Second, "RemoveTrack", func() { q.RemoveTrack(0) }) +} + +// Order is the whole reason these run on one goroutine: "clear and +// rewrite the queue" followed by "insert at 4" is not the same thing in +// the other order. +func TestWritesRunInSubmissionOrder(t *testing.T) { + t.Parallel() + + q, _ := setupTestQueue(t) + + var order []int + + for i := range 20 { + q.submitWrite(func() { order = append(order, i) }) + } + + q.flushWrites() + + if len(order) != 20 { + t.Fatalf("ran %d writes, want 20", len(order)) + } + + for i, got := range order { + if got != i { + t.Fatalf("write %d ran at position %d", got, i) + } + } +} + +// Making the writes asynchronous introduces one hazard the inline +// version could not have: a path that reads back what it wrote. There +// are two — RestoreState and CompactAfterLibraryRemoval — and both must +// see the writes that are still in flight, or they rebuild the queue +// from the one before it. +func TestRestoreStateWaitsForPendingWrites(t *testing.T) { + t.Parallel() + + q, db := setupTestQueue(t) + paths := seedAudioFiles(t, db, 4) + + // Hold the writer so SetQueue's rows are provably still pending. + release := make(chan struct{}) + + q.submitWrite(func() { <-release }) + + q.SetQueue(paths, 0, false, Source{}) + + go func() { + time.Sleep(50 * time.Millisecond) + close(release) + }() + + q.RestoreState() + + if got := len(q.GetState().Tracks); got != len(paths) { + t.Errorf("tracks after restore = %d, want %d", got, len(paths)) + } +} diff --git a/backend/queue/queue.go b/backend/queue/queue.go index 0739594..f141550 100644 --- a/backend/queue/queue.go +++ b/backend/queue/queue.go @@ -181,6 +181,12 @@ type Queue struct { // setQueueGen is incremented each time SetQueue is called. Background // goroutines check this to detect if they have been superseded. setQueueGen atomic.Int64 + + // persistCh carries database writes to the single goroutine that + // runs them, so no mutation path waits on the writer connection + // while holding mu. See persistwriter.go. + persistOnce sync.Once + persistCh chan func() } // NewQueue creates a new queue manager. @@ -1061,6 +1067,13 @@ func (q *Queue) Play() { "Resume requested but player not ready", "err", err, ) + + // A play button that does nothing and says nothing is the + // fault PlaybackFailed exists for (errors.C1); this was the + // one press path still ending at a log line. + if q.currentIndex < len(q.tracks) { + q.emitPlaybackFailed(q.tracks[q.currentIndex], err) + } } return @@ -1427,6 +1440,13 @@ func (q *Queue) CompactAfterLibraryRemoval() { // Reload surviving tracks from the database. The CASCADE delete // already removed the rows from queue_tracks — we just need to // reload and reindex. + // + // This is one of the two places that reads back what it wrote, so + // it waits for the pending writes first: a queue built moments ago + // may still be in flight, and reloading from rows it has not + // reached yet would replace the queue with the previous one. + q.flushWrites() + rows, err := q.db.Queries.GetQueueTracks(q.db.Ctx) if err != nil { q.logger.Error("could not reload queue tracks after library removal", "err", err)