From 6379b6c31dcdd089dad789ed34d286a7ea02194b Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 11 Sep 2026 17:00:07 -0400 Subject: [PATCH] =?UTF-8?q?feat(library):=20the=20duplicate=20sweep=20?= =?UTF-8?q?=E2=80=94=20propose=20duplicate=20groups=20from=20fingerprints?= =?UTF-8?q?=20(M400=20#3910)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) Claude-Session: https://claude.ai/code/session_01SQ31KQpYbStyK5y58UmPLH --- cmd/minstrel/main.go | 5 + internal/db/dbq/duplicates.sql.go | 320 +++++++++++++++ internal/db/dbq/models.go | 26 ++ .../migrations/0059_duplicate_groups.down.sql | 4 + .../migrations/0059_duplicate_groups.up.sql | 54 +++ internal/db/queries/duplicates.sql | 100 +++++ internal/dbtest/reset.go | 3 + internal/library/duplicate_match.go | 365 +++++++++++------- internal/library/duplicate_sweep.go | 360 +++++++++++++++++ internal/library/duplicate_sweep_test.go | 173 +++++++++ 10 files changed, 1264 insertions(+), 146 deletions(-) create mode 100644 internal/db/dbq/duplicates.sql.go create mode 100644 internal/db/migrations/0059_duplicate_groups.down.sql create mode 100644 internal/db/migrations/0059_duplicate_groups.up.sql create mode 100644 internal/db/queries/duplicates.sql create mode 100644 internal/library/duplicate_sweep.go create mode 100644 internal/library/duplicate_sweep_test.go diff --git a/cmd/minstrel/main.go b/cmd/minstrel/main.go index aa57a6a7..6ad3e0ec 100644 --- a/cmd/minstrel/main.go +++ b/cmd/minstrel/main.go @@ -220,6 +220,11 @@ func run() error { // internal/library/fingerprint_backfill.go for why. go library.NewFingerprintBackfillWorker(pool, logger.With("component", "fingerprint_backfill")).Run(ctx) + // Duplicate sweep (M400 #3910): proposes groups of tracks holding one + // recording, from the fingerprints above. Sweeps only when fingerprints have + // changed since the last sweep. + go library.NewDuplicateSweepWorker(pool, logger.With("component", "duplicate_sweep")).Run(ctx) + // Start the tag-enrichment worker (#1490). Reconciles the compiled-in // tag providers with tag_provider_settings, bumps the sources version if // the provider set changed (re-opening settled rows), then drains tracks diff --git a/internal/db/dbq/duplicates.sql.go b/internal/db/dbq/duplicates.sql.go new file mode 100644 index 00000000..94a8cb0e --- /dev/null +++ b/internal/db/dbq/duplicates.sql.go @@ -0,0 +1,320 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: duplicates.sql + +package dbq + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const addDuplicateGroupMember = `-- name: AddDuplicateGroupMember :exec +INSERT INTO duplicate_group_members (group_id, track_id) +VALUES ($1, $2) +ON CONFLICT DO NOTHING +` + +type AddDuplicateGroupMemberParams struct { + GroupID pgtype.UUID + TrackID pgtype.UUID +} + +func (q *Queries) AddDuplicateGroupMember(ctx context.Context, arg AddDuplicateGroupMemberParams) error { + _, err := q.db.Exec(ctx, addDuplicateGroupMember, arg.GroupID, arg.TrackID) + return err +} + +const deleteStalePendingDuplicateGroups = `-- name: DeleteStalePendingDuplicateGroups :execrows +DELETE FROM duplicate_groups g + WHERE g.status = 'pending' + AND g.last_seen_sweep_id IS DISTINCT FROM $1 + AND (g.last_seen_sweep_id IS NULL + OR (SELECT s.started_at FROM duplicate_sweeps s WHERE s.id = g.last_seen_sweep_id) + < (SELECT s.started_at FROM duplicate_sweeps s WHERE s.id = $1)) +` + +// A pending proposal this sweep did not find again no longer describes the +// library: a member was re-fingerprinted, merged away or went missing. Dismissed +// groups are kept regardless — they are the memory of a decision. +// +// Only proposals last confirmed by an EARLIER sweep go. Should two sweeps ever +// overlap (a manual trigger racing the worker), neither may delete what the other +// has just found. +func (q *Queries) DeleteStalePendingDuplicateGroups(ctx context.Context, sweepID pgtype.UUID) (int64, error) { + result, err := q.db.Exec(ctx, deleteStalePendingDuplicateGroups, sweepID) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const finishDuplicateSweep = `-- name: FinishDuplicateSweep :exec +UPDATE duplicate_sweeps + SET finished_at = now(), + candidates = $1, + groups_found = $2, + oversize_clusters = $3, + error_message = NULLIF($4::text, '') + WHERE id = $5 +` + +type FinishDuplicateSweepParams struct { + Candidates *int32 + GroupsFound *int32 + OversizeClusters *int32 + ErrorMessage string + ID pgtype.UUID +} + +func (q *Queries) FinishDuplicateSweep(ctx context.Context, arg FinishDuplicateSweepParams) error { + _, err := q.db.Exec(ctx, finishDuplicateSweep, + arg.Candidates, + arg.GroupsFound, + arg.OversizeClusters, + arg.ErrorMessage, + arg.ID, + ) + return err +} + +const getInFlightDuplicateSweep = `-- name: GetInFlightDuplicateSweep :one +SELECT id, started_at + FROM duplicate_sweeps + WHERE finished_at IS NULL + ORDER BY started_at DESC + LIMIT 1 +` + +type GetInFlightDuplicateSweepRow struct { + ID pgtype.UUID + StartedAt pgtype.Timestamptz +} + +// The guard against two sweeps at once: "in flight" is finished_at IS NULL. +func (q *Queries) GetInFlightDuplicateSweep(ctx context.Context) (GetInFlightDuplicateSweepRow, error) { + row := q.db.QueryRow(ctx, getInFlightDuplicateSweep) + var i GetInFlightDuplicateSweepRow + err := row.Scan(&i.ID, &i.StartedAt) + return i, err +} + +const getLatestDuplicateSweep = `-- name: GetLatestDuplicateSweep :one +SELECT id, started_at, finished_at, candidates, groups_found, oversize_clusters, error_message + FROM duplicate_sweeps + ORDER BY started_at DESC + LIMIT 1 +` + +func (q *Queries) GetLatestDuplicateSweep(ctx context.Context) (DuplicateSweep, error) { + row := q.db.QueryRow(ctx, getLatestDuplicateSweep) + var i DuplicateSweep + err := row.Scan( + &i.ID, + &i.StartedAt, + &i.FinishedAt, + &i.Candidates, + &i.GroupsFound, + &i.OversizeClusters, + &i.ErrorMessage, + ) + return i, err +} + +const getLatestFingerprintComputedAt = `-- name: GetLatestFingerprintComputedAt :one +SELECT max(computed_at)::timestamptz AS latest FROM track_fingerprints +` + +// Whether a sweep has anything new to look at: fingerprints written since the +// last sweep started. +func (q *Queries) GetLatestFingerprintComputedAt(ctx context.Context) (pgtype.Timestamptz, error) { + row := q.db.QueryRow(ctx, getLatestFingerprintComputedAt) + var latest pgtype.Timestamptz + err := row.Scan(&latest) + return latest, err +} + +const listDismissedDuplicateMemberSets = `-- name: ListDismissedDuplicateMemberSets :many +SELECT g.id, array_agg(m.track_id ORDER BY m.track_id)::uuid[] AS track_ids + FROM duplicate_groups g + JOIN duplicate_group_members m ON m.group_id = g.id + WHERE g.status = 'dismissed' + GROUP BY g.id +` + +type ListDismissedDuplicateMemberSetsRow struct { + ID pgtype.UUID + TrackIds []pgtype.UUID +} + +// What the operator has already said are not duplicates. A new proposal whose +// every member sat together in one of these is not proposed again. +func (q *Queries) ListDismissedDuplicateMemberSets(ctx context.Context) ([]ListDismissedDuplicateMemberSetsRow, error) { + rows, err := q.db.Query(ctx, listDismissedDuplicateMemberSets) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListDismissedDuplicateMemberSetsRow + for rows.Next() { + var i ListDismissedDuplicateMemberSetsRow + if err := rows.Scan(&i.ID, &i.TrackIds); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listDuplicateCandidates = `-- name: ListDuplicateCandidates :many +SELECT t.id, t.duration_ms, f.audio_stream_sha256, f.chromaprint + FROM tracks t + JOIN track_fingerprints f ON f.track_id = t.id + WHERE t.missing_since IS NULL + AND f.fingerprint_version >= $1 + AND f.chromaprint IS NOT NULL + AND (t.duration_ms, t.id) > ($2::integer, $3::uuid) + ORDER BY t.duration_ms, t.id + LIMIT $4 +` + +type ListDuplicateCandidatesParams struct { + CurrentVersion int16 + AfterDurationMs int32 + AfterID pgtype.UUID + PageLimit int32 +} + +type ListDuplicateCandidatesRow struct { + ID pgtype.UUID + DurationMs int32 + AudioStreamSha256 []byte + Chromaprint []int32 +} + +// The acoustic tier's input, one page at a time in (duration_ms, id) order so the +// sweep holds only a sliding window of durations. Tracks without a chromaprint +// cannot be compared acoustically and are left out; any exact duplicates among +// them come from ListExactDuplicateHashes. +func (q *Queries) ListDuplicateCandidates(ctx context.Context, arg ListDuplicateCandidatesParams) ([]ListDuplicateCandidatesRow, error) { + rows, err := q.db.Query(ctx, listDuplicateCandidates, + arg.CurrentVersion, + arg.AfterDurationMs, + arg.AfterID, + arg.PageLimit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListDuplicateCandidatesRow + for rows.Next() { + var i ListDuplicateCandidatesRow + if err := rows.Scan( + &i.ID, + &i.DurationMs, + &i.AudioStreamSha256, + &i.Chromaprint, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listExactDuplicateHashes = `-- name: ListExactDuplicateHashes :many +SELECT f.audio_stream_sha256, + array_agg(t.id ORDER BY t.id)::uuid[] AS track_ids + FROM track_fingerprints f + JOIN tracks t ON t.id = f.track_id + WHERE t.missing_since IS NULL + AND f.fingerprint_version >= $1 + AND f.audio_stream_sha256 IS NOT NULL + GROUP BY f.audio_stream_sha256 +HAVING count(*) > 1 +` + +type ListExactDuplicateHashesRow struct { + AudioStreamSha256 []byte + TrackIds []pgtype.UUID +} + +// The exact tier, library-wide in one pass: identical encoded audio shared by +// more than one present track. +func (q *Queries) ListExactDuplicateHashes(ctx context.Context, currentVersion int16) ([]ListExactDuplicateHashesRow, error) { + rows, err := q.db.Query(ctx, listExactDuplicateHashes, currentVersion) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListExactDuplicateHashesRow + for rows.Next() { + var i ListExactDuplicateHashesRow + if err := rows.Scan(&i.AudioStreamSha256, &i.TrackIds); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const startDuplicateSweep = `-- name: StartDuplicateSweep :one +INSERT INTO duplicate_sweeps DEFAULT VALUES RETURNING id, started_at +` + +type StartDuplicateSweepRow struct { + ID pgtype.UUID + StartedAt pgtype.Timestamptz +} + +func (q *Queries) StartDuplicateSweep(ctx context.Context) (StartDuplicateSweepRow, error) { + row := q.db.QueryRow(ctx, startDuplicateSweep) + var i StartDuplicateSweepRow + err := row.Scan(&i.ID, &i.StartedAt) + return i, err +} + +const upsertDuplicateGroup = `-- name: UpsertDuplicateGroup :one +INSERT INTO duplicate_groups (member_key, tier, worst_bit_error_rate, last_seen_sweep_id) +VALUES ($1, $2, $3, $4) +ON CONFLICT (member_key) DO UPDATE + SET tier = EXCLUDED.tier, + worst_bit_error_rate = EXCLUDED.worst_bit_error_rate, + last_seen_sweep_id = EXCLUDED.last_seen_sweep_id + WHERE duplicate_groups.status = 'pending' +RETURNING id +` + +type UpsertDuplicateGroupParams struct { + MemberKey string + Tier string + WorstBitErrorRate *float32 + SweepID pgtype.UUID +} + +// Proposes a group, or refreshes one already pending. A group already dismissed +// or merged is left exactly as it is: the WHERE on the update makes the conflict +// a no-op, and the caller sees no row. +func (q *Queries) UpsertDuplicateGroup(ctx context.Context, arg UpsertDuplicateGroupParams) (pgtype.UUID, error) { + row := q.db.QueryRow(ctx, upsertDuplicateGroup, + arg.MemberKey, + arg.Tier, + arg.WorstBitErrorRate, + arg.SweepID, + ) + var id pgtype.UUID + err := row.Scan(&id) + return id, err +} diff --git a/internal/db/dbq/models.go b/internal/db/dbq/models.go index 07ca6f75..cddadd0c 100644 --- a/internal/db/dbq/models.go +++ b/internal/db/dbq/models.go @@ -297,6 +297,32 @@ type DiscoverTuning struct { UpdatedAt pgtype.Timestamptz } +type DuplicateGroup struct { + ID pgtype.UUID + MemberKey string + Tier string + WorstBitErrorRate *float32 + Status string + DetectedAt pgtype.Timestamptz + LastSeenSweepID pgtype.UUID + ResolvedAt pgtype.Timestamptz +} + +type DuplicateGroupMember struct { + GroupID pgtype.UUID + TrackID pgtype.UUID +} + +type DuplicateSweep struct { + ID pgtype.UUID + StartedAt pgtype.Timestamptz + FinishedAt pgtype.Timestamptz + Candidates *int32 + GroupsFound *int32 + OversizeClusters *int32 + ErrorMessage *string +} + type GeneralLike struct { UserID pgtype.UUID TrackID pgtype.UUID diff --git a/internal/db/migrations/0059_duplicate_groups.down.sql b/internal/db/migrations/0059_duplicate_groups.down.sql new file mode 100644 index 00000000..1d37bf4c --- /dev/null +++ b/internal/db/migrations/0059_duplicate_groups.down.sql @@ -0,0 +1,4 @@ +DROP INDEX IF EXISTS tracks_duration_id_idx; +DROP TABLE duplicate_group_members; +DROP TABLE duplicate_groups; +DROP TABLE duplicate_sweeps; diff --git a/internal/db/migrations/0059_duplicate_groups.up.sql b/internal/db/migrations/0059_duplicate_groups.up.sql new file mode 100644 index 00000000..17a112b5 --- /dev/null +++ b/internal/db/migrations/0059_duplicate_groups.up.sql @@ -0,0 +1,54 @@ +-- 0059_duplicate_groups.up.sql — proposed duplicates and the sweeps that find +-- them (Scribe milestone #400: #3910). +-- +-- The sweep compares fingerprints (track_fingerprints, 0058) and proposes groups +-- of tracks that hold one recording. Nothing here merges anything: a group is a +-- proposal the operator reviews, and the merge (#3911) is a separate act. + +-- One row per sweep. Lets the report tell "the sweep has never run" apart from +-- "it ran and found nothing", and gives the in-flight guard something to check, +-- the same way scan_runs does for the library scan. +CREATE TABLE duplicate_sweeps ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + started_at timestamptz NOT NULL DEFAULT now(), + finished_at timestamptz, + candidates integer, + groups_found integer, + oversize_clusters integer, + error_message text +); +CREATE INDEX duplicate_sweeps_started_at_idx ON duplicate_sweeps (started_at DESC); + +CREATE TABLE duplicate_groups ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + -- The group's identity: its member track ids, sorted and joined. A sweep + -- that finds the same tracks again updates this row rather than proposing + -- them twice, and a dismissal stays attached to the set it was made about. + member_key text NOT NULL UNIQUE, + -- Rule 36: a new value for either CHECK swaps the constraint in the same + -- migration. + tier text NOT NULL CHECK (tier IN ('exact', 'acoustic')), + -- Largest disagreement between any two members; NULL for exact groups, + -- which have no score. + worst_bit_error_rate real, + status text NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'dismissed', 'merged')), + detected_at timestamptz NOT NULL DEFAULT now(), + last_seen_sweep_id uuid REFERENCES duplicate_sweeps (id) ON DELETE SET NULL, + resolved_at timestamptz +); +CREATE INDEX duplicate_groups_status_idx ON duplicate_groups (status); + +CREATE TABLE duplicate_group_members ( + group_id uuid NOT NULL REFERENCES duplicate_groups (id) ON DELETE CASCADE, + -- CASCADE is right here: a track that genuinely leaves the library has no + -- place in a proposal about its duplicates. + track_id uuid NOT NULL REFERENCES tracks (id) ON DELETE CASCADE, + PRIMARY KEY (group_id, track_id) +); +CREATE INDEX duplicate_group_members_track_idx ON duplicate_group_members (track_id); + +-- The sweep streams candidates in (duration_ms, id) order, keyset-paged, so it +-- only ever holds a few seconds' worth of durations in memory. Without this each +-- page would sort the whole library again. +CREATE INDEX tracks_duration_id_idx ON tracks (duration_ms, id); diff --git a/internal/db/queries/duplicates.sql b/internal/db/queries/duplicates.sql new file mode 100644 index 00000000..def52620 --- /dev/null +++ b/internal/db/queries/duplicates.sql @@ -0,0 +1,100 @@ +-- name: StartDuplicateSweep :one +INSERT INTO duplicate_sweeps DEFAULT VALUES RETURNING id, started_at; + +-- name: FinishDuplicateSweep :exec +UPDATE duplicate_sweeps + SET finished_at = now(), + candidates = sqlc.arg(candidates), + groups_found = sqlc.arg(groups_found), + oversize_clusters = sqlc.arg(oversize_clusters), + error_message = NULLIF(sqlc.arg(error_message)::text, '') + WHERE id = sqlc.arg(id); + +-- name: GetInFlightDuplicateSweep :one +-- The guard against two sweeps at once: "in flight" is finished_at IS NULL. +SELECT id, started_at + FROM duplicate_sweeps + WHERE finished_at IS NULL + ORDER BY started_at DESC + LIMIT 1; + +-- name: GetLatestDuplicateSweep :one +SELECT id, started_at, finished_at, candidates, groups_found, oversize_clusters, error_message + FROM duplicate_sweeps + ORDER BY started_at DESC + LIMIT 1; + +-- name: GetLatestFingerprintComputedAt :one +-- Whether a sweep has anything new to look at: fingerprints written since the +-- last sweep started. +SELECT max(computed_at)::timestamptz AS latest FROM track_fingerprints; + +-- name: ListExactDuplicateHashes :many +-- The exact tier, library-wide in one pass: identical encoded audio shared by +-- more than one present track. +SELECT f.audio_stream_sha256, + array_agg(t.id ORDER BY t.id)::uuid[] AS track_ids + FROM track_fingerprints f + JOIN tracks t ON t.id = f.track_id + WHERE t.missing_since IS NULL + AND f.fingerprint_version >= sqlc.arg(current_version) + AND f.audio_stream_sha256 IS NOT NULL + GROUP BY f.audio_stream_sha256 +HAVING count(*) > 1; + +-- name: ListDuplicateCandidates :many +-- The acoustic tier's input, one page at a time in (duration_ms, id) order so the +-- sweep holds only a sliding window of durations. Tracks without a chromaprint +-- cannot be compared acoustically and are left out; any exact duplicates among +-- them come from ListExactDuplicateHashes. +SELECT t.id, t.duration_ms, f.audio_stream_sha256, f.chromaprint + FROM tracks t + JOIN track_fingerprints f ON f.track_id = t.id + WHERE t.missing_since IS NULL + AND f.fingerprint_version >= sqlc.arg(current_version) + AND f.chromaprint IS NOT NULL + AND (t.duration_ms, t.id) > (sqlc.arg(after_duration_ms)::integer, sqlc.arg(after_id)::uuid) + ORDER BY t.duration_ms, t.id + LIMIT sqlc.arg(page_limit); + +-- name: ListDismissedDuplicateMemberSets :many +-- What the operator has already said are not duplicates. A new proposal whose +-- every member sat together in one of these is not proposed again. +SELECT g.id, array_agg(m.track_id ORDER BY m.track_id)::uuid[] AS track_ids + FROM duplicate_groups g + JOIN duplicate_group_members m ON m.group_id = g.id + WHERE g.status = 'dismissed' + GROUP BY g.id; + +-- name: UpsertDuplicateGroup :one +-- Proposes a group, or refreshes one already pending. A group already dismissed +-- or merged is left exactly as it is: the WHERE on the update makes the conflict +-- a no-op, and the caller sees no row. +INSERT INTO duplicate_groups (member_key, tier, worst_bit_error_rate, last_seen_sweep_id) +VALUES (sqlc.arg(member_key), sqlc.arg(tier), sqlc.narg(worst_bit_error_rate), sqlc.arg(sweep_id)) +ON CONFLICT (member_key) DO UPDATE + SET tier = EXCLUDED.tier, + worst_bit_error_rate = EXCLUDED.worst_bit_error_rate, + last_seen_sweep_id = EXCLUDED.last_seen_sweep_id + WHERE duplicate_groups.status = 'pending' +RETURNING id; + +-- name: AddDuplicateGroupMember :exec +INSERT INTO duplicate_group_members (group_id, track_id) +VALUES (sqlc.arg(group_id), sqlc.arg(track_id)) +ON CONFLICT DO NOTHING; + +-- name: DeleteStalePendingDuplicateGroups :execrows +-- A pending proposal this sweep did not find again no longer describes the +-- library: a member was re-fingerprinted, merged away or went missing. Dismissed +-- groups are kept regardless — they are the memory of a decision. +-- +-- Only proposals last confirmed by an EARLIER sweep go. Should two sweeps ever +-- overlap (a manual trigger racing the worker), neither may delete what the other +-- has just found. +DELETE FROM duplicate_groups g + WHERE g.status = 'pending' + AND g.last_seen_sweep_id IS DISTINCT FROM sqlc.arg(sweep_id) + AND (g.last_seen_sweep_id IS NULL + OR (SELECT s.started_at FROM duplicate_sweeps s WHERE s.id = g.last_seen_sweep_id) + < (SELECT s.started_at FROM duplicate_sweeps s WHERE s.id = sqlc.arg(sweep_id))); diff --git a/internal/dbtest/reset.go b/internal/dbtest/reset.go index a99db739..25e0b076 100644 --- a/internal/dbtest/reset.go +++ b/internal/dbtest/reset.go @@ -87,6 +87,9 @@ var dataTables = []string{ // pristine Discover knobs rather than whatever a previous test tuned. "discover_tuning", "recommendation_tuning_audit", + "duplicate_group_members", // M400 + "duplicate_groups", + "duplicate_sweeps", "track_fingerprints", // M400 "tracks", "albums", diff --git a/internal/library/duplicate_match.go b/internal/library/duplicate_match.go index 75539e88..d45b83f4 100644 --- a/internal/library/duplicate_match.go +++ b/internal/library/duplicate_match.go @@ -82,15 +82,52 @@ type acousticScore struct { BitErrorRate float64 } +// preparedPrint is a fingerprint with the parts every comparison needs worked +// out once. The sweep compares each track with every other track within a few +// seconds of its duration, so rebuilding the alignment index for each pair would +// dominate its cost. +type preparedPrint struct { + items []int32 + index map[uint32][]int + informative bool +} + +// preparePrint indexes a fingerprint's items by their high bits and records +// whether it varies enough to be compared at all. +func preparePrint(fp []int32) *preparedPrint { + // Each bucket keeps only a few positions: a value repeating many times is + // uninformative, and letting it vote once per repeat would make every + // pairing O(n²). + const keepPerBucket = 4 + p := &preparedPrint{items: fp, index: make(map[uint32][]int, len(fp))} + seen := make(map[int32]struct{}, len(fp)) + for i, v := range fp { + seen[v] = struct{}{} + key := alignKey(v) + if pos := p.index[key]; len(pos) < keepPerBucket { + p.index[key] = append(pos, i) + } + } + p.informative = len(fp) > 0 && float64(len(seen)) >= minDistinctFraction*float64(len(fp)) + return p +} + +func alignKey(v int32) uint32 { return uint32(v) >> (32 - alignMatchBits) } + // compareChromaprint aligns two raw fingerprints and measures how much they // disagree. ok is false when no verdict is possible: no offset gathered any // votes, the overlap at the best offset is too short, or either side carries // too little information to mean anything. func compareChromaprint(a, b []int32) (acousticScore, bool) { - if len(a) < minOverlapItems || len(b) < minOverlapItems { + return comparePrepared(preparePrint(a), preparePrint(b)) +} + +// comparePrepared is compareChromaprint over fingerprints already prepared. +func comparePrepared(a, b *preparedPrint) (acousticScore, bool) { + if len(a.items) < minOverlapItems || len(b.items) < minOverlapItems { return acousticScore{}, false } - if !informative(a) || !informative(b) { + if !a.informative || !b.informative { return acousticScore{}, false } @@ -101,14 +138,14 @@ func compareChromaprint(a, b []int32) (acousticScore, bool) { // a[i] aligns with b[i+offset]; walk the indices valid on both sides. start := max(0, -offset) - end := min(len(a), len(b)-offset) + end := min(len(a.items), len(b.items)-offset) overlap := end - start if overlap < minOverlapItems { return acousticScore{}, false } errBits := 0 for i := start; i < end; i++ { - errBits += bits.OnesCount32(uint32(a[i]) ^ uint32(b[i+offset])) + errBits += bits.OnesCount32(uint32(a.items[i]) ^ uint32(b.items[i+offset])) } return acousticScore{ Offset: offset, @@ -118,22 +155,10 @@ func compareChromaprint(a, b []int32) (acousticScore, bool) { } // bestOffset returns the relative shift most items agree on. -func bestOffset(a, b []int32) (int, bool) { - // Index a's items by their high bits. Each bucket keeps only a few - // positions: a value repeating many times is uninformative, and letting it - // vote once per repeat would make every pairing O(n²). - const keepPerBucket = 4 - positions := make(map[uint32][]int, len(a)) - for i, v := range a { - key := uint32(v) >> (32 - alignMatchBits) - if p := positions[key]; len(p) < keepPerBucket { - positions[key] = append(p, i) - } - } - +func bestOffset(a, b *preparedPrint) (int, bool) { votes := make([]int, 2*maxAlignOffsetItems+1) - for j, v := range b { - for _, i := range positions[uint32(v)>>(32-alignMatchBits)] { + for j, v := range b.items { + for _, i := range a.index[alignKey(v)] { off := j - i if off >= -maxAlignOffsetItems && off <= maxAlignOffsetItems { votes[off+maxAlignOffsetItems]++ @@ -152,15 +177,6 @@ func bestOffset(a, b []int32) (int, bool) { return best, bestVotes > 0 } -// informative reports whether a fingerprint varies enough to be compared. -func informative(fp []int32) bool { - seen := make(map[int32]struct{}, len(fp)) - for _, v := range fp { - seen[v] = struct{}{} - } - return float64(len(seen)) >= minDistinctFraction*float64(len(fp)) -} - // fingerprintCandidate is one track as the grouping sees it. type fingerprintCandidate struct { ID string @@ -197,149 +213,206 @@ type groupingResult struct { OversizeClusters int } -// groupDuplicates proposes duplicate groups among candidates. +// groupUnit is one thing the acoustic pass compares: a single track, or an exact +// group standing in for all its byte-identical copies. +type groupUnit struct { + ids []string // every member, sorted + durationMs int32 + sortKey string // the representative's id: ties on duration break on it + print *preparedPrint + exact bool // more than one member with identical audio + assigned bool +} + +// streamGrouper is the acoustic pass over units arriving in (durationMs, sortKey) +// order. It holds only the units within durationToleranceMs of the oldest one +// not yet settled, so memory is bounded by the densest few seconds of the +// library rather than by its size — the whole library's fingerprints would be +// hundreds of megabytes. // -// Exact groups come first: tracks sharing an audio stream hash. Each exact group -// is then treated as a single unit for the acoustic pass, so its members are -// never compared with each other again. +// A seed can be settled as soon as a unit arrives beyond its window: everything +// it could group with has already arrived, and no later seed can reach back to +// it because seeds only look forward. That is what makes the streamed result +// identical to running the same pass over the whole sorted list. // -// Acoustic grouping is COMPLETE-LINKAGE: a unit joins a group only if it matches -// every unit already in it, within the duration tolerance and the bit-error -// limit. Single-linkage would let a chain of near-misses — A close to B, B close -// to C — drag A and C, which are not close, into one proposed merge. Complete -// linkage also means any member can be chosen as the survivor (#3911). +// Grouping is COMPLETE-LINKAGE: a unit joins a group only if it matches every +// unit already in it, within the duration tolerance and the bit-error limit. +// Single-linkage would let a chain of near-misses — A close to B, B close to C — +// drag A and C, which are not close, into one proposed merge. Complete linkage +// also means any member can be chosen as the survivor (#3911). // -// When an acoustic group absorbs an exact group, the result is tier acoustic: -// a group is only as certain as its weakest link. +// When an acoustic group absorbs an exact group, the result is tier acoustic: a +// group is only as certain as its weakest link. +type streamGrouper struct { + maxBitErrorRate float64 + window []*groupUnit + res groupingResult +} + +func newStreamGrouper(maxBitErrorRate float64) *streamGrouper { + return &streamGrouper{maxBitErrorRate: maxBitErrorRate} +} + +// push adds the next unit. Units must arrive in non-decreasing +// (durationMs, sortKey) order. +func (g *streamGrouper) push(u *groupUnit) { + g.window = append(g.window, u) + for len(g.window) > 1 && u.durationMs-g.window[0].durationMs > durationToleranceMs { + g.settleOldest() + } +} + +// finish settles every unit still waiting and returns what was found. Groups +// are in no particular order; callers sort with sortGroups. +func (g *streamGrouper) finish() groupingResult { + for len(g.window) > 0 { + g.settleOldest() + } + return g.res +} + +func (g *streamGrouper) settleOldest() { + seed := g.window[0] + g.window[0] = nil // release it: the window's backing array outlives the slide + g.window = g.window[1:] + if seed.assigned { + return + } + + group := []*groupUnit{seed} + worst := 0.0 + for _, cand := range g.window { + if cand.durationMs-seed.durationMs > durationToleranceMs { + break + } + if cand.assigned { + continue + } + joined, worstWithCand := true, worst + for _, member := range group { + if abs32(cand.durationMs-member.durationMs) > durationToleranceMs { + joined = false + break + } + score, ok := comparePrepared(member.print, cand.print) + if !ok || score.BitErrorRate > g.maxBitErrorRate { + joined = false + break + } + worstWithCand = math.Max(worstWithCand, score.BitErrorRate) + } + if joined { + group = append(group, cand) + worst = worstWithCand + } + } + + if len(group) == 1 { + if seed.exact { + g.res.Groups = append(g.res.Groups, duplicateGroup{Tier: tierExact, Members: seed.ids}) + } + return + } + // Count units, not tracks: an absorbed exact group is one piece of acoustic + // evidence however many identical files it holds. + if len(group) > maxAcousticGroupSize { + g.res.OversizeClusters++ + for _, member := range group { + member.assigned = true + // The acoustic evidence is untrustworthy; identical bytes are not. + // An exact group caught inside an oversize cluster is still proposed. + if member.exact { + g.res.Groups = append(g.res.Groups, duplicateGroup{Tier: tierExact, Members: member.ids}) + } + } + return + } + var members []string + for _, member := range group { + member.assigned = true + members = append(members, member.ids...) + } + sort.Strings(members) + g.res.Groups = append(g.res.Groups, duplicateGroup{ + Tier: tierAcoustic, Members: members, WorstBitErrorRate: worst, + }) +} + +// groupDuplicates proposes duplicate groups among candidates held in memory. It +// runs the same streamGrouper the sweep uses, so there is one grouping rule. +// +// Exact groups come first: tracks sharing an audio stream hash. Each becomes a +// single unit for the acoustic pass, represented by its member with the lowest +// (duration, id) that has a chromaprint. That is the member the sweep's +// duration-ordered stream meets first, which keeps the two identical. An exact +// group with no chromaprint at all cannot be compared acoustically and stands +// on its own. // // The output does not depend on input order. func groupDuplicates(cands []fingerprintCandidate, maxBitErrorRate float64) groupingResult { - var res groupingResult - - // Exact tier. byHash := map[string][]fingerprintCandidate{} - var noHash []fingerprintCandidate + var units []*groupUnit + var printless []duplicateGroup for _, c := range cands { - if len(c.StreamSHA256) == 0 { - noHash = append(noHash, c) + if len(c.StreamSHA256) > 0 { + byHash[string(c.StreamSHA256)] = append(byHash[string(c.StreamSHA256)], c) continue } - k := string(c.StreamSHA256) - byHash[k] = append(byHash[k], c) + if len(c.Chromaprint) > 0 { + units = append(units, &groupUnit{ + ids: []string{c.ID}, durationMs: c.DurationMs, sortKey: c.ID, print: preparePrint(c.Chromaprint), + }) + } } - - // A unit is one exact group, or one track with no exact duplicate. - type unit struct { - members []fingerprintCandidate - durationMs int32 - print []int32 - exact bool - } - var units []unit for _, group := range byHash { - sortCandidates(group) - u := unit{members: group, durationMs: group[0].DurationMs, exact: len(group) > 1} - for _, m := range group { - if len(m.Chromaprint) > 0 { - u.print = m.Chromaprint - break + ids := make([]string, len(group)) + for i, m := range group { + ids[i] = m.ID + } + sort.Strings(ids) + + var rep *fingerprintCandidate + for i := range group { + m := &group[i] + if len(m.Chromaprint) == 0 { + continue + } + if rep == nil || m.DurationMs < rep.DurationMs || (m.DurationMs == rep.DurationMs && m.ID < rep.ID) { + rep = m } } - units = append(units, u) - } - for _, c := range noHash { - units = append(units, unit{members: []fingerprintCandidate{c}, durationMs: c.DurationMs, print: c.Chromaprint}) + if rep == nil { + if len(group) > 1 { + printless = append(printless, duplicateGroup{Tier: tierExact, Members: ids}) + } + continue + } + units = append(units, &groupUnit{ + ids: ids, durationMs: rep.DurationMs, sortKey: rep.ID, + print: preparePrint(rep.Chromaprint), exact: len(group) > 1, + }) } - // Deterministic order: duration, then the first member's ID. Sorting by - // duration also lets the scan below stop as soon as durations are too far - // apart, which is the blocking #3910 relies on. sort.Slice(units, func(i, j int) bool { if units[i].durationMs != units[j].durationMs { return units[i].durationMs < units[j].durationMs } - return units[i].members[0].ID < units[j].members[0].ID + return units[i].sortKey < units[j].sortKey }) - - assigned := make([]bool, len(units)) - for i := range units { - if assigned[i] || len(units[i].print) == 0 { - continue - } - group := []int{i} - worst := 0.0 - for j := i + 1; j < len(units); j++ { - if units[j].durationMs-units[i].durationMs > durationToleranceMs { - break - } - if assigned[j] || len(units[j].print) == 0 { - continue - } - // Complete linkage: j must match every member so far. - joined, worstWithJ := true, worst - for _, g := range group { - if abs32(units[j].durationMs-units[g].durationMs) > durationToleranceMs { - joined = false - break - } - score, ok := compareChromaprint(units[g].print, units[j].print) - if !ok || score.BitErrorRate > maxBitErrorRate { - joined = false - break - } - worstWithJ = math.Max(worstWithJ, score.BitErrorRate) - } - if joined { - group = append(group, j) - worst = worstWithJ - } - } - - if len(group) == 1 { - continue - } - // Count units, not tracks: an absorbed exact group is one piece of - // acoustic evidence however many identical files it holds. - if len(group) > maxAcousticGroupSize { - res.OversizeClusters++ - for _, g := range group { - assigned[g] = true - } - continue - } - var members []string - for _, g := range group { - assigned[g] = true - for _, m := range units[g].members { - members = append(members, m.ID) - } - } - sort.Strings(members) - res.Groups = append(res.Groups, duplicateGroup{ - Tier: tierAcoustic, Members: members, WorstBitErrorRate: worst, - }) + g := newStreamGrouper(maxBitErrorRate) + for _, u := range units { + g.push(u) } - - // Exact groups that no acoustic group absorbed stand on their own. - for i, u := range units { - if assigned[i] || !u.exact { - continue - } - members := make([]string, len(u.members)) - for k, m := range u.members { - members[k] = m.ID - } - res.Groups = append(res.Groups, duplicateGroup{Tier: tierExact, Members: members}) - } - - sort.Slice(res.Groups, func(i, j int) bool { - return res.Groups[i].Members[0] < res.Groups[j].Members[0] - }) + res := g.finish() + res.Groups = append(res.Groups, printless...) + sortGroups(res.Groups) return res } -func sortCandidates(cs []fingerprintCandidate) { - sort.Slice(cs, func(i, j int) bool { return cs[i].ID < cs[j].ID }) +// sortGroups orders groups by their first member. Groups are disjoint, so that +// is a total order. +func sortGroups(groups []duplicateGroup) { + sort.Slice(groups, func(i, j int) bool { return groups[i].Members[0] < groups[j].Members[0] }) } func abs(n int) int { diff --git a/internal/library/duplicate_sweep.go b/internal/library/duplicate_sweep.go new file mode 100644 index 00000000..f1c3cf0f --- /dev/null +++ b/internal/library/duplicate_sweep.go @@ -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 +} diff --git a/internal/library/duplicate_sweep_test.go b/internal/library/duplicate_sweep_test.go new file mode 100644 index 00000000..9f30c778 --- /dev/null +++ b/internal/library/duplicate_sweep_test.go @@ -0,0 +1,173 @@ +package library + +import ( + "bytes" + "context" + "io" + "log/slog" + "path/filepath" + "sort" + "strings" + "testing" + + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" + syncpkg "git.fabledsword.com/bvandeusen/minstrel/internal/sync" +) + +func TestCoveredByDismissal(t *testing.T) { + dismissed := []map[string]struct{}{{"a": {}, "b": {}, "c": {}}} + for _, tc := range []struct { + name string + members []string + want bool + }{ + {"the same set", []string{"a", "b", "c"}, true}, + {"a subset of it", []string{"a", "b"}, true}, + // A new copy joining is new evidence: ask again. + {"a superset of it", []string{"a", "b", "c", "d"}, false}, + {"overlapping only in part", []string{"a", "d"}, false}, + {"unrelated", []string{"x", "y"}, false}, + } { + if got := coveredByDismissal(tc.members, dismissed); got != tc.want { + t.Errorf("%s: coveredByDismissal = %v, want %v", tc.name, got, tc.want) + } + } +} + +// TestDuplicateSweep_Integration pins what the sweep proposes, what it leaves out, +// and how re-sweeping treats a dismissal and a proposal that no longer holds. +func TestDuplicateSweep_Integration(t *testing.T) { + pool := newPool(t) + ctx := context.Background() + q := dbq.New(pool) + dir := t.TempDir() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + + // seedTrack's own track has no fingerprint row: it must be absent from the + // report, not grouped with every other track lacking one. + _, album, artist := seedTrack(t, pool, filepath.Join(dir, "unfingerprinted.mp3")) + hash := func(b byte) []byte { return bytes.Repeat([]byte{b}, 32) } + add := func(name string, durationMs int32, sum []byte, print []int32) string { + t.Helper() + tr, err := q.UpsertTrack(ctx, dbq.UpsertTrackParams{ + Title: name, AlbumID: album.ID, ArtistID: artist.ID, DurationMs: durationMs, + FilePath: filepath.Join(dir, name+".mp3"), FileSize: 100, FileFormat: "mp3", + }) + if err != nil { + t.Fatalf("track %s: %v", name, err) + } + if err := q.UpsertTrackFingerprint(ctx, dbq.UpsertTrackFingerprintParams{ + TrackID: tr.ID, AudioStreamSha256: sum, Chromaprint: print, FingerprintVersion: fingerprintVersion, + }); err != nil { + t.Fatalf("fingerprint %s: %v", name, err) + } + return syncpkg.FormatUUID(tr.ID) + } + key := func(ids ...string) string { + sorted := append([]string(nil), ids...) + sort.Strings(sorted) + return strings.Join(sorted, ",") + } + + recording := randomPrint(200, printLen) + onAlbum := add("recording-album", 240000, hash(1), recording) + onCompilation := add("recording-compilation", 241000, hash(2), withBitNoise(recording, 0.05, 201)) + www1 := add("www-01", 215000, hash(9), randomPrint(210, printLen)) + www2 := add("www-02", 215000, hash(9), randomPrint(210, printLen)) + // Near-identical duration to the recording, different audio. + add("different-song", 240500, hash(3), randomPrint(220, printLen)) + // Identical to the album copy, but its file is gone: nothing to compare. + missing := add("missing-copy", 240000, hash(4), recording) + if _, err := pool.Exec(ctx, "UPDATE tracks SET missing_since = now() WHERE file_path LIKE '%missing-copy.mp3'"); err != nil { + t.Fatalf("mark missing: %v", err) + } + + type stored struct { + tier, status string + } + groups := func() map[string]stored { + t.Helper() + rows, err := pool.Query(ctx, `SELECT member_key, tier, status FROM duplicate_groups`) + if err != nil { + t.Fatalf("read groups: %v", err) + } + defer rows.Close() + out := map[string]stored{} + for rows.Next() { + var k string + var s stored + if err := rows.Scan(&k, &s.tier, &s.status); err != nil { + t.Fatalf("scan group: %v", err) + } + out[k] = s + } + return out + } + + // 1. A page size of one forces the keyset cursor across every candidate. + res, err := runDuplicateSweep(ctx, pool, logger, 1) + if err != nil { + t.Fatalf("first sweep: %v", err) + } + // Five tracks carry a chromaprint and a present file. + if res.Candidates != 5 || res.Groups != 2 || res.Proposed != 2 { + t.Fatalf("first sweep = %+v, want 5 candidates, 2 groups, 2 proposed", res) + } + acousticKey, exactKey := key(onAlbum, onCompilation), key(www1, www2) + got := groups() + want := map[string]stored{ + acousticKey: {"acoustic", "pending"}, + exactKey: {"exact", "pending"}, + } + if len(got) != len(want) || got[acousticKey] != want[acousticKey] || got[exactKey] != want[exactKey] { + t.Fatalf("groups = %+v, want %+v", got, want) + } + for k := range got { + if strings.Contains(k, missing) { + t.Fatalf("a missing track was proposed: %s", k) + } + } + + // 2. A dismissed group is not proposed again, and the pending one is + // refreshed in place rather than duplicated. + if _, err := pool.Exec(ctx, "UPDATE duplicate_groups SET status = 'dismissed' WHERE member_key = $1", acousticKey); err != nil { + t.Fatalf("dismiss: %v", err) + } + res, err = runDuplicateSweep(ctx, pool, logger, duplicateCandidatePage) + if err != nil { + t.Fatalf("second sweep: %v", err) + } + if res.Proposed != 1 || res.Suppressed != 1 { + t.Fatalf("second sweep = %+v, want 1 proposed, 1 suppressed", res) + } + got = groups() + if len(got) != 2 || got[acousticKey].status != "dismissed" || got[exactKey].status != "pending" { + t.Fatalf("after dismissal groups = %+v, want the dismissal kept and one pending group", got) + } + + // 3. A proposal that no longer holds is retired; the dismissal survives it. + if _, err := pool.Exec(ctx, + "DELETE FROM track_fingerprints f USING tracks t WHERE f.track_id = t.id AND t.file_path LIKE '%www-02.mp3'"); err != nil { + t.Fatalf("drop fingerprint: %v", err) + } + res, err = runDuplicateSweep(ctx, pool, logger, duplicateCandidatePage) + if err != nil { + t.Fatalf("third sweep: %v", err) + } + if res.Retired != 1 { + t.Fatalf("third sweep = %+v, want 1 retired", res) + } + got = groups() + if len(got) != 1 || got[acousticKey].status != "dismissed" { + t.Fatalf("after retiring groups = %+v, want only the dismissal", got) + } + + // 4. The sweep record reflects the last run. + last, err := q.GetLatestDuplicateSweep(ctx) + if err != nil { + t.Fatalf("latest sweep: %v", err) + } + if !last.FinishedAt.Valid || last.ErrorMessage != nil { + t.Fatalf("latest sweep = %+v, want finished without error", last) + } +}