diff --git a/internal/library/duplicate_resolve.go b/internal/library/duplicate_resolve.go index c7bf6625..a4951d59 100644 --- a/internal/library/duplicate_resolve.go +++ b/internal/library/duplicate_resolve.go @@ -46,6 +46,9 @@ const ( resolveMergeCap = 200 // duplicateResolveTick is how often the worker runs a pass. duplicateResolveTick = time.Hour + // scanWaitRetry is how soon a pass put off for a running scan is tried + // again: a boot scan takes under a minute on the operator's library. + scanWaitRetry = time.Minute ) // DuplicateResolveResult tallies one pass. @@ -420,38 +423,58 @@ func NewDuplicateResolveWorker( } } -// Run blocks until ctx is cancelled, running a pass at start and then each tick. +// Run blocks until ctx is cancelled, running a pass at start and then each +// tick. A pass put off for a running scan is tried again after scanWaitRetry. func (w *DuplicateResolveWorker) Run(ctx context.Context) { - w.tickOnce(ctx) t := time.NewTicker(w.tick) defer t.Stop() + soon := time.NewTimer(0) + defer soon.Stop() for { select { case <-ctx.Done(): return case <-t.C: - w.tickOnce(ctx) + case <-soon.C: + } + if w.tickOnce(ctx) { + soon.Reset(scanWaitRetry) } } } -// tickOnce contains one pass so nothing it does can stop the next tick (rule 157). -func (w *DuplicateResolveWorker) tickOnce(ctx context.Context) { +// tickOnce contains one pass so nothing it does can stop the next tick (rule +// 157). It reports whether the pass was put off for a running scan. +func (w *DuplicateResolveWorker) tickOnce(ctx context.Context) (waitForScan bool) { defer func() { if r := recover(); r != nil { w.logger.Error("duplicate resolve: tick panicked", "panic", r) } }() + q := dbq.New(w.pool) // A sweep in flight is rewriting the groups; the next tick sees its result. - _, err := dbq.New(w.pool).GetInFlightDuplicateSweep(ctx) + _, err := q.GetInFlightDuplicateSweep(ctx) switch { case err == nil: - return + return false case !errors.Is(err, pgx.ErrNoRows): if ctx.Err() == nil { w.logger.Warn("duplicate resolve: sweep check failed", "err", err) } - return + return false + } + // A scan in flight has listed the files it will reconcile: a copy removed + // now is marked present again when it does (7 of the 13 removed at boot + // on 2026-10-20, #6169). A run past StuckScanThreshold is a dead one. + run, err := q.GetInFlightScanRun(ctx) + switch { + case err == nil && time.Since(run.StartedAt.Time) <= StuckScanThreshold: + return true + case err != nil && !errors.Is(err, pgx.ErrNoRows): + if ctx.Err() == nil { + w.logger.Warn("duplicate resolve: scan check failed", "err", err) + } + return false } res, err := ResolveDuplicatesWith(ctx, w.pool, w.logger, w.dataDir, w.settings.Get().AutoResolve, w.src) if err != nil && ctx.Err() == nil { @@ -461,4 +484,5 @@ func (w *DuplicateResolveWorker) tickOnce(ctx context.Context) { "groups", res.Groups, "merged", res.Merged, "merge_failed", res.MergeFailed, "songs_linked", res.SongsLinked, "dismissed", res.Dismissed, "wrong_files", res.WrongFiles, "held_back", res.HeldBack, "released", res.Released) + return false } diff --git a/internal/library/duplicate_resolve_test.go b/internal/library/duplicate_resolve_test.go index 99328870..6eb44c72 100644 --- a/internal/library/duplicate_resolve_test.go +++ b/internal/library/duplicate_resolve_test.go @@ -2,6 +2,8 @@ package library import ( "context" + "io" + "log/slog" "os" "path/filepath" "testing" @@ -549,6 +551,30 @@ func TestResolveDuplicates_AgreeingLookupsDecideAnAcousticGroup_Integration(t *t }) } +// A pass while a scan is walking the library waits for it: a copy removed +// under the walk is marked present again when the scan reconciles. A run +// stuck past StuckScanThreshold does not hold the resolver up. +func TestDuplicateResolveWorker_WaitsForARunningScan_Integration(t *testing.T) { + f := seedTrimmedAlbum(t) + ctx := context.Background() + w := NewDuplicateResolveWorker(f.pool, slog.New(slog.NewTextHandler(io.Discard, nil)), "", nil, ResolveSources{}) + if _, err := f.pool.Exec(ctx, `INSERT INTO scan_runs DEFAULT VALUES`); err != nil { + t.Fatal(err) + } + if !w.tickOnce(ctx) { + t.Error("a pass ran under a running scan") + } + if !exists(f.aPath) || !exists(f.bPath) { + t.Error("a copy was removed while the scan was walking the library") + } + if _, err := f.pool.Exec(ctx, `UPDATE scan_runs SET started_at = now() - interval '2 hours'`); err != nil { + t.Fatal(err) + } + if w.tickOnce(ctx) { + t.Error("a scan stuck for two hours held the resolver up") + } +} + // AcoustID knows the recording as no song of the album: every copy goes. With // AcoustID unreachable, nothing is decided and the group waits. func TestResolveDuplicates_TrimmedCopiesAcoustIDCannotName_Integration(t *testing.T) {