minstrel/internal/library/fingerprint_backfill.go

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.
//
// 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: 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.
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: 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,
	})
}