feat(library): the duplicate sweep — propose duplicate groups from fingerprints (M400 #3910)

Reads fingerprints, runs them through the matcher, and records proposals in duplicate_groups (migration 0059). Nothing is merged or deleted: a group is a proposal for the admin report (#3912). Streaming. The whole library's fingerprints are hundreds of megabytes, but tracks are only compared within 3s of each other in duration. So candidates stream in (duration_ms, id) order, keyset-paged on a new tracks(duration_ms, id) index. The grouper holds only the tracks within 3s of the oldest one not yet settled. A seed is settled once a track arrives beyond its window, which gives the same result as grouping the whole sorted list. groupDuplicates is rebuilt on the same streamGrouper, so there is one grouping rule and the #3909 tests still cover it. Each fingerprint's alignment index and variety check are computed once instead of for every pair. Exact duplicates are grouped library-wide in SQL. The first member the stream meets stands in for the whole group in the acoustic pass. An exact group caught in an oversize acoustic cluster is still proposed: the acoustic evidence is discarded, identical bytes are not. Re-sweeping: - a group is identified by its sorted member ids, so finding it again refreshes the row in place - a proposal whose members all sat in one dismissed group is not proposed again (a subset repeats the verdict; a superset is new evidence) - a pending proposal no sweep has found again is retired, but only after a complete sweep, and only if an earlier sweep last confirmed it, so two overlapping sweeps cannot delete each other's findings - dismissals are kept DuplicateSweepWorker checks hourly and sweeps only when a fingerprint was written after the last sweep started. TryStartDuplicateSweep guards against two sweeps at once and reaps one stuck in flight for 2h. The sweep row is closed on a detached context with a deadline, so a sweep cancelled at shutdown still records that it ended. The integration test pages one row at a time and checks: - an acoustic pair and an exact pair are found - a track with no fingerprint, a missing track and a near-duration unrelated song are left out - a dismissed group is suppressed while the pending one refreshes without duplicating - a proposal that stops holding is retired and the dismissal survives Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SQ31KQpYbStyK5y58UmPLH
2026-09-11 17:00:07 -04:00
parent c06af48cd6
commit 6379b6c31d
10 changed files with 1264 additions and 146 deletions
@@ -0,0 +1,360 @@
+package library
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
+	syncpkg "git.fabledsword.com/bvandeusen/minstrel/internal/sync"
+)
+
+// Duplicate sweep (M400 #3910).
+//
+// Reads fingerprints, runs them through the matcher and records what it proposes
+// in duplicate_groups. It never merges or deletes anything the operator has not
+// asked for: a group is a proposal, reviewed in the admin report (#3912).
+//
+// Blocked on duration, deliberately not on title: the #3885 pair are titled
+// "WWW" and "WWW (instrumental)", so a title block would have missed the case
+// that started the milestone. Candidates stream in (duration_ms, id) order and
+// the grouper holds only a few seconds of durations at a time.
+
+// duplicateCandidatePage is how many candidates one query returns. Each row
+// carries a ~4 KB fingerprint, so a page is about 2 MB.
+const duplicateCandidatePage = 500
+
+// duplicateSweepTick is how often the worker checks for anything new to sweep.
+// With nothing new, a tick is two cheap aggregate queries.
+const duplicateSweepTick = time.Hour
+
+// staleDuplicateSweepThreshold is the age past which an in-flight sweep is
+// assumed dead — a crash mid-sweep leaves finished_at NULL for ever — and another
+// may start. Twice the library scan's threshold, because a sweep compares
+// fingerprints across the whole library and can legitimately run long on a big
+// one.
+const staleDuplicateSweepThreshold = 2 * time.Hour
+
+// duplicateSweepFinishTimeout bounds recording that a sweep ended. It runs on a
+// context detached from the sweep's own, so a sweep cancelled at shutdown still
+// closes its row rather than leaving it in flight until the reaper.
+const duplicateSweepFinishTimeout = 10 * time.Second
+
+// DuplicateSweepResult tallies one sweep.
+type DuplicateSweepResult struct {
+	Candidates int // tracks with a chromaprint that were streamed
+	Groups     int // groups the matcher found
+	Proposed   int // written as pending, new or refreshed
+	Suppressed int // not proposed: already dismissed or resolved by the operator
+	Retired    int // pending proposals this sweep did not find again, removed
+	Oversize   int // acoustic clusters too large to propose
+}
+
+// RunDuplicateSweep runs one sweep and records it in duplicate_sweeps.
+func RunDuplicateSweep(ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger) (DuplicateSweepResult, error) {
+	return runDuplicateSweep(ctx, pool, logger, duplicateCandidatePage)
+}
+
+func runDuplicateSweep(
+	ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, pageSize int32,
+) (DuplicateSweepResult, error) {
+	q := dbq.New(pool)
+	sweep, err := q.StartDuplicateSweep(ctx)
+	if err != nil {
+		return DuplicateSweepResult{}, fmt.Errorf("start duplicate sweep: %w", err)
+	}
+
+	res, runErr := sweepDuplicates(ctx, q, sweep.ID, pageSize)
+
+	finishCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), duplicateSweepFinishTimeout)
+	defer cancel()
+	errMsg := ""
+	if runErr != nil {
+		errMsg = runErr.Error()
+	}
+	candidates, groups, oversize := int32(res.Candidates), int32(res.Groups), int32(res.Oversize)
+	if ferr := q.FinishDuplicateSweep(finishCtx, dbq.FinishDuplicateSweepParams{
+		ID: sweep.ID, Candidates: &candidates, GroupsFound: &groups, OversizeClusters: &oversize,
+		ErrorMessage: errMsg,
+	}); ferr != nil {
+		logger.Error("duplicate sweep: recording the end of the sweep failed", "err", ferr)
+		if runErr == nil {
+			runErr = fmt.Errorf("finish duplicate sweep: %w", ferr)
+		}
+	}
+
+	logger.Info("duplicate sweep complete",
+		"candidates", res.Candidates, "groups", res.Groups, "proposed", res.Proposed,
+		"suppressed", res.Suppressed, "retired", res.Retired, "oversize", res.Oversize, "err", runErr)
+	return res, runErr
+}
+
+func sweepDuplicates(
+	ctx context.Context, q *dbq.Queries, sweepID pgtype.UUID, pageSize int32,
+) (DuplicateSweepResult, error) {
+	var res DuplicateSweepResult
+
+	// Exact tier, library-wide, in one query.
+	exactRows, err := q.ListExactDuplicateHashes(ctx, fingerprintVersion)
+	if err != nil {
+		return res, fmt.Errorf("list exact duplicates: %w", err)
+	}
+	exactMembers := make([][]string, len(exactRows))
+	exactOf := map[string]int{}
+	for i, row := range exactRows {
+		exactMembers[i] = formatUUIDs(row.TrackIds)
+		for _, id := range exactMembers[i] {
+			exactOf[id] = i
+		}
+	}
+	exactSeen := make([]bool, len(exactRows))
+
+	// Acoustic tier, streamed in duration order. The first member of an exact
+	// group the stream meets stands in for the whole group; the rest are skipped.
+	grouper := newStreamGrouper(defaultAcousticMaxBitErrorRate)
+	params := dbq.ListDuplicateCandidatesParams{
+		CurrentVersion: fingerprintVersion,
+		// Durations are never negative, and the all-zero uuid sorts first: every
+		// row is after this cursor. Valid must be true, or "> NULL" matches nothing.
+		AfterDurationMs: -1,
+		AfterID:         pgtype.UUID{Valid: true},
+		PageLimit:       pageSize,
+	}
+	for {
+		if err := ctx.Err(); err != nil {
+			return res, err
+		}
+		rows, err := q.ListDuplicateCandidates(ctx, params)
+		if err != nil {
+			return res, fmt.Errorf("list duplicate candidates: %w", err)
+		}
+		for _, row := range rows {
+			res.Candidates++
+			id := syncpkg.FormatUUID(row.ID)
+			unit := &groupUnit{ids: []string{id}, durationMs: row.DurationMs, sortKey: id}
+			if gi, ok := exactOf[id]; ok {
+				if exactSeen[gi] {
+					continue
+				}
+				exactSeen[gi] = true
+				unit.ids, unit.exact = exactMembers[gi], true
+			}
+			unit.print = preparePrint(row.Chromaprint)
+			grouper.push(unit)
+		}
+		if int32(len(rows)) < pageSize {
+			break
+		}
+		last := rows[len(rows)-1]
+		params.AfterDurationMs, params.AfterID = last.DurationMs, last.ID
+	}
+
+	found := grouper.finish()
+	// Exact groups none of whose members has a chromaprint never reached the
+	// stream. Identical bytes need no acoustic evidence.
+	for gi, seen := range exactSeen {
+		if !seen {
+			found.Groups = append(found.Groups, duplicateGroup{Tier: tierExact, Members: exactMembers[gi]})
+		}
+	}
+	sortGroups(found.Groups)
+	res.Groups, res.Oversize = len(found.Groups), found.OversizeClusters
+
+	dismissed, err := q.ListDismissedDuplicateMemberSets(ctx)
+	if err != nil {
+		return res, fmt.Errorf("list dismissed duplicate groups: %w", err)
+	}
+	dismissedSets := make([]map[string]struct{}, 0, len(dismissed))
+	for _, d := range dismissed {
+		set := map[string]struct{}{}
+		for _, id := range formatUUIDs(d.TrackIds) {
+			set[id] = struct{}{}
+		}
+		dismissedSets = append(dismissedSets, set)
+	}
+
+	for _, group := range found.Groups {
+		if coveredByDismissal(group.Members, dismissedSets) {
+			res.Suppressed++
+			continue
+		}
+		up := dbq.UpsertDuplicateGroupParams{
+			MemberKey: strings.Join(group.Members, ","),
+			Tier:      string(group.Tier),
+			SweepID:   sweepID,
+		}
+		if group.Tier == tierAcoustic {
+			worst := float32(group.WorstBitErrorRate)
+			up.WorstBitErrorRate = &worst
+		}
+		groupID, err := q.UpsertDuplicateGroup(ctx, up)
+		if errors.Is(err, pgx.ErrNoRows) {
+			// This exact member set was already dismissed or merged.
+			res.Suppressed++
+			continue
+		}
+		if err != nil {
+			return res, fmt.Errorf("upsert duplicate group: %w", err)
+		}
+		for _, id := range group.Members {
+			var trackID pgtype.UUID
+			if err := trackID.Scan(id); err != nil {
+				return res, fmt.Errorf("parse track id %q: %w", id, err)
+			}
+			if err := q.AddDuplicateGroupMember(ctx, dbq.AddDuplicateGroupMemberParams{
+				GroupID: groupID, TrackID: trackID,
+			}); err != nil {
+				return res, fmt.Errorf("add duplicate group member: %w", err)
+			}
+		}
+		res.Proposed++
+	}
+
+	// Only after a complete sweep: a sweep that failed partway has no basis for
+	// concluding that anything it did not reach has gone.
+	retired, err := q.DeleteStalePendingDuplicateGroups(ctx, sweepID)
+	if err != nil {
+		return res, fmt.Errorf("retire stale duplicate groups: %w", err)
+	}
+	res.Retired = int(retired)
+	return res, nil
+}
+
+// coveredByDismissal reports whether every member of a proposal sat together in
+// one group the operator dismissed. A subset counts: dismissing {A, B, C} said
+// none of them are copies of each other, so proposing {A, B} again would be
+// asking the same question twice. A superset does not count: a new copy joining
+// is new evidence, and worth asking about.
+func coveredByDismissal(members []string, dismissed []map[string]struct{}) bool {
+	for _, set := range dismissed {
+		covered := true
+		for _, id := range members {
+			if _, ok := set[id]; !ok {
+				covered = false
+				break
+			}
+		}
+		if covered {
+			return true
+		}
+	}
+	return false
+}
+
+func formatUUIDs(ids []pgtype.UUID) []string {
+	out := make([]string, len(ids))
+	for i, id := range ids {
+		out[i] = syncpkg.FormatUUID(id)
+	}
+	sort.Strings(out)
+	return out
+}
+
+// TryStartDuplicateSweep starts a sweep in the background unless one is already
+// running, reaping a sweep that has been in flight past
+// staleDuplicateSweepThreshold. Mirrors TryStartScan. The sweep runs on ctx, so
+// a caller answering an HTTP request must pass a context that outlives it.
+func TryStartDuplicateSweep(ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger) (bool, error) {
+	q := dbq.New(pool)
+	row, err := q.GetInFlightDuplicateSweep(ctx)
+	switch {
+	case err == nil:
+		age := time.Since(row.StartedAt.Time)
+		if age <= staleDuplicateSweepThreshold {
+			return false, nil
+		}
+		logger.Warn("reaping stale duplicate sweep", "id", syncpkg.FormatUUID(row.ID), "age", age)
+		if ferr := q.FinishDuplicateSweep(ctx, dbq.FinishDuplicateSweepParams{
+			ID: row.ID, ErrorMessage: "reaped (stale)",
+		}); ferr != nil {
+			return false, fmt.Errorf("reap stale duplicate sweep: %w", ferr)
+		}
+	case !errors.Is(err, pgx.ErrNoRows):
+		return false, fmt.Errorf("duplicate sweep in-flight check: %w", err)
+	}
+
+	go func() {
+		if _, err := RunDuplicateSweep(ctx, pool, logger); err != nil {
+			logger.Warn("duplicate sweep failed", "err", err)
+		}
+	}()
+	return true, nil
+}
+
+// DuplicateSweepWorker sweeps whenever fingerprints have changed.
+type DuplicateSweepWorker struct {
+	pool   *pgxpool.Pool
+	logger *slog.Logger
+	tick   time.Duration
+}
+
+// NewDuplicateSweepWorker builds a worker with the production cadence.
+func NewDuplicateSweepWorker(pool *pgxpool.Pool, logger *slog.Logger) *DuplicateSweepWorker {
+	return &DuplicateSweepWorker{pool: pool, logger: logger, tick: duplicateSweepTick}
+}
+
+// Run blocks until ctx is cancelled, checking once at start and then each tick.
+func (w *DuplicateSweepWorker) Run(ctx context.Context) {
+	w.tickOnce(ctx)
+	t := time.NewTicker(w.tick)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			w.tickOnce(ctx)
+		}
+	}
+}
+
+// tickOnce contains one check so nothing it does can stop the next tick (rule 157).
+func (w *DuplicateSweepWorker) tickOnce(ctx context.Context) {
+	defer func() {
+		if r := recover(); r != nil {
+			w.logger.Error("duplicate sweep: tick panicked", "panic", r)
+		}
+	}()
+	due, err := duplicateSweepDue(ctx, dbq.New(w.pool))
+	if err != nil {
+		if ctx.Err() == nil {
+			w.logger.Warn("duplicate sweep: due check failed", "err", err)
+		}
+		return
+	}
+	if !due {
+		return
+	}
+	if _, err := TryStartDuplicateSweep(ctx, w.pool, w.logger); err != nil {
+		w.logger.Warn("duplicate sweep: start failed", "err", err)
+	}
+}
+
+// duplicateSweepDue reports whether any fingerprint was written after the latest
+// sweep started. Fingerprints are the sweep's only input, so nothing else can
+// change its answer; while the backfill is running this is true every tick.
+func duplicateSweepDue(ctx context.Context, q *dbq.Queries) (bool, error) {
+	latest, err := q.GetLatestFingerprintComputedAt(ctx)
+	if err != nil {
+		return false, fmt.Errorf("latest fingerprint: %w", err)
+	}
+	if !latest.Valid {
+		return false, nil // nothing fingerprinted yet
+	}
+	last, err := q.GetLatestDuplicateSweep(ctx)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return true, nil
+	}
+	if err != nil {
+		return false, fmt.Errorf("latest duplicate sweep: %w", err)
+	}
+	return latest.Time.After(last.StartedAt.Time), nil
+}