package library import ( "context" "fmt" "log/slog" "sync" "time" "github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgxpool" "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" ) // Fingerprint backfill (M400 #3908). // // The scan fingerprints only bytes it has not seen (see scanFile), so every track // imported before fingerprinting existed — and every row derived by an older // fingerprintVersion — needs a pass of its own. That pass is this worker. // // Its own worker rather than a stage in RunScan, for two reasons, both about // time: // - RunScan runs at boot and then every safetyNetScanInterval (12h), and an // in-flight scan older than StuckScanThreshold (1h) is reaped and a second // one started beside it. A stage would have to stop well inside the hour — a // few hundred decodes — so a 50k-track library would take about a month. // - A long stage holds the scan run in flight, and a manual rescan answers 409 // for as long as it runs. // // Progress is read live (FingerprintCoverage, the admin gauge) rather than from a // scan_runs tally: the work spans many passes with no single run to attach to. // fingerprintBackfillTick is how often the worker looks for work. Once the // library has caught up, a tick is one indexed query; mostly the hour bounds how // long a file that timed out on a slow mount waits before it is tried again. const fingerprintBackfillTick = time.Hour // fingerprintBackfillBatch is how many tracks one query hands the worker. Small, // so tracks the scan adds mid-pass are not stuck behind one enormous page. const fingerprintBackfillBatch = 50 // fingerprintBackfillConcurrency is the shipped value of the concurrency setting // (#3913): how many files are decoded at once. Two is deliberately low: fpcalc // and the stream hash compete with playback transcoding for CPU and with // streaming for the mount, and a backfill that makes playback stutter is worse // than one that takes longer. const fingerprintBackfillConcurrency = 2 // BackfillFingerprintsResult tallies one pass. type BackfillFingerprintsResult struct { Processed int Fingerprinted int // both halves stored Rejected int // stored with a NULL half: a tool refused the file (settled) Inconclusive int // nothing stored; tried again on a later pass } func (r *BackfillFingerprintsResult) add(o fingerprintOutcome) { r.Processed++ switch o { case outcomeFingerprinted: r.Fingerprinted++ case outcomeRejected: r.Rejected++ default: r.Inconclusive++ } } // FingerprintBackfillWorker fingerprints the tracks the scan never will. type FingerprintBackfillWorker struct { pool *pgxpool.Pool logger *slog.Logger settings *FingerprintSettingsService tick time.Duration batch int32 // fingerprint is a field for the same reason as Scanner.fingerprint: an // integration test pins which tracks a pass touches, not what the tools print. fingerprint func(ctx context.Context, path string, opts fingerprintOptions) fingerprintResult } // NewFingerprintBackfillWorker builds a worker with the production cadence. // settings is shared with the scanner and the admin API; nil runs on defaults. func NewFingerprintBackfillWorker( pool *pgxpool.Pool, logger *slog.Logger, settings *FingerprintSettingsService, ) *FingerprintBackfillWorker { return &FingerprintBackfillWorker{ pool: pool, logger: logger, settings: settings, tick: fingerprintBackfillTick, batch: fingerprintBackfillBatch, fingerprint: computeFingerprint, } } // Run blocks until ctx is cancelled: one pass at start, so a fresh deploy does // not sit idle for an hour, then one per tick. func (w *FingerprintBackfillWorker) Run(ctx context.Context) { w.runOnce(ctx) t := time.NewTicker(w.tick) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: w.runOnce(ctx) } } } // runOnce contains a pass so that nothing it does — an error, a panic — can stop // the next tick from firing (rule 157). func (w *FingerprintBackfillWorker) runOnce(ctx context.Context) { defer func() { if r := recover(); r != nil { w.logger.Error("fingerprint backfill: pass panicked", "panic", r) } }() res, err := w.pass(ctx) if err != nil && ctx.Err() == nil { w.logger.Warn("fingerprint backfill: pass failed", "err", err, "processed", res.Processed) } if res.Processed > 0 { w.logger.Info("fingerprint backfill: pass complete", "processed", res.Processed, "fingerprinted", res.Fingerprinted, "rejected", res.Rejected, "inconclusive", res.Inconclusive) } } // pass walks every track needing a fingerprint once, keyset-paged on id. The // cursor is what lets a pass end: an inconclusive attempt writes no row, so a // file that keeps timing out would otherwise be listed again immediately and // retried forever within the pass. // // Settings are read before every batch, so a save takes effect within a batch // rather than an hour (#3913): switching fingerprinting off ends the pass, a new // concurrency applies to the next batch, and a new length restarts the walk from // the top at that length, because every row written at the old one went stale // the moment it changed. func (w *FingerprintBackfillWorker) pass(ctx context.Context) (BackfillFingerprintsResult, error) { q := dbq.New(w.pool) var ( res BackfillFingerprintsResult mu sync.Mutex ) // The all-zero uuid sorts before every real id. Valid must be true: a NULL // cursor would make "id > NULL" match nothing and every pass a silent no-op. start := pgtype.UUID{Valid: true} after := start lengthSec := w.settings.Get().ChromaprintLengthSec for { if err := ctx.Err(); err != nil { return res, err } cfg := w.settings.Get() if !cfg.Enabled { return res, nil } if cfg.ChromaprintLengthSec != lengthSec { lengthSec, after = cfg.ChromaprintLengthSec, start } rows, err := q.ListTracksNeedingFingerprint(ctx, dbq.ListTracksNeedingFingerprintParams{ CurrentVersion: fingerprintVersion, ChromaprintLengthSec: lengthSec, AfterID: after, BatchLimit: w.batch, }) if err != nil { return res, fmt.Errorf("list tracks needing fingerprint: %w", err) } if len(rows) == 0 { return res, nil } opts := fingerprintOptions{lengthSec: lengthSec, chromaprint: true} // Validation keeps concurrency at one or more; the floor guards a zero // that would block the first send for ever. sem := make(chan struct{}, max(1, int(cfg.BackfillConcurrency))) var wg sync.WaitGroup for _, row := range rows { if ctx.Err() != nil { break } sem <- struct{}{} wg.Add(1) go func(trackID pgtype.UUID, path string) { defer wg.Done() defer func() { <-sem }() defer func() { if r := recover(); r != nil { w.logger.Error("fingerprint backfill: track panicked", "path", path, "panic", r) } }() outcome := storeFingerprint(ctx, q, w.logger, trackID, path, w.fingerprintFile(ctx, path, opts), lengthSec) mu.Lock() res.add(outcome) mu.Unlock() }(row.ID, row.FilePath) } wg.Wait() after = rows[len(rows)-1].ID } } func (w *FingerprintBackfillWorker) fingerprintFile(ctx context.Context, path string, opts fingerprintOptions) fingerprintResult { if w.fingerprint == nil { return computeFingerprint(ctx, path, opts) } return w.fingerprint(ctx, path, opts) } // FingerprintCoverage reports how much of the library carries a current // fingerprint, for the admin gauge. It lives here, beside the backfill, so the // version and length it counts against are the ones the backfill writes. func FingerprintCoverage( ctx context.Context, pool *pgxpool.Pool, cfg FingerprintSettings, ) (dbq.GetFingerprintCoverageRow, error) { return dbq.New(pool).GetFingerprintCoverage(ctx, dbq.GetFingerprintCoverageParams{ CurrentVersion: fingerprintVersion, ChromaprintLengthSec: cfg.ChromaprintLengthSec, }) }