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 (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 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. Operator-tunable in #3913.
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
	tick        time.Duration
	batch       int32
	concurrency int
	// 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) fingerprintResult
}

// NewFingerprintBackfillWorker builds a worker with the production cadence.
func NewFingerprintBackfillWorker(pool *pgxpool.Pool, logger *slog.Logger) *FingerprintBackfillWorker {
	return &FingerprintBackfillWorker{
		pool:        pool,
		logger:      logger,
		tick:        fingerprintBackfillTick,
		batch:       fingerprintBackfillBatch,
		concurrency: fingerprintBackfillConcurrency,
		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.
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.
	after := pgtype.UUID{Valid: true}
	for {
		if err := ctx.Err(); err != nil {
			return res, err
		}
		rows, err := q.ListTracksNeedingFingerprint(ctx, dbq.ListTracksNeedingFingerprintParams{
			CurrentVersion: fingerprintVersion,
			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
		}

		sem := make(chan struct{}, w.concurrency)
		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))
				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) fingerprintResult {
	if w.fingerprint == nil {
		return computeFingerprint(ctx, path)
	}
	return w.fingerprint(ctx, path)
}

// FingerprintCoverage reports how much of the library carries a current
// fingerprint, for the admin gauge. It lives here, beside the backfill, so the
// version it counts against is the one the backfill writes.
func FingerprintCoverage(ctx context.Context, pool *pgxpool.Pool) (dbq.GetFingerprintCoverageRow, error) {
	return dbq.New(pool).GetFingerprintCoverage(ctx, fingerprintVersion)
}