From 878cf4b561faa579337b83d2d677d77ede67d475 Mon Sep 17 00:00:00 2001 From: Caleb Allen Date: Fri, 14 Aug 2026 13:33:29 -0400 Subject: [PATCH] fix(playback): submit a durability write, do not perform it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every write goes through one connection — MaxOpenConns(1), because SQLite has one writer — and a background pass can hold it for a long time. The player and the queue wrote inline from paths that hold their own mutexes, so a contended writer did not merely slow persistence down: SetQueue blocked in LoadFile's saveState and then in persistState, while holding q.mu and p.mu. That is the exact shape of the report: the track changed and the transport sat at paused, nothing appeared in the queue, and the play button did nothing because Queue.Play waited on the same held q.mu. Diagnosed by profiling the running app — 91% of its CPU was BackfillLibraryDiscographies → upsertBatch, with four of its six workers parked in sql.(*DB).conn. Jobs now run in submission order on one goroutine per component, each carrying its own snapshot. A job must not touch the component's fields — it holds no lock and the state has moved on — which is why persistTracks clones. SaveState still flushes and waits, because that is the one caller for which the row has to exist on return. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01UDCbcCZQepnpSQYJ6SxxZm --- backend/player/persistwriter.go | 61 ++++++++++++++ backend/player/player.go | 72 ++++++++++------- backend/queue/persistence.go | 84 ++++++++++++++------ backend/queue/persistwriter.go | 66 +++++++++++++++ backend/queue/persistwriter_test.go | 119 ++++++++++++++++++++++++++++ backend/queue/queue.go | 20 +++++ 6 files changed, 370 insertions(+), 52 deletions(-) create mode 100644 backend/player/persistwriter.go create mode 100644 backend/queue/persistwriter.go create mode 100644 backend/queue/persistwriter_test.go 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)