From ee6dd1bc031154788988c7b62cac18f2a49c67b5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Deluan=20Quint=C3=A3o?= Date: Wed, 23 Sep 2026 17:04:50 -0400 Subject: [PATCH] fix(scanner): stop DB lock starvation during scans on slow storage (#6201) * fix(artwork): pause the artwork worker while a scan is running The artwork worker added in 0.64 writes to the database continuously, including while a scan runs. On slow storage the scanner holds the write lock for many seconds per folder, so the two writers keep timing each other out: artwork writes fail with "database is locked", and a single busy timeout on the scanner side aborts the whole scan. The worker now stops dispatching queue items while scanner.IsScanning reports true, including mid-batch, and resumes on the next poll after the scan ends. Artwork requests are unaffected, since they serve local art without the worker. * fix(db): run ANALYZE one index at a time so writers are not starved A full ANALYZE is a single write transaction, so every other write waits for it to finish and fails after the 15s busy timeout. On slow NAS storage it was measured taking over 26 minutes. The analysis now runs ANALYZE per index (per table for unindexed and WITHOUT ROWID tables), which produces the same sqlite_stat1 rows as a full ANALYZE, and pauses briefly between steps (up to 150ms, just above SQLite's longest busy-handler sleep) so waiting writers get the lock. * fix(scanner): ignore Synology @eaDir metadata folders Synology creates an @eaDir folder next to media files, holding one subfolder per file with generated thumbnails. The scanner and watcher treated them as regular folders, which on one reported library added tens of thousands of extra folders to every scan. * fix(db): analyze tables with only partial indexes as a whole A partial index does not record the table's row count, so a table whose only indexes are partial needs a table-level ANALYZE to get the sqlite_stat1 row a full ANALYZE would write. Navidrome's schema has no such table today, but the stepped analysis should match a full ANALYZE for any schema a future migration creates. * fix(scanner): retry busy folder saves and stop phase 1 on a fatal error On slow storage, a single SQLITE_BUSY while saving a folder aborted the whole scan, even when another writer held the lock only briefly. The folder save now runs as a retryable unit: on a busy error it waits (5s, 10s, 15s) and reruns the transaction, up to three times, before failing. Side effects that do not survive a rollback (the album ID map consumed by persistAlbum, the artwork queue items, the image-change record) are rebuilt per attempt or recorded only after a successful commit. When a folder save does fail, phase 1 used to keep walking the library and reading tags for every remaining folder, discarding the results, before reporting the error; a reporter saw 40 silent minutes. The walk now stops as soon as the save fails, and the walker honors cancellation instead of blocking on its channel. Because an early stop leaves folders unvisited, phase 1 no longer marks unvisited folders missing when the phase failed; the resumed scan handles them. * refactor(persistence): move busy retry into DataStore.WithTxRetry The scanner retried its folder save itself, which meant it had to know SQLite error codes. WithTxRetry now owns that policy: it reruns the block in a fresh transaction on SQLITE_BUSY, up to three times with growing delays, and runs it only once when already inside a transaction, since the outer transaction would still hold the lock. The block receives the context to use, and attempts that will be retried carry a marker so a busy statement in them is logged as a warning; only the final attempt logs errors. The scanner's inner error logs are folded into wrapped errors, so a recovered retry no longer prints error-level lines, and the folder path travels in the log context. * fix(persistence): join the enclosing transaction in a nested WithTxRetry Called on a store that is already inside a transaction, WithTxRetry went through WithTx, which opens a second, independent transaction on another connection. That transaction waits on the lock the outer one holds and fails with SQLITE_BUSY, and if it does succeed the outer transaction cannot roll it back. It now runs the block on the enclosing transaction, which owns the lock, the commit and the rollback. Found by a Codex (gpt-6-sol) review. * fix(scanner): retry the remaining scan writes on a busy database Every write step after phase 1 still aborted the whole scan on a single SQLITE_BUSY: phase 1 finalize, phase 2 moves and purge, phase 3 album saves and play count refreshes, the deferred playlist import flag, library ScanBegin, GC, the missing-artwork enqueue, tag counts, and the final library update. They now go through WithTxRetry. The phase 2 move had to be made rerun-safe first: it changed the target track's ID inside the transaction, so a rerun would have deleted the moved track itself, and it marked album annotations as handled even when the transaction rolled back. It now works on a copy per attempt and records the annotation reassignment only after a commit. Artist.RefreshStats is left alone: it updates artists in batches outside a transaction, and one transaction around all of them would hold the write lock for the whole refresh on slow storage. Phase 4 playlist imports go through the playlist service and are left for a follow-up. * fix(scanner): claim the album before moving its annotations The rerun-safe moveMatched checked processedAlbumAnnotations before its transaction and marked the album only after the commit. Phase 2 runs same-library and cross-library moves in separate pipeline stages, so two moves into one album could both pass the check; the second would reassign annotations again and overwrite the album's created_at. The album is now claimed under the lock before the transaction, as the old code effectively did, and the claim is released if the move fails so a later move can still reassign. Found by a Codex (gpt-6-sol) review. * fix(artwork): keep artwork housekeeping from writing during scans The artwork worker already pauses while a scan runs, but its housekeeping jobs did not: the hourly missing-artwork recheck (a bulk INSERT ... SELECT over albums and artists), the startup run of the same recheck, and the daily prune all kept competing with the scanner for the write lock. They now run through LockForMaintenance, like the scheduled DB analysis: they skip while a scan is running and keep a scan from starting until they finish. Skipping the recheck loses nothing, since each scan with changes queues missing artwork at its end. * refactor(scanner): log retried step errors once, from the caller Blocks passed to WithTxRetry still logged their own errors at error level on every attempt, so a busy error that a retry absorbed printed several error lines (GC printed three). They now return wrapped errors and the callers, which already log them, report the final outcome once. Also: drop a leftover variable in phase 1 finalize, check the walk context once, stop repeating the folder field that is already in the log context, stop shadowing finalize's err in phase 3, and format the WithTxRetry scope the same way as WithTx. * test(scanner): make the scanner suite's temp DB cleanup best effort Which DB file the process-wide DB handle opens depends on which spec touches it first. When the Scanner container wins the random order, its temp DB stays open until db.Close after RunSpecs, and on Windows removing the temp dir fails with 'being used by another process'. Ginkgo pins that on the container's last spec, which is now one of the busy-database specs. The sibling suites skip Windows for the same reason; this one now removes its temp dir on a best-effort basis instead, so it keeps running there. --- cmd/root.go | 28 ++- core/artwork/worker.go | 13 ++ core/artwork/worker_test.go | 52 +++++ db/db.go | 6 + db/db_test.go | 16 ++ db/optimize.go | 56 ++++- db/optimize_test.go | 37 ++++ model/datastore.go | 3 + persistence/persistence.go | 60 +++++- persistence/persistence_test.go | 73 +++++++ persistence/sql_base_repository.go | 4 + scanner/phase_1_folders.go | 280 +++++++++++++------------ scanner/phase_2_missing_tracks.go | 97 +++++---- scanner/phase_2_missing_tracks_test.go | 114 ++++++++++ scanner/phase_3_refresh_albums.go | 17 +- scanner/phase_4_playlists.go | 5 +- scanner/scanner.go | 41 ++-- scanner/scanner_test.go | 62 +++++- scanner/walk_dir_tree.go | 13 +- scanner/walk_dir_tree_test.go | 1 + tests/mock_data_store.go | 4 + 21 files changed, 754 insertions(+), 228 deletions(-) diff --git a/cmd/root.go b/cmd/root.go index f632b0425..02cd30240 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -375,10 +375,26 @@ func startPlaybackServer(ctx context.Context) func() error { func startArtworkWorker(ctx context.Context, worker *artwork.Worker) func() error { return func() error { log.Info(ctx, "Starting artwork worker") + // The scanner writes to the DB for its whole run; competing for the write lock makes both fail. + worker.PauseWhile(scanner.IsScanning) return worker.Run(ctx) } } +// outsideScan runs a DB maintenance job unless a scan is running, and keeps a scan from starting +// until it ends; both write to the DB, and competing for the lock can make either fail. +func outsideScan(ctx context.Context, job string, run func(context.Context) error) { + release, ok := scanner.LockForMaintenance() + if !ok { + log.Debug(ctx, "Skipping "+job+" because a scan is in progress") + return + } + defer release() + if err := run(ctx); err != nil { + log.Error(ctx, "Error running "+job, err) + } +} + // scheduleArtworkHousekeeping registers the recurring missing-state and prune jobs, and // reports an artwork config change without acting on it. func scheduleArtworkHousekeeping(ctx context.Context, worker *artwork.Worker) func() error { @@ -386,26 +402,20 @@ func scheduleArtworkHousekeeping(ctx context.Context, worker *artwork.Worker) fu schedulerInstance := scheduler.GetInstance() if _, err := schedulerInstance.Add(consts.ArtworkEnqueueMissingSchedule, func() { - if err := worker.EnqueueMissingAll(ctx); err != nil { - log.Error(ctx, "Error enqueueing missing artwork rechecks", err) - } + outsideScan(ctx, "artwork missing-state recheck", worker.EnqueueMissingAll) }); err != nil { log.Error(ctx, "Error scheduling artwork missing-state recheck", err) } if _, err := schedulerInstance.Add(consts.ArtworkPruneSchedule, func() { - if err := worker.RunPrune(ctx); err != nil { - log.Error(ctx, "Error running artwork prune", err) - } + outsideScan(ctx, "artwork prune", worker.RunPrune) }); err != nil { log.Error(ctx, "Error scheduling artwork prune", err) } // Also run the missing-row recheck once at startup so a never-scanned entity is picked up // immediately, not only on the next hourly tick (e.g. after enabling the feature). - if err := worker.EnqueueMissingAll(ctx); err != nil { - log.Error(ctx, "Error enqueueing missing artwork rechecks", err) - } + outsideScan(ctx, "artwork missing-state recheck", worker.EnqueueMissingAll) if err := worker.ReconcileConfig(ctx); err != nil { log.Error(ctx, "Error checking the artwork config fingerprint", err) diff --git a/core/artwork/worker.go b/core/artwork/worker.go index bb09be55e..e8fb3a11c 100644 --- a/core/artwork/worker.go +++ b/core/artwork/worker.go @@ -46,6 +46,7 @@ type Worker struct { pruneMu sync.RWMutex pools []*drainPool runCtx context.Context + paused func() bool gatesMu sync.Mutex gates map[string]*extGate @@ -59,6 +60,7 @@ func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg broker: broker, pools: newDrainPools(), runCtx: context.Background(), + paused: func() bool { return false }, gates: map[string]*extGate{}, } w.proc.resolver = newResolver(ds, ag, ffmpeg, w.gate) @@ -90,6 +92,11 @@ var ( } ) +// PauseWhile holds off queue draining whenever paused reports true. Call it before Run. +func (w *Worker) PauseWhile(paused func() bool) { + w.paused = paused +} + // Run blocks draining the queue until ctx is cancelled. func (w *Worker) Run(ctx context.Context) error { w.runCtx = ctx @@ -143,6 +150,9 @@ func (w *Worker) EnqueueMissingAll(ctx context.Context) error { } func (w *Worker) drain(ctx context.Context, concurrency int, kinds ...string) (int, error) { + if w.paused() { + return 0, nil + } // Dequeue well past the pool size so a slow external lookup never idles the other slots. // DequeueBatch does not mark rows taken, so this is one query per pass, not per slot. items, err := w.proc.ds.ArtworkQueue(ctx).DequeueBatch(max(16, 4*concurrency), kinds...) @@ -170,6 +180,9 @@ func (w *Worker) drain(ctx context.Context, concurrency int, kinds ...string) (i wg.Wait() return len(items), nil //nolint:nilerr // a cancelled drain is a clean stop, not an error } + if w.paused() { + break + } wg.Go(func() { defer func() { <-sem }() out, got := w.process(ctx, item) diff --git a/core/artwork/worker_test.go b/core/artwork/worker_test.go index ebb8de251..d74e3a5ed 100644 --- a/core/artwork/worker_test.go +++ b/core/artwork/worker_test.go @@ -866,6 +866,34 @@ var _ = Describe("Worker", func() { } }) + It("stops dispatching and leaves the rest queued when paused mid-batch", func() { + folderRepo.result = []model.Folder{{ + Path: "tests/fixtures/artist/an-album", + ImageFiles: []string{"cover.jpg"}, + }} + albums := model.Albums{} + for i := range 8 { + id := fmt.Sprintf("alp%d", i) + albums = append(albums, model.Album{ID: id, Name: "Album", FolderIDs: []string{"f1"}}) + Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ + ItemKind: "al", ItemID: id, Priority: model.ArtworkPriorityScan, + })).To(Succeed()) + } + ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(albums) + // Pauses as soon as the first item has left the queue. + w.PauseWhile(func() bool { + n, _ := queueRepo.Count() + return n < 8 + }) + + _, err := w.drain(ctx, 1) + Expect(err).ToNot(HaveOccurred()) + + count, err := queueRepo.Count() + Expect(err).ToNot(HaveOccurred()) + Expect(count).To(Equal(int64(7)), "only the item dispatched before the pause may leave the queue") + }) + It("dequeues past the worker pool so one drain covers many items", func() { for i := range 16 { ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{{ID: fmt.Sprintf("alb%d", i), Name: "Album"}}) @@ -896,6 +924,30 @@ var _ = Describe("Worker", func() { Eventually(done, time.Second).Should(Receive(BeNil())) }) + It("does not drain the queue while paused", func() { + folderRepo.result = []model.Folder{{ + Path: "tests/fixtures/artist/an-album", + ImageFiles: []string{"cover.jpg"}, + }} + ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{ + {ID: "al1", Name: "Album", FolderIDs: []string{"f1"}}, + }) + Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ + ItemKind: "al", ItemID: "al1", Priority: model.ArtworkPriorityScan, + })).To(Succeed()) + w.PauseWhile(func() bool { return true }) + + runCtx, cancel := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- w.Run(runCtx) }() + DeferCleanup(func() { + cancel() + Eventually(done, 2*time.Second).Should(Receive(BeNil())) + }) + + Consistently(func() any { return findQueued(queueRepo, "al", "al1") }, 300*time.Millisecond).ShouldNot(BeNil()) + }) + It("does not leak goroutines after Run exits", func() { DeferCleanup(configtest.SetupConfig()) diff --git a/db/db.go b/db/db.go index 685886edb..a9c6c4a15 100644 --- a/db/db.go +++ b/db/db.go @@ -133,6 +133,12 @@ func ErrorCodes(err error) (code, extended int, ok bool) { return int(se.Code), int(se.ExtendedCode), true } +// IsBusy reports whether err is SQLITE_BUSY, including BUSY_SNAPSHOT, which only a new transaction clears. +func IsBusy(err error) bool { + code, _, ok := ErrorCodes(err) + return ok && code == int(sqlite3.ErrBusy) +} + type statusLogger struct{ numPending int } func (*statusLogger) Fatalf(format string, v ...any) { log.Fatal(fmt.Sprintf(format, v...)) } diff --git a/db/db_test.go b/db/db_test.go index 2ce01dc3d..e3a52e1fb 100644 --- a/db/db_test.go +++ b/db/db_test.go @@ -3,8 +3,12 @@ package db_test import ( "context" "database/sql" + "errors" + "fmt" "testing" + "github.com/mattn/go-sqlite3" + "github.com/navidrome/navidrome/db" "github.com/navidrome/navidrome/log" "github.com/navidrome/navidrome/tests" @@ -19,6 +23,18 @@ func TestDB(t *testing.T) { RunSpecs(t, "DB Suite") } +var _ = DescribeTable("IsBusy", + func(err error, expected bool) { + Expect(db.IsBusy(err)).To(Equal(expected)) + }, + Entry("SQLITE_BUSY", sqlite3.Error{Code: sqlite3.ErrBusy}, true), + Entry("SQLITE_BUSY_SNAPSHOT", sqlite3.Error{Code: sqlite3.ErrBusy, ExtendedCode: sqlite3.ErrBusySnapshot}, true), + Entry("a wrapped SQLITE_BUSY", fmt.Errorf("persisting: %w", sqlite3.Error{Code: sqlite3.ErrBusy}), true), + Entry("another SQLite error", sqlite3.Error{Code: sqlite3.ErrConstraint}, false), + Entry("a non-SQLite error", errors.New("database is locked"), false), + Entry("nil", nil, false), +) + var _ = Describe("IsSchemaEmpty", func() { var database *sql.DB var ctx context.Context diff --git a/db/optimize.go b/db/optimize.go index f46906c4e..aff36e8fd 100644 --- a/db/optimize.go +++ b/db/optimize.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "strconv" + "strings" "sync" "time" @@ -134,16 +135,65 @@ func optimizeAt(ctx context.Context, db *sql.DB, now time.Time) error { return recordAnalyzeError(ctx, db, now, fmt.Errorf("marking ANALYZE pending: %w", err)) } log.Debug(ctx, "Refreshing query planner statistics") - _, err := db.ExecContext(ctx, "ANALYZE") - if err != nil { + if err := analyzeInSteps(ctx, db); err != nil { return recordAnalyzeError(ctx, db, now, fmt.Errorf("running ANALYZE: %w", err)) } - if err = recordAnalyzeSuccess(ctx, db, now); err != nil { + if err := recordAnalyzeSuccess(ctx, db, now); err != nil { return recordAnalyzeError(ctx, db, now, err) } return nil } +// One ANALYZE per index (whole table if WITHOUT ROWID or lacking a non-partial index) yields the +// same sqlite_stat1 rows as a full ANALYZE, but frees the write lock between steps. +const analyzeTargetsSQL = ` +SELECT i.name FROM sqlite_schema i JOIN pragma_table_list t ON t.schema = 'main' AND t.name = i.tbl_name +WHERE i.type = 'index' AND t.wr = 0 AND EXISTS (SELECT 1 FROM pragma_index_list(t.name) l WHERE l.partial = 0) +UNION ALL +SELECT t.name FROM pragma_table_list t +WHERE t.schema = 'main' AND t.type IN ('table', 'shadow') AND t.name NOT LIKE 'sqlite_%' + AND (t.wr = 1 OR NOT EXISTS (SELECT 1 FROM pragma_index_list(t.name) l WHERE l.partial = 0))` + +// analyzeMaxYield is just above SQLite's longest busy-handler sleep, so every waiting writer +// retries during the pause. +const analyzeMaxYield = 150 * time.Millisecond + +func analyzeInSteps(ctx context.Context, db *sql.DB) error { + targets, err := analyzeTargets(ctx, db) + if err != nil { + return err + } + for _, target := range targets { + start := time.Now() + if _, err := db.ExecContext(ctx, `ANALYZE "`+strings.ReplaceAll(target, `"`, `""`)+`"`); err != nil { + return fmt.Errorf("analyzing %s: %w", target, err) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(min(time.Since(start), analyzeMaxYield)): + } + } + return nil +} + +func analyzeTargets(ctx context.Context, db *sql.DB) ([]string, error) { + rows, err := db.QueryContext(ctx, analyzeTargetsSQL) + if err != nil { + return nil, fmt.Errorf("listing ANALYZE targets: %w", err) + } + defer rows.Close() + var targets []string + for rows.Next() { + var name string + if err := rows.Scan(&name); err != nil { + return nil, fmt.Errorf("listing ANALYZE targets: %w", err) + } + targets = append(targets, name) + } + return targets, rows.Err() +} + func recordAnalyzeSuccess(ctx context.Context, db *sql.DB, now time.Time) error { tx, err := db.BeginTx(ctx, nil) if err != nil { diff --git a/db/optimize_test.go b/db/optimize_test.go index da9b3b9c9..6ad9b9cb4 100644 --- a/db/optimize_test.go +++ b/db/optimize_test.go @@ -75,6 +75,43 @@ var _ = Describe("Optimize", func() { Expect(getProperty(consts.DBAnalyzePendingKey)).To(Equal("0")) }) + It("produces the same statistics as a single full ANALYZE", func() { + putProperty(consts.DBAnalyzePendingKey, "1") + for _, stmt := range []string{ + "create table unindexed(id integer primary key, v int)", + "insert into unindexed(v) select flag from analyze_probe", + "create table no_rowid(k text primary key, v int) without rowid", + "insert into no_rowid select 'k' || id, id % 7 from analyze_probe", + "create index no_rowid_v on no_rowid(v)", + "create table partial_only(id integer primary key, v int)", + "insert into partial_only(v) select id % 5 from analyze_probe", + "create index partial_only_v on partial_only(v) where v = 1", + "analyze", + } { + _, err := database.Exec(stmt) + Expect(err).ToNot(HaveOccurred()) + } + statRows := func() []string { + rows, err := database.Query("select tbl || '|' || coalesce(idx, '') || '|' || stat from sqlite_stat1 order by 1") + Expect(err).ToNot(HaveOccurred()) + defer rows.Close() + var res []string + for rows.Next() { + var s string + Expect(rows.Scan(&s)).To(Succeed()) + res = append(res, s) + } + return res + } + fullAnalyze := statRows() + _, err := database.Exec("delete from sqlite_stat1") + Expect(err).ToNot(HaveOccurred()) + + Expect(db.OptimizeDBAt(ctx, database, now)).To(Succeed()) + + Expect(statRows()).To(Equal(fullAnalyze)) + }) + It("runs when no previous analysis was recorded", func() { ran, err := db.OptimizeDBIfNeeded(ctx, database, now) Expect(err).ToNot(HaveOccurred()) diff --git a/model/datastore.go b/model/datastore.go index 273ca714b..26687d5d4 100644 --- a/model/datastore.go +++ b/model/datastore.go @@ -47,5 +47,8 @@ type DataStore interface { WithTx(block func(tx DataStore) error, scope ...string) error WithTxImmediate(block func(tx DataStore) error, scope ...string) error + // WithTxRetry runs block in a transaction, rerunning it while SQLite reports the database busy. + // For background work only (it can take minutes), and block must be safe to rerun after a rollback. + WithTxRetry(ctx context.Context, block func(ctx context.Context, tx DataStore) error, scope ...string) error GC(ctx context.Context, libraryIDs ...int) error } diff --git a/persistence/persistence.go b/persistence/persistence.go index 9d3a33cfc..589812266 100644 --- a/persistence/persistence.go +++ b/persistence/persistence.go @@ -3,6 +3,7 @@ package persistence import ( "context" "database/sql" + "fmt" "reflect" "time" @@ -138,11 +139,15 @@ func (s *SQLStore) Resource(ctx context.Context, m any) model.ResourceRepository return nil } -func (s *SQLStore) WithTx(block func(tx model.DataStore) error, scope ...string) error { - var msg string +func scopeLabel(scope []string) string { if len(scope) > 0 { - msg = scope[0] + return scope[0] } + return "" +} + +func (s *SQLStore) WithTx(block func(tx model.DataStore) error, scope ...string) error { + msg := scopeLabel(scope) start := time.Now() conn, inTx := s.db.(*dbx.DB) if !inTx { @@ -177,6 +182,51 @@ func (s *SQLStore) WithTxImmediate(block func(tx model.DataStore) error, scope . }, scope...) } +// txRetryDelay spaces out reruns of a busy transaction. Each attempt has already waited out the +// busy timeout, so WithTxRetry gives up only after a sustained lock. +var txRetryDelay = 5 * time.Second + +const txMaxRetries = 3 + +func (s *SQLStore) WithTxRetry(ctx context.Context, block func(ctx context.Context, tx model.DataStore) error, scope ...string) error { + // Inside a transaction, join it: the outer one holds the lock and owns commit and rollback + if _, ok := s.db.(*dbx.DB); !ok { + return block(ctx, s) + } + for attempt := 0; ; attempt++ { + attemptCtx := ctx + if attempt < txMaxRetries { + attemptCtx = withBusyRetry(ctx) + } + err := s.WithTx(func(tx model.DataStore) error { return block(attemptCtx, tx) }, scope...) + if attempt == txMaxRetries || !db.IsBusy(err) { + return err + } + log.Warn(ctx, "Database busy, retrying transaction", "scope", scopeLabel(scope), "attempt", attempt+1, err) + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(time.Duration(attempt+1) * txRetryDelay): + } + } +} + +type busyRetryKey struct{} + +// withBusyRetry marks a transaction attempt that WithTxRetry will rerun, so a busy statement in it +// is logged as a warning rather than an error. +func withBusyRetry(ctx context.Context) context.Context { + return context.WithValue(ctx, busyRetryKey{}, true) +} + +func hasBusyRetry(ctx context.Context) bool { + if ctx == nil { + return false + } + retry, _ := ctx.Value(busyRetryKey{}).(bool) + return retry +} + func (s *SQLStore) GC(ctx context.Context, libraryIDs ...int) error { trace := func(ctx context.Context, msg string, f func() error) func() error { return func() error { @@ -207,9 +257,9 @@ func (s *SQLStore) GC(ctx context.Context, libraryIDs ...int) error { trace(ctx, "remove orphan playlist tracks", func() error { return s.Playlist(ctx).(*playlistRepository).removeOrphans() }), ) if err != nil { - log.Error(ctx, "Error tidying up database", err) + return fmt.Errorf("tidying up database: %w", err) } - return err + return nil } func (s *SQLStore) getDBXBuilder() dbx.Builder { diff --git a/persistence/persistence_test.go b/persistence/persistence_test.go index 13e56bde1..e13f6a231 100644 --- a/persistence/persistence_test.go +++ b/persistence/persistence_test.go @@ -2,7 +2,10 @@ package persistence import ( "context" + "errors" + "time" + "github.com/mattn/go-sqlite3" "github.com/navidrome/navidrome/db" "github.com/navidrome/navidrome/model" . "github.com/onsi/ginkgo/v2" @@ -55,4 +58,74 @@ var _ = Describe("SQLStore", func() { }) }) }) + + Describe("WithTxRetry", func() { + busy := sqlite3.Error{Code: sqlite3.ErrBusy} + BeforeEach(func() { + DeferCleanup(func(d time.Duration) { txRetryDelay = d }, txRetryDelay) + txRetryDelay = 0 + }) + + It("reruns a busy transaction from a clean rollback", func() { + var attempts []bool + err := ds.WithTxRetry(ctx, func(ctx context.Context, tx model.DataStore) error { + attempts = append(attempts, hasBusyRetry(ctx)) + Expect(tx.Property(ctx).Put("retry-key", "attempt")).To(Succeed()) + if len(attempts) < 3 { + return busy + } + return nil + }) + Expect(err).ToNot(HaveOccurred()) + Expect(attempts).To(Equal([]bool{true, true, true})) + Expect(ds.Property(ctx).Get("retry-key")).To(Equal("attempt")) + }) + + It("gives up after the last retry, which is not marked as retried", func() { + var attempts []bool + err := ds.WithTxRetry(ctx, func(ctx context.Context, _ model.DataStore) error { + attempts = append(attempts, hasBusyRetry(ctx)) + return busy + }) + Expect(db.IsBusy(err)).To(BeTrue()) + Expect(attempts).To(Equal([]bool{true, true, true, false})) + }) + + It("does not rerun on other errors", func() { + calls := 0 + err := ds.WithTxRetry(ctx, func(context.Context, model.DataStore) error { + calls++ + return sqlite3.Error{Code: sqlite3.ErrConstraint} + }) + Expect(err).To(HaveOccurred()) + Expect(calls).To(Equal(1)) + }) + + It("does not rerun when called inside a transaction", func() { + calls := 0 + err := ds.WithTx(func(tx model.DataStore) error { + return tx.WithTxRetry(ctx, func(context.Context, model.DataStore) error { + calls++ + return busy + }) + }) + Expect(db.IsBusy(err)).To(BeTrue()) + Expect(calls).To(Equal(1)) + }) + + It("joins the enclosing transaction instead of opening another", func() { + rollback := errors.New("rollback") + err := ds.WithTx(func(tx model.DataStore) error { + Expect(tx.Property(ctx).Put("outer-key", "v")).To(Succeed()) + Expect(tx.WithTxRetry(ctx, func(ctx context.Context, inner model.DataStore) error { + Expect(inner.Property(ctx).Get("outer-key")).To(Equal("v")) + return inner.Property(ctx).Put("inner-key", "v") + })).To(Succeed()) + return rollback + }) + Expect(err).To(MatchError(rollback)) + _, err = ds.Property(ctx).Get("inner-key") + Expect(err).To(MatchError(model.ErrNotFound)) + }) + }) }) diff --git a/persistence/sql_base_repository.go b/persistence/sql_base_repository.go index bc841db03..026e42b03 100644 --- a/persistence/sql_base_repository.go +++ b/persistence/sql_base_repository.go @@ -643,5 +643,9 @@ func (r sqlRepository) logSQL(sql string, args dbx.Params, err error, rowsAffect if code, extended, ok := db.ErrorCodes(err); ok { fields = append(fields, "sqliteCode", code, "sqliteExtended", extended) } + if db.IsBusy(err) && hasBusyRetry(r.ctx) { + log.Warn(append(fields, err)...) + return + } log.Error(append(fields, err)...) } diff --git a/scanner/phase_1_folders.go b/scanner/phase_1_folders.go index 6107b3316..feefde032 100644 --- a/scanner/phase_1_folders.go +++ b/scanner/phase_1_folders.go @@ -45,7 +45,9 @@ func createPhaseFolders(ctx context.Context, state *scanState, ds model.DataStor jobs = append(jobs, job) } - return &phaseFolders{jobs: jobs, ctx: ctx, ds: ds, state: state, imageChanges: &imageChangeCollector{ds: ds}} + walkCtx, stopWalk := context.WithCancelCause(ctx) + return &phaseFolders{jobs: jobs, ctx: ctx, walkCtx: walkCtx, stopWalk: stopWalk, ds: ds, state: state, + imageChanges: &imageChangeCollector{ds: ds}} } type scanJob struct { @@ -123,6 +125,8 @@ type phaseFolders struct { jobs []*scanJob ds model.DataStore ctx context.Context + walkCtx context.Context // cancelled when a folder fails to persist, so the walk stops early + stopWalk context.CancelCauseFunc state *scanState prevAlbumPIDConf string imageChanges *imageChangeCollector @@ -144,15 +148,15 @@ func (p *phaseFolders) producer() ppl.Producer[*folderEntry] { var total int64 var totalChanged int64 for _, job := range p.jobs { - if utils.IsCtxDone(p.ctx) { + if utils.IsCtxDone(p.walkCtx) { break } - outputChan, err := walkDirTree(p.ctx, job, job.targetFolders...) + outputChan, err := walkDirTree(p.walkCtx, job, job.targetFolders...) if err != nil { log.Warn(p.ctx, "Scanner: Error scanning library", "lib", job.lib.Name, err) } - for folder := range pl.ReadOrDone(p.ctx, outputChan) { + for folder := range pl.ReadOrDone(p.walkCtx, outputChan) { job.numFolders.Add(1) p.state.sendProgress(&ProgressInfo{ LibID: job.lib.ID, @@ -208,6 +212,9 @@ func (p *phaseFolders) stages() []ppl.Stage[*folderEntry] { func (p *phaseFolders) processFolder(entry *folderEntry) (*folderEntry, error) { defer p.measure(entry)() + if err := context.Cause(p.walkCtx); err != nil { + return entry, err + } // Load children mediafiles from DB cursor, err := p.ds.MediaFile(p.ctx).GetCursor(model.QueryOptions{ @@ -331,128 +338,129 @@ func (p *phaseFolders) persistChanges(entry *folderEntry) (*folderEntry, error) defer p.measure(entry)() p.state.changesDetected.Store(true) - // Collect artwork queue items for changed albums/artists, enqueued in the same transaction - var queueItems []model.ArtworkQueueItem - - err := p.ds.WithTx(func(tx model.DataStore) error { - // Instantiate all repositories just once per folder - folderRepo := tx.Folder(p.ctx) - tagRepo := tx.Tag(p.ctx) - artistRepo := tx.Artist(p.ctx) - libraryRepo := tx.Library(p.ctx) - albumRepo := tx.Album(p.ctx) - mfRepo := tx.MediaFile(p.ctx) - - // A new folder's albums/artists are enqueued below; only pre-existing folders need the diff. - if !entry.isNew() { - if changed, artistImage := entry.imagesChanged(); changed { - p.imageChanges.record(entry.job.lib, imageChangedFolder{ - id: entry.id, path: entry.path, artistImage: artistImage, - }) - } - } - - // Save folder to DB - folder := entry.toFolder() - err := folderRepo.Put(folder) - if err != nil { - log.Error(p.ctx, "Scanner: Error persisting folder to DB", "folder", entry.path, err) - return err - } - - // Save all tags to DB - err = tagRepo.Add(entry.job.lib.ID, entry.tags...) - if err != nil { - log.Error(p.ctx, "Scanner: Error persisting tags to DB", "folder", entry.path, err) - return err - } - - // Save all new/modified artists to DB. Their information will be incomplete, but they will be refreshed later - for i := range entry.artists { - err = artistRepo.Put(&entry.artists[i], "name", - "mbz_artist_id", "sort_artist_name", "order_artist_name", "full_text", "search_normalized", "updated_at") - if err != nil { - log.Error(p.ctx, "Scanner: Error persisting artist to DB", "folder", entry.path, "artist", entry.artists[i].Name, err) - return err - } - err = libraryRepo.AddArtist(entry.job.lib.ID, entry.artists[i].ID) - if err != nil { - log.Error(p.ctx, "Scanner: Error adding artist to library", "lib", entry.job.lib.ID, "artist", entry.artists[i].Name, err) - return err - } - if entry.artists[i].Name != consts.UnknownArtist && entry.artists[i].Name != consts.VariousArtists { - queueItems = append(queueItems, scanArtworkItem(model.KindArtistArtwork, entry.artists[i].ID)) - } - } - - // Save all new/modified albums to DB. Their information will be incomplete, but they will be refreshed later - for i := range entry.albums { - err = p.persistAlbum(albumRepo, &entry.albums[i], entry.albumIDMap) - if err != nil { - log.Error(p.ctx, "Scanner: Error persisting album to DB", "folder", entry.path, "album", entry.albums[i], err) - return err - } - if entry.albums[i].Name != consts.UnknownAlbum { - queueItems = append(queueItems, scanArtworkItem(model.KindAlbumArtwork, entry.albums[i].ID)) - } - } - - // Save all tracks to DB - for i := range entry.tracks { - err = mfRepo.Put(&entry.tracks[i]) - if err != nil { - log.Error(p.ctx, "Scanner: Error persisting mediafile to DB", "folder", entry.path, "track", entry.tracks[i], err) - return err - } - } - - // A re-imported track returns to unresolved so new embedded art is picked up lazily. - if len(entry.tracks) > 0 { - trackIDs := slice.Map(entry.tracks, func(t model.MediaFile) string { return t.ID }) - if err := tx.Artwork(p.ctx).DeleteForItems(model.KindMediaFileArtwork, trackIDs); err != nil { - log.Warn(p.ctx, "Scanner: could not invalidate media_file artwork", "folder", entry.path, err) - } - } - - // Mark all missing tracks as not available - if len(entry.missingTracks) > 0 { - err = mfRepo.MarkMissing(true, entry.missingTracks...) - if err != nil { - log.Error(p.ctx, "Scanner: Error marking missing tracks", "folder", entry.path, err) - return err - } - - // Touch all albums that have missing tracks, so they get refreshed in later phases - groupedMissingTracks := slice.ToMap(entry.missingTracks, func(mf *model.MediaFile) (string, struct{}) { - return mf.AlbumID, struct{}{} - }) - albumsToUpdate := slices.Collect(maps.Keys(groupedMissingTracks)) - err = albumRepo.Touch(albumsToUpdate...) - if err != nil { - log.Error(p.ctx, "Scanner: Error touching album", "folder", entry.path, "albums", albumsToUpdate, err) - return err - } - } - - // Enqueue artwork resolution for changed albums/artists. Never fails the scan. - // A full scan re-imports every track, so a re-import is no evidence the art changed. - if len(queueItems) > 0 { - queue := tx.ArtworkQueue(p.ctx) - enqueue := queue.Enqueue - if p.state.fullScan { - enqueue = queue.EnqueueIfMissing - } - if err := enqueue(queueItems...); err != nil { - log.Warn(p.ctx, "Scanner: could not enqueue artwork resolution", "folder", entry.path, err) - } - } - return nil + ctx := log.NewContext(p.ctx, "folder", entry.path) + err := p.ds.WithTxRetry(ctx, func(ctx context.Context, tx model.DataStore) error { + return p.persistFolder(ctx, tx, entry) }, "scanner: persist changes") if err != nil { - log.Error(p.ctx, "Scanner: Error persisting changes to DB", "folder", entry.path, err) + log.Error(ctx, "Scanner: Error persisting changes to DB", err) + p.stopWalk(err) + return entry, err } - return entry, err + // A new folder's albums/artists were enqueued with it; only pre-existing folders need the diff. + if !entry.isNew() { + if changed, artistImage := entry.imagesChanged(); changed { + p.imageChanges.record(entry.job.lib, imageChangedFolder{ + id: entry.id, path: entry.path, artistImage: artistImage, + }) + } + } + return entry, nil +} + +// persistFolder writes the folder in tx. WithTxRetry may rerun it after a rollback. +func (p *phaseFolders) persistFolder(ctx context.Context, tx model.DataStore, entry *folderEntry) error { + // Collect artwork queue items for changed albums/artists, enqueued in the same transaction + var queueItems []model.ArtworkQueueItem + // persistAlbum consumes the map, so a rerun needs the original + albumIDMap := maps.Clone(entry.albumIDMap) + + // Instantiate all repositories just once per folder + folderRepo := tx.Folder(ctx) + tagRepo := tx.Tag(ctx) + artistRepo := tx.Artist(ctx) + libraryRepo := tx.Library(ctx) + albumRepo := tx.Album(ctx) + mfRepo := tx.MediaFile(ctx) + + // Save folder to DB + folder := entry.toFolder() + err := folderRepo.Put(folder) + if err != nil { + return fmt.Errorf("persisting folder: %w", err) + } + + // Save all tags to DB + err = tagRepo.Add(entry.job.lib.ID, entry.tags...) + if err != nil { + return fmt.Errorf("persisting tags: %w", err) + } + + // Save all new/modified artists to DB. Their information will be incomplete, but they will be refreshed later + for i := range entry.artists { + err = artistRepo.Put(&entry.artists[i], "name", + "mbz_artist_id", "sort_artist_name", "order_artist_name", "full_text", "search_normalized", "updated_at") + if err != nil { + return fmt.Errorf("persisting artist %q: %w", entry.artists[i].Name, err) + } + err = libraryRepo.AddArtist(entry.job.lib.ID, entry.artists[i].ID) + if err != nil { + return fmt.Errorf("adding artist %q to library: %w", entry.artists[i].Name, err) + } + if entry.artists[i].Name != consts.UnknownArtist && entry.artists[i].Name != consts.VariousArtists { + queueItems = append(queueItems, scanArtworkItem(model.KindArtistArtwork, entry.artists[i].ID)) + } + } + + // Save all new/modified albums to DB. Their information will be incomplete, but they will be refreshed later + for i := range entry.albums { + err = p.persistAlbum(albumRepo, &entry.albums[i], albumIDMap) + if err != nil { + return err + } + if entry.albums[i].Name != consts.UnknownAlbum { + queueItems = append(queueItems, scanArtworkItem(model.KindAlbumArtwork, entry.albums[i].ID)) + } + } + + // Save all tracks to DB + for i := range entry.tracks { + err = mfRepo.Put(&entry.tracks[i]) + if err != nil { + return fmt.Errorf("persisting track %q: %w", entry.tracks[i].Path, err) + } + } + + // A re-imported track returns to unresolved so new embedded art is picked up lazily. + if len(entry.tracks) > 0 { + trackIDs := slice.Map(entry.tracks, func(t model.MediaFile) string { return t.ID }) + if err := tx.Artwork(ctx).DeleteForItems(model.KindMediaFileArtwork, trackIDs); err != nil { + log.Warn(ctx, "Scanner: could not invalidate media_file artwork", err) + } + } + + // Mark all missing tracks as not available + if len(entry.missingTracks) > 0 { + err = mfRepo.MarkMissing(true, entry.missingTracks...) + if err != nil { + return fmt.Errorf("marking missing tracks: %w", err) + } + + // Touch all albums that have missing tracks, so they get refreshed in later phases + groupedMissingTracks := slice.ToMap(entry.missingTracks, func(mf *model.MediaFile) (string, struct{}) { + return mf.AlbumID, struct{}{} + }) + albumsToUpdate := slices.Collect(maps.Keys(groupedMissingTracks)) + err = albumRepo.Touch(albumsToUpdate...) + if err != nil { + return fmt.Errorf("touching albums %v: %w", albumsToUpdate, err) + } + } + + // Enqueue artwork resolution for changed albums/artists. Never fails the scan. + // A full scan re-imports every track, so a re-import is no evidence the art changed. + if len(queueItems) > 0 { + queue := tx.ArtworkQueue(ctx) + enqueue := queue.Enqueue + if p.state.fullScan { + enqueue = queue.EnqueueIfMissing + } + if err := enqueue(queueItems...); err != nil { + log.Warn(ctx, "Scanner: could not enqueue artwork resolution", err) + } + } + return nil } // persistAlbum persists the given album to the database, and reassigns annotations from the previous album ID @@ -499,34 +507,32 @@ func (p *phaseFolders) logFolder(entry *folderEntry) (*folderEntry, error) { } func (p *phaseFolders) finalize(err error) error { - errF := p.ds.WithTx(func(tx model.DataStore) error { + p.stopWalk(nil) + defer p.imageChanges.enqueue(p.ctx) + // A failed phase may not have walked every folder, and unvisited ones must not be marked missing + if err != nil { + return err + } + return p.ds.WithTxRetry(p.ctx, func(ctx context.Context, tx model.DataStore) error { for _, job := range p.jobs { // Mark all folders that were not updated as missing if len(job.lastUpdates) == 0 { continue } folderIDs := slices.Collect(maps.Keys(job.lastUpdates)) - err := tx.Folder(p.ctx).MarkMissing(true, folderIDs...) - if err != nil { - log.Error(p.ctx, "Scanner: Error marking missing folders", "lib", job.lib.Name, err) - return err + if err := tx.Folder(ctx).MarkMissing(true, folderIDs...); err != nil { + return fmt.Errorf("marking missing folders in %s: %w", job.lib.Name, err) } - err = tx.MediaFile(p.ctx).MarkMissingByFolder(true, folderIDs...) - if err != nil { - log.Error(p.ctx, "Scanner: Error marking tracks in missing folders", "lib", job.lib.Name, err) - return err + if err := tx.MediaFile(ctx).MarkMissingByFolder(true, folderIDs...); err != nil { + return fmt.Errorf("marking tracks in missing folders in %s: %w", job.lib.Name, err) } // Touch all albums that have missing folders, so they get refreshed in later phases - _, err = tx.Album(p.ctx).TouchByMissingFolder() - if err != nil { - log.Error(p.ctx, "Scanner: Error touching albums with missing folders", "lib", job.lib.Name, err) - return err + if _, err := tx.Album(ctx).TouchByMissingFolder(); err != nil { + return fmt.Errorf("touching albums with missing folders in %s: %w", job.lib.Name, err) } } return nil }, "scanner: finalize phaseFolders") - p.imageChanges.enqueue(p.ctx) - return errors.Join(err, errF) } var _ phase[*folderEntry] = (*phaseFolders)(nil) diff --git a/scanner/phase_2_missing_tracks.go b/scanner/phase_2_missing_tracks.go index 8c258b833..6ccc9a46c 100644 --- a/scanner/phase_2_missing_tracks.go +++ b/scanner/phase_2_missing_tracks.go @@ -273,66 +273,68 @@ func (p *phaseMissingTracks) findCrossLibraryMatch(missing model.MediaFile) (mod } func (p *phaseMissingTracks) moveMatched(target, missing model.MediaFile) error { - return p.ds.WithTx(func(tx model.DataStore) error { - discardedID := target.ID - oldAlbumID := missing.AlbumID - newAlbumID := target.AlbumID + oldAlbumID := missing.AlbumID + newAlbumID := target.AlbumID + // Use newAlbumID as key since we only care about avoiding duplicate reassignments to the same target. + // Claimed before the transaction so a concurrent move skips it, and released if the move fails. + reassignAlbum := oldAlbumID != newAlbumID + if reassignAlbum { + p.annotationMutex.Lock() + reassignAlbum = !p.processedAlbumAnnotations[newAlbumID] + p.processedAlbumAnnotations[newAlbumID] = true + p.annotationMutex.Unlock() + if !reassignAlbum { + log.Trace(p.ctx, "Scanner: Skipping album annotation reassignment", "from", oldAlbumID, "to", newAlbumID) + } + } + err := p.ds.WithTxRetry(p.ctx, func(ctx context.Context, tx model.DataStore) error { + // A rerun must start from the original target, not the one the rolled-back attempt changed + moved := target // Preserve the original created_at from the missing file, so moved tracks // don't appear in "Recently Added" - target.CreatedAt = missing.CreatedAt + moved.CreatedAt = missing.CreatedAt // Update the target media file with the missing file's ID. This effectively "moves" the track // to the new location while keeping its annotations and references intact. - target.ID = missing.ID - err := tx.MediaFile(p.ctx).Put(&target) - if err != nil { + moved.ID = missing.ID + if err := tx.MediaFile(ctx).Put(&moved); err != nil { return fmt.Errorf("update matched track: %w", err) } // Discard the new mediafile row (the one that was moved to) - err = tx.MediaFile(p.ctx).Delete(discardedID) - if err != nil { + if err := tx.MediaFile(ctx).Delete(target.ID); err != nil { return fmt.Errorf("delete discarded track: %w", err) } - // Handle album annotation reassignment if AlbumID changed - if oldAlbumID != newAlbumID { - // Use newAlbumID as key since we only care about avoiding duplicate reassignments to the same target - p.annotationMutex.RLock() - alreadyProcessed := p.processedAlbumAnnotations[newAlbumID] - p.annotationMutex.RUnlock() - - if !alreadyProcessed { - p.annotationMutex.Lock() - // Double-check pattern to avoid race conditions - if !p.processedAlbumAnnotations[newAlbumID] { - // Reassign direct album annotations (starred, rating) - log.Debug(p.ctx, "Scanner: Reassigning album annotations", "from", oldAlbumID, "to", newAlbumID) - if err := tx.Album(p.ctx).ReassignAnnotation(oldAlbumID, newAlbumID); err != nil { - log.Warn(p.ctx, "Scanner: Could not reassign album annotations", "from", oldAlbumID, "to", newAlbumID, err) - } - - // Keep created_at field from previous instance of the album, so moved albums - // don't appear in "Recently Added" - if err := tx.Album(p.ctx).CopyAttributes(oldAlbumID, newAlbumID, "created_at"); err != nil { - if !errors.Is(err, model.ErrNotFound) { - log.Warn(p.ctx, "Scanner: Could not copy album created_at", "from", oldAlbumID, "to", newAlbumID, err) - } - } - - // Note: RefreshPlayCounts will be called in later phases, so we don't need to call it here - p.processedAlbumAnnotations[newAlbumID] = true - } - p.annotationMutex.Unlock() - } else { - log.Trace(p.ctx, "Scanner: Skipping album annotation reassignment", "from", oldAlbumID, "to", newAlbumID) + if reassignAlbum { + // Reassign direct album annotations (starred, rating) + log.Debug(ctx, "Scanner: Reassigning album annotations", "from", oldAlbumID, "to", newAlbumID) + if err := tx.Album(ctx).ReassignAnnotation(oldAlbumID, newAlbumID); err != nil { + log.Warn(ctx, "Scanner: Could not reassign album annotations", "from", oldAlbumID, "to", newAlbumID, err) } - } - p.state.changesDetected.Store(true) + // Keep created_at field from previous instance of the album, so moved albums + // don't appear in "Recently Added" + if err := tx.Album(ctx).CopyAttributes(oldAlbumID, newAlbumID, "created_at"); err != nil { + if !errors.Is(err, model.ErrNotFound) { + log.Warn(ctx, "Scanner: Could not copy album created_at", "from", oldAlbumID, "to", newAlbumID, err) + } + } + // Note: RefreshPlayCounts will be called in later phases, so we don't need to call it here + } return nil - }) + }, "scanner: move matched track") + if err != nil { + if reassignAlbum { + p.annotationMutex.Lock() + delete(p.processedAlbumAnnotations, newAlbumID) + p.annotationMutex.Unlock() + } + return err + } + p.state.changesDetected.Store(true) + return nil } func (p *phaseMissingTracks) finalize(err error) error { @@ -355,7 +357,12 @@ func (p *phaseMissingTracks) finalize(err error) error { } func (p *phaseMissingTracks) purgeMissing() error { - deletedCount, err := p.ds.MediaFile(p.ctx).DeleteAllMissing() + var deletedCount int64 + err := p.ds.WithTxRetry(p.ctx, func(ctx context.Context, tx model.DataStore) error { + var err error + deletedCount, err = tx.MediaFile(ctx).DeleteAllMissing() + return err + }, "scanner: purge missing") if err != nil { return fmt.Errorf("error deleting missing files: %w", err) } diff --git a/scanner/phase_2_missing_tracks_test.go b/scanner/phase_2_missing_tracks_test.go index f93b166c1..b7aa52f90 100644 --- a/scanner/phase_2_missing_tracks_test.go +++ b/scanner/phase_2_missing_tracks_test.go @@ -2,6 +2,8 @@ package scanner import ( "context" + "errors" + "maps" "time" "github.com/navidrome/navidrome/conf" @@ -146,6 +148,87 @@ var _ = Describe("phaseMissingTracks", func() { Expect(movedTrack.Path).To(Equal(matchedTrack.Path)) }) + Context("claiming the album annotation reassignment", func() { + var probe *probeTxDS + missingTrack := model.MediaFile{ID: "1", PID: "A", AlbumID: "old-album", Path: "dir1/path1.mp3", Tags: model.Tags{"title": []string{"title1"}}, Size: 100} + matchedTrack := model.MediaFile{ID: "2", PID: "A", AlbumID: "new-album", Path: "dir2/path2.mp3", Tags: model.Tags{"title": []string{"title1"}}, Size: 100} + BeforeEach(func() { + probe = &probeTxDS{MockDataStore: ds.(*tests.MockDataStore)} + probe.MockedAlbum = tests.CreateMockAlbumRepo() + phase = createPhaseMissingTracks(ctx, state, probe) + _ = ds.MediaFile(ctx).Put(&missingTrack) + _ = ds.MediaFile(ctx).Put(&matchedTrack) + }) + + It("claims the target album before the transaction, so a concurrent move skips it", func() { + probe.during = func() { + phase.annotationMutex.RLock() + defer phase.annotationMutex.RUnlock() + Expect(phase.processedAlbumAnnotations).To(HaveKeyWithValue("new-album", true)) + } + Expect(phase.moveMatched(matchedTrack, missingTrack)).To(Succeed()) + }) + + It("releases the claim when the move fails, so a later move can reassign", func() { + probe.err = errors.New("boom") + Expect(phase.moveMatched(matchedTrack, missingTrack)).To(MatchError("boom")) + Expect(phase.processedAlbumAnnotations).ToNot(HaveKey("new-album")) + }) + }) + + Context("when the move transaction is rerun after a busy rollback", func() { + var rerunDS *rerunTxDS + BeforeEach(func() { + rerunDS = &rerunTxDS{MockDataStore: ds.(*tests.MockDataStore)} + rerunDS.snapshot = func() func() { + saved := maps.Clone(mr.Data) + return func() { mr.Data = saved } + } + phase = createPhaseMissingTracks(ctx, state, rerunDS) + }) + + It("keeps the moved track", func() { + missingTrack := model.MediaFile{ID: "1", PID: "A", Path: "dir1/path1.mp3", Tags: model.Tags{"title": []string{"title1"}}, Size: 100} + matchedTrack := model.MediaFile{ID: "2", PID: "A", Path: "dir2/path2.mp3", Tags: model.Tags{"title": []string{"title1"}}, Size: 100} + _ = ds.MediaFile(ctx).Put(&missingTrack) + _ = ds.MediaFile(ctx).Put(&matchedTrack) + + _, err := phase.processMissingTracks(&missingTracks{ + missing: []model.MediaFile{missingTrack}, + matched: []model.MediaFile{matchedTrack}, + }) + Expect(err).ToNot(HaveOccurred()) + + movedTrack, err := ds.MediaFile(ctx).Get("1") + Expect(err).ToNot(HaveOccurred()) + Expect(movedTrack.Path).To(Equal(matchedTrack.Path)) + }) + + It("reassigns the album annotations in the attempt that commits", func() { + albumRepo := tests.CreateMockAlbumRepo() + rerunDS.MockedAlbum = albumRepo + restoreTracks := rerunDS.snapshot + rerunDS.snapshot = func() func() { + restore := restoreTracks() + return func() { + restore() + albumRepo.ReassignAnnotationCalls = nil + } + } + missingTrack := model.MediaFile{ID: "1", PID: "A", AlbumID: "old-album", Path: "dir1/path1.mp3", Tags: model.Tags{"title": []string{"title1"}}, Size: 100} + matchedTrack := model.MediaFile{ID: "2", PID: "A", AlbumID: "new-album", Path: "dir2/path2.mp3", Tags: model.Tags{"title": []string{"title1"}}, Size: 100} + _ = ds.MediaFile(ctx).Put(&missingTrack) + _ = ds.MediaFile(ctx).Put(&matchedTrack) + + _, err := phase.processMissingTracks(&missingTracks{ + missing: []model.MediaFile{missingTrack}, + matched: []model.MediaFile{matchedTrack}, + }) + Expect(err).ToNot(HaveOccurred()) + Expect(albumRepo.ReassignAnnotationCalls).To(HaveKeyWithValue("old-album", "new-album")) + }) + }) + It("should move the matched track when the missing track has the same tags and filename", func() { missingTrack := model.MediaFile{ID: "1", PID: "A", Path: "path1.mp3", Tags: model.Tags{"title": []string{"title1"}}, Size: 100} matchedTrack := model.MediaFile{ID: "2", PID: "A", Path: "path1.flac", Tags: model.Tags{"title": []string{"title1"}}, Size: 200} @@ -957,3 +1040,34 @@ var _ = Describe("phaseMissingTracks", func() { }) }) }) + +// rerunTxDS runs every WithTxRetry block twice, as a retry after a rolled-back busy attempt would. +// The mock is not transactional, so snapshot returns the function that plays the rollback. +type rerunTxDS struct { + *tests.MockDataStore + snapshot func() (rollback func()) +} + +func (d *rerunTxDS) WithTxRetry(ctx context.Context, block func(context.Context, model.DataStore) error, _ ...string) error { + rollback := d.snapshot() + _ = block(ctx, d.MockDataStore) + rollback() + return block(ctx, d.MockDataStore) +} + +// probeTxDS runs a hook inside each WithTxRetry block, and can fail the transaction after it. +type probeTxDS struct { + *tests.MockDataStore + during func() + err error +} + +func (d *probeTxDS) WithTxRetry(ctx context.Context, block func(context.Context, model.DataStore) error, _ ...string) error { + if err := block(ctx, d.MockDataStore); err != nil { + return err + } + if d.during != nil { + d.during() + } + return d.err +} diff --git a/scanner/phase_3_refresh_albums.go b/scanner/phase_3_refresh_albums.go index 33e0fed01..964ab7408 100644 --- a/scanner/phase_3_refresh_albums.go +++ b/scanner/phase_3_refresh_albums.go @@ -103,7 +103,9 @@ func (p *phaseRefreshAlbums) refreshAlbum(album *model.Album) (*model.Album, err return nil, nil } start := time.Now() - err := p.ds.Album(p.ctx).Put(album) + err := p.ds.WithTxRetry(p.ctx, func(ctx context.Context, tx model.DataStore) error { + return tx.Album(ctx).Put(album) + }, "scanner: refresh album") log.Debug(p.ctx, "Scanner: refreshing album", "album_id", album.ID, "name", album.Name, "songCount", album.SongCount, "elapsed", time.Since(start), err) if err != nil { return nil, fmt.Errorf("refreshing album %s: %w", album.ID, err) @@ -130,7 +132,12 @@ func (p *phaseRefreshAlbums) finalize(err error) error { } // Refresh album annotations start := time.Now() - cnt, err := p.ds.Album(p.ctx).RefreshPlayCounts() + var cnt int64 + err = p.ds.WithTxRetry(p.ctx, func(ctx context.Context, tx model.DataStore) error { + var txErr error + cnt, txErr = tx.Album(ctx).RefreshPlayCounts() + return txErr + }, "scanner: refresh album play counts") if err != nil { return fmt.Errorf("refreshing album annotations: %w", err) } @@ -138,7 +145,11 @@ func (p *phaseRefreshAlbums) finalize(err error) error { // Refresh artist annotations start = time.Now() - cnt, err = p.ds.Artist(p.ctx).RefreshPlayCounts() + err = p.ds.WithTxRetry(p.ctx, func(ctx context.Context, tx model.DataStore) error { + var txErr error + cnt, txErr = tx.Artist(ctx).RefreshPlayCounts() + return txErr + }, "scanner: refresh artist play counts") if err != nil { return fmt.Errorf("refreshing artist annotations: %w", err) } diff --git a/scanner/phase_4_playlists.go b/scanner/phase_4_playlists.go index baa8b749a..4e11fa81d 100644 --- a/scanner/phase_4_playlists.go +++ b/scanner/phase_4_playlists.go @@ -101,7 +101,10 @@ func (p *phasePlaylists) produce(put func(entry *model.Folder)) error { // import the playlists, and returns an error if the flag can't be persisted (so // the scan does not complete as successful without recording the recovery). func (p *phasePlaylists) deferImport() error { - if err := p.ds.Property(p.ctx).Put(consts.PlaylistsImportPendingFlagKey, "1"); err != nil { + err := p.ds.WithTxRetry(p.ctx, func(ctx context.Context, tx model.DataStore) error { + return tx.Property(ctx).Put(consts.PlaylistsImportPendingFlagKey, "1") + }, "scanner: defer playlist import") + if err != nil { return fmt.Errorf("recording pending playlist import: %w", err) } log.Warn(p.ctx, "Playlists will not be imported, as there are no admin users yet. "+ diff --git a/scanner/scanner.go b/scanner/scanner.go index d73007bdd..a8a192771 100644 --- a/scanner/scanner.go +++ b/scanner/scanner.go @@ -217,7 +217,9 @@ func (s *scannerImpl) prepareLibrariesForScan(ctx context.Context, state *scanSt for _, lib := range state.libraries { if lib.LastScanStartedAt.IsZero() { // This is a new scan - mark it as started - err := s.ds.Library(ctx).ScanBegin(lib.ID, state.fullScan) + err := s.ds.WithTxRetry(ctx, func(ctx context.Context, tx model.DataStore) error { + return tx.Library(ctx).ScanBegin(lib.ID, state.fullScan) + }, "scanner: begin library scan") if err != nil { log.Error(ctx, "Scanner: Error marking scan start", "lib", lib.Name, err) state.sendWarning(err.Error()) @@ -253,7 +255,7 @@ func (s *scannerImpl) prepareLibrariesForScan(ctx context.Context, state *scanSt func (s *scannerImpl) runGC(ctx context.Context, state *scanState) func() error { return func() error { state.sendProgress(&ProgressInfo{ForceUpdate: true}) - return s.ds.WithTx(func(tx model.DataStore) error { + return s.ds.WithTxRetry(ctx, func(ctx context.Context, tx model.DataStore) error { if state.changesDetected.Load() { start := time.Now() @@ -264,9 +266,7 @@ func (s *scannerImpl) runGC(ctx context.Context, state *scanState) func() error log.Debug(ctx, "Scanner: Running selective GC", "libraryIDs", libraryIDs) } - err := tx.GC(ctx, libraryIDs...) - if err != nil { - log.Error(ctx, "Scanner: Error running GC", err) + if err := tx.GC(ctx, libraryIDs...); err != nil { return fmt.Errorf("running GC: %w", err) } log.Debug(ctx, "Scanner: GC completed", "elapsed", time.Since(start)) @@ -286,10 +286,14 @@ func (s *scannerImpl) runEnqueueMissingArtwork(ctx context.Context, state *scanS return nil } start := time.Now() - queue := s.ds.ArtworkQueue(ctx) var total int64 for _, kind := range []model.Kind{model.KindAlbumArtwork, model.KindArtistArtwork} { - n, err := queue.EnqueueAllMissing(kind, model.ArtworkPriorityScan) + var n int64 + err := s.ds.WithTxRetry(ctx, func(ctx context.Context, tx model.DataStore) error { + var err error + n, err = tx.ArtworkQueue(ctx).EnqueueAllMissing(kind, model.ArtworkPriorityScan) + return err + }, "scanner: enqueue missing artwork") if err != nil { log.Error(ctx, "Scanner: Error enqueueing missing artwork", "kind", kind, err) return fmt.Errorf("enqueueing missing artwork: %w", err) @@ -316,7 +320,9 @@ func (s *scannerImpl) runRefreshStats(ctx context.Context, state *scanState) fun log.Debug(ctx, "Scanner: Refreshed artist stats", "stats", stats, "elapsed", time.Since(start)) start = time.Now() - err = s.ds.Tag(ctx).UpdateCounts() + err = s.ds.WithTxRetry(ctx, func(ctx context.Context, tx model.DataStore) error { + return tx.Tag(ctx).UpdateCounts() + }, "scanner: update tag counts") if err != nil { log.Error(ctx, "Scanner: Error updating tag counts", err) return fmt.Errorf("updating tag counts: %w", err) @@ -329,28 +335,21 @@ func (s *scannerImpl) runRefreshStats(ctx context.Context, state *scanState) fun func (s *scannerImpl) runUpdateLibraries(ctx context.Context, state *scanState) func() error { return func() error { start := time.Now() - return s.ds.WithTx(func(tx model.DataStore) error { + return s.ds.WithTxRetry(ctx, func(ctx context.Context, tx model.DataStore) error { for _, lib := range state.libraries { - err := tx.Library(ctx).ScanEnd(lib.ID) - if err != nil { - log.Error(ctx, "Scanner: Error updating last scan completed", "lib", lib.Name, err) - return fmt.Errorf("updating last scan completed: %w", err) + if err := tx.Library(ctx).ScanEnd(lib.ID); err != nil { + return fmt.Errorf("updating last scan completed for %s: %w", lib.Name, err) } - err = tx.Property(ctx).Put(consts.PIDTrackKey, conf.Server.PID.Track) - if err != nil { - log.Error(ctx, "Scanner: Error updating track PID conf", err) + if err := tx.Property(ctx).Put(consts.PIDTrackKey, conf.Server.PID.Track); err != nil { return fmt.Errorf("updating track PID conf: %w", err) } - err = tx.Property(ctx).Put(consts.PIDAlbumKey, conf.Server.PID.Album) - if err != nil { - log.Error(ctx, "Scanner: Error updating album PID conf", err) + if err := tx.Property(ctx).Put(consts.PIDAlbumKey, conf.Server.PID.Album); err != nil { return fmt.Errorf("updating album PID conf: %w", err) } if state.changesDetected.Load() { log.Debug(ctx, "Scanner: Refreshing library stats", "lib", lib.Name) if err := tx.Library(ctx).RefreshStats(lib.ID); err != nil { - log.Error(ctx, "Scanner: Error refreshing library stats", "lib", lib.Name, err) - return fmt.Errorf("refreshing library stats: %w", err) + return fmt.Errorf("refreshing library stats for %s: %w", lib.Name, err) } } else { log.Debug(ctx, "Scanner: No changes detected, skipping library stats refresh", "lib", lib.Name) diff --git a/scanner/scanner_test.go b/scanner/scanner_test.go index 8542b3ac6..30f4a2b97 100644 --- a/scanner/scanner_test.go +++ b/scanner/scanner_test.go @@ -4,12 +4,16 @@ import ( "context" "database/sql" "errors" + "fmt" + "os" "path/filepath" + "sync/atomic" "testing/fstest" "time" "github.com/Masterminds/squirrel" "github.com/google/uuid" + "github.com/mattn/go-sqlite3" "github.com/navidrome/navidrome/conf" "github.com/navidrome/navidrome/conf/configtest" "github.com/navidrome/navidrome/consts" @@ -52,7 +56,10 @@ var _ = Describe("Scanner", Ordered, func() { BeforeAll(func() { ctx = request.WithUser(GinkgoT().Context(), model.User{ID: "123", IsAdmin: true}) - tmpDir := GinkgoT().TempDir() + // The DB stays open until the suite ends, and Windows can't delete an open file + tmpDir, err := os.MkdirTemp("", "scanner-test") + Expect(err).ToNot(HaveOccurred()) + DeferCleanup(func() { _ = os.RemoveAll(tmpDir) }) conf.Server.DbPath = filepath.Join(tmpDir, "test-scanner.db?_journal_mode=WAL") log.Warn("Using DB at " + conf.Server.DbPath) //conf.Server.DbPath = ":memory:" @@ -1240,8 +1247,55 @@ var _ = Describe("Scanner", Ordered, func() { Expect(albumArtistStats.SongCount).To(Equal(3)) // 3 songs }) }) + + Context("when the database is busy", func() { + var busyDS *busyPersistDS + BeforeEach(func() { + // One album across many folders: the suite's single DB connection deadlocks phase 3 on many albums + album := template(_t{"albumartist": "Artist", "album": "Album"}) + files := fstest.MapFS{} + for i := range 30 { + files[fmt.Sprintf("Artist/Part %02d/%02d - Song.mp3", i, i+1)] = album(track(i+1, fmt.Sprintf("Song %02d", i+1))) + } + createFS(files) + busyDS = &busyPersistDS{MockDataStore: ds} + s = scanner.New(ctx, busyDS, events.NoopBroker(), + playlists.NewPlaylists(busyDS, artwork.NewUploader(busyDS)), metrics.NewNoopInstance()) + }) + + It("gives up and stops walking the library when the database stays busy", func() { + busyDS.failures.Store(1000) + + Expect(runScanner(ctx, true)).To(MatchError(ContainSubstring("database is locked"))) + + Expect(mfRepo.cursorCalls.Load()).To(BeNumerically("<", 30)) + }) + + It("does not mark unvisited folders missing when the scan gives up", func() { + Expect(runScanner(ctx, true)).To(Succeed()) + busyDS.failures.Store(1000) + + Expect(runScanner(ctx, true)).ToNot(Succeed()) + + Expect(ds.Folder(ctx).CountAll(model.QueryOptions{Filters: squirrel.Eq{"missing": true}})).To(BeZero()) + Expect(ds.MediaFile(ctx).CountAll(model.QueryOptions{Filters: squirrel.Eq{"missing": true}})).To(BeZero()) + }) + }) }) +// busyPersistDS fails the scanner's folder saves with SQLITE_BUSY, as if WithTxRetry ran out of retries. +type busyPersistDS struct { + *tests.MockDataStore + failures atomic.Int32 +} + +func (b *busyPersistDS) WithTxRetry(ctx context.Context, block func(context.Context, model.DataStore) error, label ...string) error { + if len(label) > 0 && label[0] == "scanner: persist changes" && b.failures.Add(-1) >= 0 { + return sqlite3.Error{Code: sqlite3.ErrBusy} + } + return b.MockDataStore.WithTxRetry(ctx, block, label...) +} + func createFindByPath(ctx context.Context, ds model.DataStore) func(string) (*model.MediaFile, error) { return func(path string) (*model.MediaFile, error) { list, err := ds.MediaFile(ctx).FindByPaths([]string{path}) @@ -1258,6 +1312,12 @@ func createFindByPath(ctx context.Context, ds model.DataStore) func(string) (*mo type mockMediaFileRepo struct { model.MediaFileRepository GetMissingAndMatchingError error + cursorCalls atomic.Int32 +} + +func (m *mockMediaFileRepo) GetCursor(options ...model.QueryOptions) (model.MediaFileCursor, error) { + m.cursorCalls.Add(1) + return m.MediaFileRepository.GetCursor(options...) } func (m *mockMediaFileRepo) GetMissingAndMatching(libId int) (model.MediaFileCursor, error) { diff --git a/scanner/walk_dir_tree.go b/scanner/walk_dir_tree.go index 20f6d4213..1864a6a44 100644 --- a/scanner/walk_dir_tree.go +++ b/scanner/walk_dir_tree.go @@ -53,6 +53,9 @@ func walkDirTree(ctx context.Context, job *scanJob, targetFolders ...string) (<- // Recursively walk this folder and all its children err = walkFolder(ctx, job, folderPath, checker, results) + if utils.IsCtxDone(ctx) { + return + } if err != nil { log.Error(ctx, "Scanner: Error walking target folder", "path", folderPath, err) continue @@ -87,9 +90,12 @@ func walkFolder(ctx context.Context, job *scanJob, currentFolder string, checker folder.path = dir folder.elapsed.Start() - results <- folder - - return nil + select { + case results <- folder: + return nil + case <-ctx.Done(): + return ctx.Err() + } } func loadDir(ctx context.Context, job *scanJob, dirPath string, checker *IgnoreChecker) (folder *folderEntry, children []string, err error) { @@ -291,6 +297,7 @@ var ignoredDirs = []string{ "$RECYCLE.BIN", "#snapshot", "@Recycle", + "@eaDir", "@Recently-Snapshot", ".git", ".streams", diff --git a/scanner/walk_dir_tree_test.go b/scanner/walk_dir_tree_test.go index 9fb650c4d..43939e5c2 100644 --- a/scanner/walk_dir_tree_test.go +++ b/scanner/walk_dir_tree_test.go @@ -564,6 +564,7 @@ var _ = Describe("walk_dir_tree", func() { Entry("dir starting with ellipsis", "...unhidden_folder", false), Entry("recycle bin", "$Recycle.Bin", true), Entry("snapshot dir", "#snapshot", true), + Entry("synology metadata dir", "@eaDir", true), ) }) diff --git a/tests/mock_data_store.go b/tests/mock_data_store.go index 32f56a4f0..6a0ebbb31 100644 --- a/tests/mock_data_store.go +++ b/tests/mock_data_store.go @@ -329,6 +329,10 @@ func (db *MockDataStore) WithTxImmediate(block func(tx model.DataStore) error, l return block(db) } +func (db *MockDataStore) WithTxRetry(ctx context.Context, block func(ctx context.Context, tx model.DataStore) error, label ...string) error { + return block(ctx, db) +} + func (db *MockDataStore) Resource(ctx context.Context, m any) model.ResourceRepository { switch m.(type) { case model.MediaFile, *model.MediaFile: