Backfill fingerprints for the existing library (M400 #3908) #133
@@ -214,6 +214,12 @@ func run() error {
|
|||||||
// SQL, no external calls; empty on single-user servers.
|
// SQL, no external calls; empty on single-user servers.
|
||||||
go coplay.NewWorker(pool, logger.With("component", "coplay")).Run(ctx)
|
go coplay.NewWorker(pool, logger.With("component", "coplay")).Run(ctx)
|
||||||
|
|
||||||
|
// Fingerprint backfill (M400 #3908): fingerprints the tracks the scan never
|
||||||
|
// will — everything imported before fingerprinting existed, and rows derived
|
||||||
|
// by an older method. A worker of its own rather than a scan stage; see
|
||||||
|
// internal/library/fingerprint_backfill.go for why.
|
||||||
|
go library.NewFingerprintBackfillWorker(pool, logger.With("component", "fingerprint_backfill")).Run(ctx)
|
||||||
|
|
||||||
// Start the tag-enrichment worker (#1490). Reconciles the compiled-in
|
// Start the tag-enrichment worker (#1490). Reconciles the compiled-in
|
||||||
// tag providers with tag_provider_settings, bumps the sources version if
|
// tag providers with tag_provider_settings, bumps the sources version if
|
||||||
// the provider set changed (re-opening settled rows), then drains tracks
|
// the provider set changed (re-opening settled rows), then drains tracks
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import (
|
|||||||
|
|
||||||
"git.fabledsword.com/bvandeusen/minstrel/internal/apierror"
|
"git.fabledsword.com/bvandeusen/minstrel/internal/apierror"
|
||||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||||
|
"git.fabledsword.com/bvandeusen/minstrel/internal/library"
|
||||||
)
|
)
|
||||||
|
|
||||||
// coverageRollupResp is the wire shape for GET /api/admin/library/coverage.
|
// coverageRollupResp is the wire shape for GET /api/admin/library/coverage.
|
||||||
@@ -36,3 +37,31 @@ func (h *handlers) handleGetLibraryCoverage(w http.ResponseWriter, r *http.Reque
|
|||||||
PendingNoMbid: row.PendingNoMbid,
|
PendingNoMbid: row.PendingNoMbid,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// fingerprintCoverageResp is the wire shape for GET /api/admin/library/fingerprints.
|
||||||
|
// fingerprinted + rejected + pending = total. Missing tracks are not counted:
|
||||||
|
// there is no file to fingerprint.
|
||||||
|
type fingerprintCoverageResp struct {
|
||||||
|
Total int64 `json:"total"`
|
||||||
|
Fingerprinted int64 `json:"fingerprinted"`
|
||||||
|
Rejected int64 `json:"rejected"`
|
||||||
|
Pending int64 `json:"pending"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// handleGetFingerprintCoverage implements GET /api/admin/library/fingerprints:
|
||||||
|
// how far the fingerprint backfill (#3908) has got. The backfill is its own
|
||||||
|
// worker spanning many passes, with no scan run to attach a tally to, so its
|
||||||
|
// progress is read live here. Always 200; zeros on an empty library.
|
||||||
|
func (h *handlers) handleGetFingerprintCoverage(w http.ResponseWriter, r *http.Request) {
|
||||||
|
row, err := library.FingerprintCoverage(r.Context(), h.pool)
|
||||||
|
if err != nil {
|
||||||
|
writeErrWithLog(w, h.logger, "admin: get fingerprint coverage", apierror.InternalMsg("lookup failed", err))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writeJSON(w, http.StatusOK, fingerprintCoverageResp{
|
||||||
|
Total: row.Total,
|
||||||
|
Fingerprinted: row.Fingerprinted,
|
||||||
|
Rejected: row.Rejected,
|
||||||
|
Pending: row.Pending,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|||||||
@@ -215,6 +215,7 @@ func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, events *playev
|
|||||||
admin.Get("/library/missing", h.handleListMissingTracks)
|
admin.Get("/library/missing", h.handleListMissingTracks)
|
||||||
|
|
||||||
admin.Get("/library/coverage", h.handleGetLibraryCoverage)
|
admin.Get("/library/coverage", h.handleGetLibraryCoverage)
|
||||||
|
admin.Get("/library/fingerprints", h.handleGetFingerprintCoverage)
|
||||||
|
|
||||||
admin.Get("/invites", h.handleListInvites)
|
admin.Get("/invites", h.handleListInvites)
|
||||||
admin.Post("/invites", h.handleCreateInvite)
|
admin.Post("/invites", h.handleCreateInvite)
|
||||||
|
|||||||
@@ -23,6 +23,95 @@ func (q *Queries) DeleteTrackFingerprint(ctx context.Context, trackID pgtype.UUI
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const getFingerprintCoverage = `-- name: GetFingerprintCoverage :one
|
||||||
|
SELECT count(*)::bigint AS total,
|
||||||
|
count(*) FILTER (
|
||||||
|
WHERE f.fingerprint_version >= $1
|
||||||
|
AND f.audio_stream_sha256 IS NOT NULL AND f.chromaprint IS NOT NULL
|
||||||
|
)::bigint AS fingerprinted,
|
||||||
|
count(*) FILTER (
|
||||||
|
WHERE f.fingerprint_version >= $1
|
||||||
|
AND (f.audio_stream_sha256 IS NULL OR f.chromaprint IS NULL)
|
||||||
|
)::bigint AS rejected,
|
||||||
|
count(*) FILTER (
|
||||||
|
WHERE f.track_id IS NULL OR f.fingerprint_version < $1
|
||||||
|
)::bigint AS pending
|
||||||
|
FROM tracks t
|
||||||
|
LEFT JOIN track_fingerprints f ON f.track_id = t.id
|
||||||
|
WHERE t.missing_since IS NULL
|
||||||
|
`
|
||||||
|
|
||||||
|
type GetFingerprintCoverageRow struct {
|
||||||
|
Total int64
|
||||||
|
Fingerprinted int64
|
||||||
|
Rejected int64
|
||||||
|
Pending int64
|
||||||
|
}
|
||||||
|
|
||||||
|
// The admin gauge for the backfill. fingerprinted + rejected + pending = total.
|
||||||
|
// rejected is a row at the current version with a NULL half: a tool ran and
|
||||||
|
// refused the file, which is settled rather than waiting. Missing tracks are
|
||||||
|
// excluded, or the gauge could never reach the end.
|
||||||
|
func (q *Queries) GetFingerprintCoverage(ctx context.Context, currentVersion int16) (GetFingerprintCoverageRow, error) {
|
||||||
|
row := q.db.QueryRow(ctx, getFingerprintCoverage, currentVersion)
|
||||||
|
var i GetFingerprintCoverageRow
|
||||||
|
err := row.Scan(
|
||||||
|
&i.Total,
|
||||||
|
&i.Fingerprinted,
|
||||||
|
&i.Rejected,
|
||||||
|
&i.Pending,
|
||||||
|
)
|
||||||
|
return i, err
|
||||||
|
}
|
||||||
|
|
||||||
|
const listTracksNeedingFingerprint = `-- name: ListTracksNeedingFingerprint :many
|
||||||
|
SELECT t.id, t.file_path
|
||||||
|
FROM tracks t
|
||||||
|
LEFT JOIN track_fingerprints f ON f.track_id = t.id
|
||||||
|
WHERE t.missing_since IS NULL
|
||||||
|
AND (f.track_id IS NULL OR f.fingerprint_version < $1)
|
||||||
|
AND t.id > $2
|
||||||
|
ORDER BY t.id
|
||||||
|
LIMIT $3
|
||||||
|
`
|
||||||
|
|
||||||
|
type ListTracksNeedingFingerprintParams struct {
|
||||||
|
CurrentVersion int16
|
||||||
|
AfterID pgtype.UUID
|
||||||
|
BatchLimit int32
|
||||||
|
}
|
||||||
|
|
||||||
|
type ListTracksNeedingFingerprintRow struct {
|
||||||
|
ID pgtype.UUID
|
||||||
|
FilePath string
|
||||||
|
}
|
||||||
|
|
||||||
|
// The backfill's work queue (#3908): tracks with no fingerprint, or one derived
|
||||||
|
// by an older method. Keyset-paged on id so a pass visits each track at most
|
||||||
|
// once. That cursor is load-bearing: an inconclusive attempt writes no row, so
|
||||||
|
// without it a file that keeps timing out would be listed again straight away
|
||||||
|
// and retried in a tight loop. Missing tracks are skipped — there is no file to
|
||||||
|
// read.
|
||||||
|
func (q *Queries) ListTracksNeedingFingerprint(ctx context.Context, arg ListTracksNeedingFingerprintParams) ([]ListTracksNeedingFingerprintRow, error) {
|
||||||
|
rows, err := q.db.Query(ctx, listTracksNeedingFingerprint, arg.CurrentVersion, arg.AfterID, arg.BatchLimit)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
var items []ListTracksNeedingFingerprintRow
|
||||||
|
for rows.Next() {
|
||||||
|
var i ListTracksNeedingFingerprintRow
|
||||||
|
if err := rows.Scan(&i.ID, &i.FilePath); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
items = append(items, i)
|
||||||
|
}
|
||||||
|
if err := rows.Err(); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return items, nil
|
||||||
|
}
|
||||||
|
|
||||||
const upsertTrackFingerprint = `-- name: UpsertTrackFingerprint :exec
|
const upsertTrackFingerprint = `-- name: UpsertTrackFingerprint :exec
|
||||||
INSERT INTO track_fingerprints (
|
INSERT INTO track_fingerprints (
|
||||||
track_id, audio_stream_sha256, chromaprint, fingerprint_version
|
track_id, audio_stream_sha256, chromaprint, fingerprint_version
|
||||||
|
|||||||
@@ -20,3 +20,40 @@ ON CONFLICT (track_id) DO UPDATE SET
|
|||||||
-- file. The stored row describes the OLD bytes, so it goes and the backfill
|
-- file. The stored row describes the OLD bytes, so it goes and the backfill
|
||||||
-- re-derives it — nothing may keep trusting a stale identity.
|
-- re-derives it — nothing may keep trusting a stale identity.
|
||||||
DELETE FROM track_fingerprints WHERE track_id = $1;
|
DELETE FROM track_fingerprints WHERE track_id = $1;
|
||||||
|
|
||||||
|
-- name: ListTracksNeedingFingerprint :many
|
||||||
|
-- The backfill's work queue (#3908): tracks with no fingerprint, or one derived
|
||||||
|
-- by an older method. Keyset-paged on id so a pass visits each track at most
|
||||||
|
-- once. That cursor is load-bearing: an inconclusive attempt writes no row, so
|
||||||
|
-- without it a file that keeps timing out would be listed again straight away
|
||||||
|
-- and retried in a tight loop. Missing tracks are skipped — there is no file to
|
||||||
|
-- read.
|
||||||
|
SELECT t.id, t.file_path
|
||||||
|
FROM tracks t
|
||||||
|
LEFT JOIN track_fingerprints f ON f.track_id = t.id
|
||||||
|
WHERE t.missing_since IS NULL
|
||||||
|
AND (f.track_id IS NULL OR f.fingerprint_version < sqlc.arg(current_version))
|
||||||
|
AND t.id > sqlc.arg(after_id)
|
||||||
|
ORDER BY t.id
|
||||||
|
LIMIT sqlc.arg(batch_limit);
|
||||||
|
|
||||||
|
-- name: GetFingerprintCoverage :one
|
||||||
|
-- The admin gauge for the backfill. fingerprinted + rejected + pending = total.
|
||||||
|
-- rejected is a row at the current version with a NULL half: a tool ran and
|
||||||
|
-- refused the file, which is settled rather than waiting. Missing tracks are
|
||||||
|
-- excluded, or the gauge could never reach the end.
|
||||||
|
SELECT count(*)::bigint AS total,
|
||||||
|
count(*) FILTER (
|
||||||
|
WHERE f.fingerprint_version >= sqlc.arg(current_version)
|
||||||
|
AND f.audio_stream_sha256 IS NOT NULL AND f.chromaprint IS NOT NULL
|
||||||
|
)::bigint AS fingerprinted,
|
||||||
|
count(*) FILTER (
|
||||||
|
WHERE f.fingerprint_version >= sqlc.arg(current_version)
|
||||||
|
AND (f.audio_stream_sha256 IS NULL OR f.chromaprint IS NULL)
|
||||||
|
)::bigint AS rejected,
|
||||||
|
count(*) FILTER (
|
||||||
|
WHERE f.track_id IS NULL OR f.fingerprint_version < sqlc.arg(current_version)
|
||||||
|
)::bigint AS pending
|
||||||
|
FROM tracks t
|
||||||
|
LEFT JOIN track_fingerprints f ON f.track_id = t.id
|
||||||
|
WHERE t.missing_since IS NULL;
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import (
|
|||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"log/slog"
|
||||||
"os/exec"
|
"os/exec"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -148,38 +149,55 @@ func (s *Scanner) fingerprintFile(ctx context.Context, path string) fingerprintR
|
|||||||
return s.fingerprint(ctx, path)
|
return s.fingerprint(ctx, path)
|
||||||
}
|
}
|
||||||
|
|
||||||
// storeFingerprint records one attempt for a track whose bytes are new or have
|
// fingerprintOutcome is what storeFingerprint did with one attempt.
|
||||||
// changed. It never fails the scan: a missing fingerprint only keeps a track
|
type fingerprintOutcome int
|
||||||
// out of duplicate detection, which is not worth dropping the track over.
|
|
||||||
func (s *Scanner) storeFingerprint(
|
const (
|
||||||
ctx context.Context, q *dbq.Queries, trackID pgtype.UUID, path string, fp fingerprintResult,
|
outcomeFingerprinted fingerprintOutcome = iota // both halves stored
|
||||||
) {
|
outcomeRejected // stored with a NULL half: a verdict
|
||||||
|
outcomeInconclusive // nothing stored; worth trying again
|
||||||
|
outcomeStoreFailed // the write itself failed
|
||||||
|
)
|
||||||
|
|
||||||
|
// storeFingerprint records one attempt, for the scan (new or changed bytes) and
|
||||||
|
// the backfill (#3908) alike, so there is one rule for what gets written. It
|
||||||
|
// never fails its caller: a missing fingerprint only keeps a track out of
|
||||||
|
// duplicate detection, which is not worth dropping a scan or a pass over.
|
||||||
|
func storeFingerprint(
|
||||||
|
ctx context.Context, q *dbq.Queries, logger *slog.Logger,
|
||||||
|
trackID pgtype.UUID, path string, fp fingerprintResult,
|
||||||
|
) fingerprintOutcome {
|
||||||
if fp.hashErr != nil {
|
if fp.hashErr != nil {
|
||||||
s.logger.Warn("library scan: audio stream hash failed", "path", path, "err", fp.hashErr)
|
logger.Warn("fingerprint: audio stream hash failed", "path", path, "err", fp.hashErr)
|
||||||
}
|
}
|
||||||
if fp.printErr != nil {
|
if fp.printErr != nil {
|
||||||
s.logger.Warn("library scan: chromaprint failed", "path", path, "err", fp.printErr)
|
logger.Warn("fingerprint: chromaprint failed", "path", path, "err", fp.printErr)
|
||||||
}
|
}
|
||||||
if fp.inconclusive() {
|
if fp.inconclusive() {
|
||||||
// Any row this track holds describes its PREVIOUS bytes. Drop it and
|
// Any row this track holds describes bytes we could not confirm — the
|
||||||
// leave the track to the backfill, rather than stamping a failure that
|
// previous bytes for the scan, an older derivation for the backfill.
|
||||||
// says nothing about this file.
|
// Drop it rather than stamp a failure that says nothing about the file.
|
||||||
if err := q.DeleteTrackFingerprint(ctx, trackID); err != nil {
|
if err := q.DeleteTrackFingerprint(ctx, trackID); err != nil {
|
||||||
s.logger.Warn("library scan: clearing stale fingerprint failed", "path", path, "err", err)
|
logger.Warn("fingerprint: clearing stale fingerprint failed", "path", path, "err", err)
|
||||||
}
|
}
|
||||||
return
|
return outcomeInconclusive
|
||||||
}
|
}
|
||||||
// A NULL half here is a verdict — the tool ran and rejected this file — and
|
// A NULL half here is a verdict — the tool ran and rejected this file — and
|
||||||
// is stamped at the current version so the backfill does not retry it on
|
// is stamped at the current version so the backfill does not retry it on
|
||||||
// every boot. It is retried when the file changes.
|
// every pass. It is retried when the file changes.
|
||||||
if err := q.UpsertTrackFingerprint(ctx, dbq.UpsertTrackFingerprintParams{
|
if err := q.UpsertTrackFingerprint(ctx, dbq.UpsertTrackFingerprintParams{
|
||||||
TrackID: trackID,
|
TrackID: trackID,
|
||||||
AudioStreamSha256: fp.streamSHA256,
|
AudioStreamSha256: fp.streamSHA256,
|
||||||
Chromaprint: fp.chromaprint,
|
Chromaprint: fp.chromaprint,
|
||||||
FingerprintVersion: fingerprintVersion,
|
FingerprintVersion: fingerprintVersion,
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
s.logger.Warn("library scan: storing fingerprint failed", "path", path, "err", err)
|
logger.Warn("fingerprint: storing fingerprint failed", "path", path, "err", err)
|
||||||
|
return outcomeStoreFailed
|
||||||
}
|
}
|
||||||
|
if fp.hashErr != nil || fp.printErr != nil {
|
||||||
|
return outcomeRejected
|
||||||
|
}
|
||||||
|
return outcomeFingerprinted
|
||||||
}
|
}
|
||||||
|
|
||||||
// computeAudioStreamSHA256 returns the SHA-256 of the file's encoded audio.
|
// computeAudioStreamSHA256 returns the SHA-256 of the file's encoded audio.
|
||||||
|
|||||||
@@ -0,0 +1,196 @@
|
|||||||
|
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)
|
||||||
|
}
|
||||||
@@ -0,0 +1,156 @@
|
|||||||
|
package library
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"log/slog"
|
||||||
|
"path/filepath"
|
||||||
|
"sync"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestFingerprintBackfill_Integration pins which tracks a pass touches, that a
|
||||||
|
// pass ends, and that the coverage gauge counts what the pass wrote.
|
||||||
|
func TestFingerprintBackfill_Integration(t *testing.T) {
|
||||||
|
pool := newPool(t)
|
||||||
|
ctx := context.Background()
|
||||||
|
q := dbq.New(pool)
|
||||||
|
dir := t.TempDir()
|
||||||
|
|
||||||
|
_, album, artist := seedTrack(t, pool, filepath.Join(dir, "unfingerprinted.mp3"))
|
||||||
|
addTrack := func(name string) dbq.Track {
|
||||||
|
t.Helper()
|
||||||
|
tr, err := q.UpsertTrack(ctx, dbq.UpsertTrackParams{
|
||||||
|
Title: name, AlbumID: album.ID, ArtistID: artist.ID,
|
||||||
|
DurationMs: 1000, FilePath: filepath.Join(dir, name+".mp3"), FileSize: 100, FileFormat: "mp3",
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("track %s: %v", name, err)
|
||||||
|
}
|
||||||
|
return tr
|
||||||
|
}
|
||||||
|
current := addTrack("current")
|
||||||
|
stale := addTrack("stale")
|
||||||
|
missing := addTrack("missing")
|
||||||
|
|
||||||
|
sum := bytes.Repeat([]byte{0xCD}, 32)
|
||||||
|
for _, seed := range []struct {
|
||||||
|
track dbq.Track
|
||||||
|
version int16
|
||||||
|
}{
|
||||||
|
{current, fingerprintVersion},
|
||||||
|
{stale, fingerprintVersion - 1},
|
||||||
|
} {
|
||||||
|
if err := q.UpsertTrackFingerprint(ctx, dbq.UpsertTrackFingerprintParams{
|
||||||
|
TrackID: seed.track.ID, AudioStreamSha256: sum, Chromaprint: []int32{1},
|
||||||
|
FingerprintVersion: seed.version,
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("seed fingerprint: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if _, err := pool.Exec(ctx, "UPDATE tracks SET missing_since = now() WHERE id = $1", missing.ID); err != nil {
|
||||||
|
t.Fatalf("mark missing: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
var mu sync.Mutex
|
||||||
|
calls := map[string]int{}
|
||||||
|
w := NewFingerprintBackfillWorker(pool, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||||
|
// A batch of one forces the keyset cursor across several queries in a pass.
|
||||||
|
w.batch = 1
|
||||||
|
w.fingerprint = func(_ context.Context, path string) fingerprintResult {
|
||||||
|
name := filepath.Base(path)
|
||||||
|
mu.Lock()
|
||||||
|
calls[name]++
|
||||||
|
mu.Unlock()
|
||||||
|
switch name {
|
||||||
|
case "stall.mp3":
|
||||||
|
return fingerprintResult{streamSHA256: sum, printErr: fmt.Errorf("fpcalc: %w", errFingerprintTimeout)}
|
||||||
|
case "rejected.mp3":
|
||||||
|
return fingerprintResult{hashErr: errors.New("ffmpeg exited 1"), printErr: errors.New("fpcalc exited 2")}
|
||||||
|
default:
|
||||||
|
return fingerprintResult{streamSHA256: sum, chromaprint: []int32{7, -7}}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
callCount := func(name string) int {
|
||||||
|
mu.Lock()
|
||||||
|
defer mu.Unlock()
|
||||||
|
return calls[name]
|
||||||
|
}
|
||||||
|
|
||||||
|
// 1. Only the track with no row and the stale one are fingerprinted — never
|
||||||
|
// the current one, never the missing one.
|
||||||
|
res, err := w.pass(ctx)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("first pass: %v", err)
|
||||||
|
}
|
||||||
|
if res.Processed != 2 || res.Fingerprinted != 2 {
|
||||||
|
t.Fatalf("first pass = %+v, want 2 processed, 2 fingerprinted", res)
|
||||||
|
}
|
||||||
|
for name, want := range map[string]int{
|
||||||
|
"unfingerprinted.mp3": 1, "stale.mp3": 1, "current.mp3": 0, "missing.mp3": 0,
|
||||||
|
} {
|
||||||
|
if got := callCount(name); got != want {
|
||||||
|
t.Errorf("%s fingerprinted %d times, want %d", name, got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 2. A pass after a complete one is a no-op. A backfill that redoes its work
|
||||||
|
// every hour is the expensive way this could be wrong.
|
||||||
|
res, err = w.pass(ctx)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("second pass: %v", err)
|
||||||
|
}
|
||||||
|
if res.Processed != 0 {
|
||||||
|
t.Fatalf("second pass processed %d tracks, want 0", res.Processed)
|
||||||
|
}
|
||||||
|
|
||||||
|
// 3. An inconclusive file is tried exactly once and the pass ENDS. Without the
|
||||||
|
// keyset cursor it would be re-listed immediately and this call would never
|
||||||
|
// return.
|
||||||
|
addTrack("stall")
|
||||||
|
addTrack("rejected")
|
||||||
|
res, err = w.pass(ctx)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("third pass: %v", err)
|
||||||
|
}
|
||||||
|
if res.Processed != 2 || res.Inconclusive != 1 || res.Rejected != 1 {
|
||||||
|
t.Fatalf("third pass = %+v, want 2 processed, 1 inconclusive, 1 rejected", res)
|
||||||
|
}
|
||||||
|
if got := callCount("stall.mp3"); got != 1 {
|
||||||
|
t.Fatalf("stalling file tried %d times in one pass, want exactly 1", got)
|
||||||
|
}
|
||||||
|
|
||||||
|
// 4. The gauge counts what the passes wrote, and its buckets add up.
|
||||||
|
cov, err := FingerprintCoverage(ctx, pool)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("coverage: %v", err)
|
||||||
|
}
|
||||||
|
// Five present tracks: unfingerprinted, current, stale, stall, rejected.
|
||||||
|
// The missing track is not counted.
|
||||||
|
if cov.Total != 5 || cov.Fingerprinted != 3 || cov.Rejected != 1 || cov.Pending != 1 {
|
||||||
|
t.Errorf("coverage = %+v, want total 5, fingerprinted 3, rejected 1, pending 1", cov)
|
||||||
|
}
|
||||||
|
if cov.Fingerprinted+cov.Rejected+cov.Pending != cov.Total {
|
||||||
|
t.Errorf("coverage buckets %+v do not sum to the total", cov)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestBackfillFingerprintsResult_Add(t *testing.T) {
|
||||||
|
var r BackfillFingerprintsResult
|
||||||
|
for _, o := range []fingerprintOutcome{
|
||||||
|
outcomeFingerprinted, outcomeFingerprinted, outcomeRejected, outcomeInconclusive, outcomeStoreFailed,
|
||||||
|
} {
|
||||||
|
r.add(o)
|
||||||
|
}
|
||||||
|
// A failed write stored nothing, so like an inconclusive attempt it is
|
||||||
|
// tried again next pass — and counts as such.
|
||||||
|
want := BackfillFingerprintsResult{Processed: 5, Fingerprinted: 2, Rejected: 1, Inconclusive: 2}
|
||||||
|
if r != want {
|
||||||
|
t.Errorf("tally = %+v, want %+v", r, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -384,7 +384,7 @@ func (s *Scanner) scanFile(
|
|||||||
s.logger.Warn("library scan: LogChange track upsert failed", "track_id", track.ID, "err", err)
|
s.logger.Warn("library scan: LogChange track upsert failed", "track_id", track.ID, "err", err)
|
||||||
}
|
}
|
||||||
if fingerprinted {
|
if fingerprinted {
|
||||||
s.storeFingerprint(ctx, q, track.ID, path, fp)
|
storeFingerprint(ctx, q, s.logger, track.ID, path, fp)
|
||||||
}
|
}
|
||||||
|
|
||||||
if knownTrack {
|
if knownTrack {
|
||||||
|
|||||||
@@ -0,0 +1,32 @@
|
|||||||
|
import { describe, it, expect, vi, beforeEach } from 'vitest';
|
||||||
|
import { getFingerprintCoverage, type FingerprintCoverage } from './admin';
|
||||||
|
|
||||||
|
vi.mock('./client', () => ({
|
||||||
|
api: { get: vi.fn(), post: vi.fn() }
|
||||||
|
}));
|
||||||
|
|
||||||
|
import { api } from './client';
|
||||||
|
|
||||||
|
describe('admin fingerprint coverage API', () => {
|
||||||
|
beforeEach(() => vi.clearAllMocks());
|
||||||
|
|
||||||
|
it('getFingerprintCoverage GETs the correct path', async () => {
|
||||||
|
const sample: FingerprintCoverage = {
|
||||||
|
total: 18026,
|
||||||
|
fingerprinted: 9400,
|
||||||
|
rejected: 12,
|
||||||
|
pending: 8614
|
||||||
|
};
|
||||||
|
(api.get as unknown as ReturnType<typeof vi.fn>).mockResolvedValueOnce(sample);
|
||||||
|
const got = await getFingerprintCoverage();
|
||||||
|
expect(api.get).toHaveBeenCalledWith('/api/admin/library/fingerprints');
|
||||||
|
expect(got).toEqual(sample);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('buckets sum to the total', async () => {
|
||||||
|
const sample: FingerprintCoverage = { total: 10, fingerprinted: 6, rejected: 1, pending: 3 };
|
||||||
|
(api.get as unknown as ReturnType<typeof vi.fn>).mockResolvedValueOnce(sample);
|
||||||
|
const got = await getFingerprintCoverage();
|
||||||
|
expect(got.fingerprinted + got.rejected + got.pending).toBe(got.total);
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -313,6 +313,30 @@ export function createCoverageQuery() {
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Fingerprint backfill (#3908) --------------------------------------------
|
||||||
|
|
||||||
|
export type FingerprintCoverage = {
|
||||||
|
total: number;
|
||||||
|
fingerprinted: number;
|
||||||
|
rejected: number;
|
||||||
|
pending: number;
|
||||||
|
};
|
||||||
|
|
||||||
|
export async function getFingerprintCoverage(): Promise<FingerprintCoverage> {
|
||||||
|
return api.get<FingerprintCoverage>('/api/admin/library/fingerprints');
|
||||||
|
}
|
||||||
|
|
||||||
|
// Polled far less often than the cover gauge: the backfill decodes files two at
|
||||||
|
// a time, so the count moves by a few tracks a minute and a 3s poll is noise.
|
||||||
|
export function createFingerprintCoverageQuery() {
|
||||||
|
return createQuery({
|
||||||
|
queryKey: qk.fingerprintCoverage(),
|
||||||
|
queryFn: getFingerprintCoverage,
|
||||||
|
staleTime: 30_000,
|
||||||
|
refetchInterval: 30_000
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
// Cover-art providers ------------------------------------------------------
|
// Cover-art providers ------------------------------------------------------
|
||||||
|
|
||||||
export type CoverProviderCapability = 'album_cover' | 'artist_thumb' | 'artist_fanart';
|
export type CoverProviderCapability = 'album_cover' | 'artist_thumb' | 'artist_fanart';
|
||||||
|
|||||||
@@ -47,6 +47,7 @@ export const qk = {
|
|||||||
['adminPlaybackErrors', { resolved: resolved ?? false }] as const,
|
['adminPlaybackErrors', { resolved: resolved ?? false }] as const,
|
||||||
scanStatus: () => ['scanStatus'] as const,
|
scanStatus: () => ['scanStatus'] as const,
|
||||||
coverage: () => ['coverage'] as const,
|
coverage: () => ['coverage'] as const,
|
||||||
|
fingerprintCoverage: () => ['fingerprintCoverage'] as const,
|
||||||
coverProviders: () => ['coverProviders'] as const,
|
coverProviders: () => ['coverProviders'] as const,
|
||||||
tagProviders: () => ['tagProviders'] as const,
|
tagProviders: () => ['tagProviders'] as const,
|
||||||
adminUsers: () => ['adminUsers'] as const,
|
adminUsers: () => ['adminUsers'] as const,
|
||||||
|
|||||||
@@ -10,6 +10,7 @@
|
|||||||
createAdminQuarantineQuery,
|
createAdminQuarantineQuery,
|
||||||
createScanStatusQuery,
|
createScanStatusQuery,
|
||||||
createCoverageQuery,
|
createCoverageQuery,
|
||||||
|
createFingerprintCoverageQuery,
|
||||||
approveRequest,
|
approveRequest,
|
||||||
rejectRequest,
|
rejectRequest,
|
||||||
resolveQuarantine,
|
resolveQuarantine,
|
||||||
@@ -177,6 +178,11 @@
|
|||||||
const coverageQ = $derived($coverageStore);
|
const coverageQ = $derived($coverageStore);
|
||||||
const coverage = $derived(coverageQ.data);
|
const coverage = $derived(coverageQ.data);
|
||||||
|
|
||||||
|
// ---- Fingerprint backfill gauge (#3908) ----
|
||||||
|
const fingerprintStore = $derived(createFingerprintCoverageQuery());
|
||||||
|
const fingerprintQ = $derived($fingerprintStore);
|
||||||
|
const fingerprints = $derived(fingerprintQ.data);
|
||||||
|
|
||||||
let triggering = $state(false);
|
let triggering = $state(false);
|
||||||
let triggerResult = $state<string | null>(null);
|
let triggerResult = $state<string | null>(null);
|
||||||
|
|
||||||
@@ -428,6 +434,28 @@
|
|||||||
{#if triggerResult}
|
{#if triggerResult}
|
||||||
<p class="mt-2 text-sm">{triggerResult}</p>
|
<p class="mt-2 text-sm">{triggerResult}</p>
|
||||||
{/if}
|
{/if}
|
||||||
|
|
||||||
|
<!-- Fingerprint backfill (#3908). A worker of its own rather than a scan
|
||||||
|
stage, so its progress is read live here, not from the run above. -->
|
||||||
|
{#if fingerprints && fingerprints.total > 0}
|
||||||
|
<div class="mt-3 flex flex-wrap items-center gap-3 text-sm">
|
||||||
|
<span class="text-xs font-medium uppercase tracking-wide text-text-muted">Fingerprints</span>
|
||||||
|
<span>{fingerprints.fingerprinted.toLocaleString()} of {fingerprints.total.toLocaleString()} tracks</span>
|
||||||
|
{#if fingerprints.pending > 0}
|
||||||
|
<span class="text-text-muted">·</span>
|
||||||
|
<span>{fingerprints.pending.toLocaleString()} pending</span>
|
||||||
|
{/if}
|
||||||
|
{#if fingerprints.rejected > 0}
|
||||||
|
<span class="text-text-muted">·</span>
|
||||||
|
<span
|
||||||
|
class="cursor-help"
|
||||||
|
title="The fingerprint tools could not read these files. Each is tried again when its file changes."
|
||||||
|
>
|
||||||
|
{fingerprints.rejected.toLocaleString()} unreadable
|
||||||
|
</span>
|
||||||
|
{/if}
|
||||||
|
</div>
|
||||||
|
{/if}
|
||||||
</section>
|
</section>
|
||||||
|
|
||||||
<!-- Cover art bulk refetch -->
|
<!-- Cover art bulk refetch -->
|
||||||
|
|||||||
Reference in New Issue
Block a user