fix(similarity): keep ListenBrainz's whole answer and resolve it locally (#5296)

The worker kept only the similar recordings already in the library, at most 20 of ListenBrainz's 50, and judged freshness by the edges it had written. Two failures followed, both measured on the operator's library (#3879): - A seed whose answer matched nothing wrote nothing, so it was never fresh. With the queue ordered by id, 25 such seeds held its head and were re-asked every hour; 17 of 2,466 played seeds had any edges. - A recording that reached the library after its seed was fetched (a Lidarr import, an MBID from the AcoustID lookup) was never linked until a refetch, which for the stuck seeds never came. Now every answer is cached whole in listenbrainz_similar_recordings and every answer, an empty one or a permanent 4xx included, is recorded in track_similarity_fetches. The queue reads the fetch record: never-fetched first, then the oldest, refreshed after 30 days. The listenbrainz edges are derived in SQL from the cache, one present track per recording and no cap, for the seed just fetched and for every seed once per tick, so new arrivals link within the hour without asking ListenBrainz again. Artists get the same queue fix via artist_similarity_fetches; their answer was already kept in artist_similarity and artist_similarity_unmatched. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-07 19:56:22 -04:00
parent aa390c711a
commit e5dac9ddf0
9 changed files with 548 additions and 177 deletions
@@ -1,18 +1,29 @@
 // Package similarity owns the inbound ListenBrainz similarity ingest
 // pipeline. A periodic worker queries the LB Labs API
 // (labs.api.listenbrainz.org similar-recordings / similar-artists) for
-// tracks the user has played, filters returned MBIDs to the local
-// library, and stores the top-K edges in track_similarity /
-// artist_similarity for M4c's radio candidate-pool builder.
+// tracks the user has played.
+//
+// For recordings it keeps LB's whole answer in listenbrainz_similar_recordings
+// and derives the track_similarity edges from it in SQL, against whatever is
+// in the library at the time (#5296). Matching once at fetch time missed every
+// recording that reached the library afterwards, through Lidarr or an MBID the
+// AcoustID lookup filled in. For artists it stores the in-library matches in
+// artist_similarity and the rest in artist_similarity_unmatched, as before.
+//
+// Every answer is recorded as a fetch, an empty one included, and the queue
+// reads those records: freshness used to be read from the edges, so a seed
+// that matched nothing was never fresh and held the head of the queue.
 package similarity

 import (
 	"context"
 	"errors"
+	"fmt"
 	"log/slog"
 	"sort"
 	"time"

+	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/jackc/pgx/v5/pgxpool"

@@ -20,23 +31,21 @@ import (
 	"git.fabledsword.com/bvandeusen/minstrel/internal/scrobble/listenbrainz"
 )

-// Worker drains played-tracks-and-artists needing similarity and POSTs
-// the results into track_similarity / artist_similarity. Failures are
-// passively retried via the timer (no durable queue table — losing one
-// tick's worth of refresh attempts is "1 hour of staleness," fine).
+// Worker drains played tracks and artists needing similarity. A transient
+// failure is not recorded as a fetch, so the timer retries it next tick.
 type Worker struct {
 	pool   *pgxpool.Pool
 	client *listenbrainz.Client
 	logger *slog.Logger
 	tick   time.Duration
 	batch  int32
-	topK   int
+	topK   int // per artist, for each of the matched and unmatched stores
 }

 // NewWorker constructs a worker with production defaults: 1h tick,
-// batch=25, topK=20. batch is tracks AND artists processed per tick;
-// at 25/h a freshly-played library converges in hours, not days, while
-// staying well under ListenBrainz rate limits (429s abort the tick).
+// batch=25, topK=20. batch is tracks AND artists processed per tick, well
+// under ListenBrainz rate limits (429s abort the tick). With never-fetched
+// seeds first, a freshly played library is covered in days.
 func NewWorker(pool *pgxpool.Pool, client *listenbrainz.Client, logger *slog.Logger) *Worker {
 	return &Worker{
 		pool:   pool,
@@ -64,15 +73,23 @@ func (w *Worker) Run(ctx context.Context) {
 	}
 }

-// tickOnce drains one batch of tracks and one batch of artists. Per-row
-// errors are logged and skipped (passive retry via timer). 429 aborts the
-// entire tick.
+// tickOnce drains one batch of tracks and one batch of artists, then
+// re-resolves every cached answer against the library. Per-row errors are
+// logged and skipped. 429 aborts the fetching, not the resolve.
 func (w *Worker) tickOnce(ctx context.Context) error {
 	q := dbq.New(w.pool)
 	if err := w.tickTracks(ctx, q); err != nil {
 		return err
 	}
-	return w.tickArtists(ctx, q)
+	if err := w.tickArtists(ctx, q); err != nil {
+		return err
+	}
+	// Links recordings that reached the library since their seed was fetched:
+	// a scan, a Lidarr import, an MBID from the AcoustID lookup.
+	if err := q.ResolveListenBrainzTrackEdges(ctx, nil); err != nil {
+		return fmt.Errorf("resolve similarity edges: %w", err)
+	}
+	return nil
 }

 func (w *Worker) tickTracks(ctx context.Context, q *dbq.Queries) error {
@@ -85,16 +102,23 @@ func (w *Worker) tickTracks(ctx context.Context, q *dbq.Queries) error {
 			continue // defensive — query already filters NULL
 		}
 		results, err := w.client.SimilarRecordings(ctx, *r.Mbid, 100)
-		if err != nil {
-			var ra *listenbrainz.RetryAfterError
-			if errors.As(err, &ra) {
-				w.logger.Warn("similarity: 429 — aborting tick", "retry_after", ra.Wait)
-				return nil
-			}
+		switch {
+		case err == nil:
+		case isRateLimited(err):
+			w.logger.Warn("similarity: 429 — aborting tick")
+			return nil
+		case errors.Is(err, listenbrainz.ErrPermanent):
+			// LB refused this MBID outright; asking again next hour won't
+			// change that. Recorded as an empty answer so it waits its turn.
+			w.logger.Warn("similarity: similar-recordings refused", "track_id", r.ID, "err", err)
+			results = nil
+		default:
 			w.logger.Warn("similarity: similar-recordings failed", "track_id", r.ID, "err", err)
 			continue
 		}
-		w.upsertTrackSimilar(ctx, q, r.ID, results)
+		if err := w.storeTrackSimilar(ctx, r.ID, results); err != nil {
+			w.logger.Warn("similarity: store similar recordings", "track_id", r.ID, "err", err)
+		}
 	}
 	return nil
 }
@@ -109,64 +133,65 @@ func (w *Worker) tickArtists(ctx context.Context, q *dbq.Queries) error {
 			continue
 		}
 		results, err := w.client.SimilarArtists(ctx, *r.Mbid, 100)
-		if err != nil {
-			var ra *listenbrainz.RetryAfterError
-			if errors.As(err, &ra) {
-				w.logger.Warn("similarity: 429 — aborting tick", "retry_after", ra.Wait)
-				return nil
-			}
+		switch {
+		case err == nil:
+		case isRateLimited(err):
+			w.logger.Warn("similarity: 429 — aborting tick")
+			return nil
+		case errors.Is(err, listenbrainz.ErrPermanent):
+			w.logger.Warn("similarity: similar-artists refused", "artist_id", r.ID, "err", err)
+			results = nil
+		default:
 			w.logger.Warn("similarity: similar-artists failed", "artist_id", r.ID, "err", err)
 			continue
 		}
 		w.upsertArtistSimilar(ctx, q, r.ID, results)
+		if err := q.RecordArtistSimilarityFetch(ctx, dbq.RecordArtistSimilarityFetchParams{
+			ArtistID: r.ID, Returned: int32(len(results)),
+		}); err != nil {
+			w.logger.Warn("similarity: record artist fetch", "artist_id", r.ID, "err", err)
+		}
 	}
 	return nil
 }

-// upsertTrackSimilar filters returned MBIDs to those in our library, takes
-// top-K by score, and upserts rows.
-func (w *Worker) upsertTrackSimilar(ctx context.Context, q *dbq.Queries, trackAID pgtype.UUID, results []listenbrainz.SimilarRecording) {
-	if len(results) == 0 {
-		return
-	}
-	sort.Slice(results, func(i, j int) bool { return results[i].Score > results[j].Score })
+func isRateLimited(err error) bool {
+	var ra *listenbrainz.RetryAfterError
+	return errors.As(err, &ra)
+}

+// storeTrackSimilar replaces the seed's cached answer with this one, records
+// the fetch, and resolves the seed's edges, in one transaction so the queue
+// never sees a fetch whose answer is not stored.
+func (w *Worker) storeTrackSimilar(ctx context.Context, seedID pgtype.UUID, results []listenbrainz.SimilarRecording) error {
 	mbids := make([]string, 0, len(results))
+	scores := make([]float64, 0, len(results))
 	for _, r := range results {
 		mbids = append(mbids, r.MBID)
+		scores = append(scores, r.Score)
 	}
-	rows, err := q.GetTracksByMBIDs(ctx, mbids)
-	if err != nil {
-		w.logger.Warn("similarity: GetTracksByMBIDs", "err", err)
-		return
-	}
-	idByMBID := make(map[string]pgtype.UUID, len(rows))
-	for _, r := range rows {
-		if r.Mbid != nil {
-			idByMBID[*r.Mbid] = r.ID
+	return pgx.BeginFunc(ctx, w.pool, func(tx pgx.Tx) error {
+		q := dbq.New(tx)
+		if err := q.DeleteSimilarRecordingsForSeed(ctx, seedID); err != nil {
+			return fmt.Errorf("clear cached answer: %w", err)
 		}
-	}
-
-	taken := 0
-	for _, r := range results {
-		if taken >= w.topK {
-			break
+		if len(mbids) > 0 {
+			if err := q.InsertSimilarRecordings(ctx, dbq.InsertSimilarRecordingsParams{
+				SeedTrackID: seedID, Mbids: mbids, Scores: scores,
+			}); err != nil {
+				return fmt.Errorf("cache answer: %w", err)
+			}
 		}
-		localID, ok := idByMBID[r.MBID]
-		if !ok {
-			continue
+		if err := q.RecordTrackSimilarityFetch(ctx, dbq.RecordTrackSimilarityFetchParams{
+			TrackID: seedID, Returned: int32(len(results)),
+		}); err != nil {
+			return fmt.Errorf("record fetch: %w", err)
 		}
-		if localID == trackAID {
-			continue // defensive — DB CHECK constraint also catches self-edges
+		if err := q.ResolveListenBrainzTrackEdges(ctx, []pgtype.UUID{seedID}); err != nil {
+			return fmt.Errorf("resolve edges: %w", err)
 		}
-		if uerr := q.UpsertTrackSimilarity(ctx, dbq.UpsertTrackSimilarityParams{
-			TrackAID: trackAID, TrackBID: localID, Score: r.Score,
-		}); uerr != nil {
-			w.logger.Warn("similarity: UpsertTrackSimilarity", "err", uerr)
-			continue
-		}
-		taken++
-	}
+		return nil
+	})
 }

 func (w *Worker) upsertArtistSimilar(ctx context.Context, q *dbq.Queries, artistAID pgtype.UUID, results []listenbrainz.SimilarArtist) {
@@ -10,6 +10,7 @@ import (
 	"net/http/httptest"
 	"os"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -208,7 +209,9 @@ func TestTickOnce_MapsLBResponseToLocalLibrary(t *testing.T) {
 	}
 }

-func TestTickOnce_TopKEnforced(t *testing.T) {
+// The 20-per-seed cap is gone with the cache (#5296): LB's answer is at most
+// 50, and every recording of it that is in the library becomes an edge.
+func TestTickOnce_KeepsEveryInLibraryMatch(t *testing.T) {
 	f := newFixture(t)
 	mbidSeed := "11111111-1111-1111-1111-111111111111"
 	seed := seedTrack(t, f, "Seed", &mbidSeed)
@@ -230,30 +233,25 @@ func TestTickOnce_TopKEnforced(t *testing.T) {
 	if err := w.tickOnce(context.Background()); err != nil {
 		t.Fatalf("tickOnce: %v", err)
 	}
-	if got := countTrackSim(t, f, seed.ID); got != 20 {
-		t.Errorf("got %d rows, want exactly 20 (all 25 in-library; top 20 by score)", got)
+	if got := countTrackSim(t, f, seed.ID); got != 25 {
+		t.Errorf("got %d rows, want all 25 in-library matches", got)
 	}
 }

-func TestTickOnce_RespectsSevenDayCap(t *testing.T) {
+func TestTickOnce_RecentFetchIsNotRepeated(t *testing.T) {
 	f := newFixture(t)
 	mbid := "11111111-1111-1111-1111-111111111111"
 	trackA := seedTrack(t, f, "A", &mbid)
 	markPlayed(t, f, trackA.ID)
-	otherMbid := "55555555-5555-5555-5555-555555555555"
-	other := seedTrack(t, f, "Other", &otherMbid)
-	if _, err := f.pool.Exec(context.Background(),
-		`INSERT INTO track_similarity (track_a_id, track_b_id, score, source, fetched_at)
-		 VALUES ($1, $2, 0.5, 'listenbrainz', now())`, trackA.ID, other.ID); err != nil {
-		t.Fatalf("seed sim: %v", err)
-	}
-	beforeCount := countTrackSim(t, f, trackA.ID)
-	srv := stubLB(`[{"recording_mbid":"`+otherMbid+`","score":0.99}]`, `[]`, http.StatusOK)
+	recordFetch(t, f, trackA.ID, "now() - interval '29 days'")
+	srv, asked := recordingLB(func(string) string { return `[]` })
 	defer srv.Close()
 	w := newTestWorker(f, srv.URL)
-	_ = w.tickOnce(context.Background())
-	if got := countTrackSim(t, f, trackA.ID); got != beforeCount {
-		t.Errorf("worker re-queried fresh track: before=%d after=%d", beforeCount, got)
+	if err := w.tickOnce(context.Background()); err != nil {
+		t.Fatalf("tickOnce: %v", err)
+	}
+	if n := len(asked()); n != 0 {
+		t.Errorf("a seed fetched 29 days ago was asked again (%d requests)", n)
 	}
 }

@@ -266,9 +264,10 @@ func TestTickOnce_RefreshesStaleRow(t *testing.T) {
 	other := seedTrack(t, f, "Other", &otherMbid)
 	if _, err := f.pool.Exec(context.Background(),
 		`INSERT INTO track_similarity (track_a_id, track_b_id, score, source, fetched_at)
-		 VALUES ($1, $2, 0.5, 'listenbrainz', now() - interval '8 days')`, trackA.ID, other.ID); err != nil {
+		 VALUES ($1, $2, 0.5, 'listenbrainz', now() - interval '31 days')`, trackA.ID, other.ID); err != nil {
 		t.Fatalf("seed sim: %v", err)
 	}
+	recordFetch(t, f, trackA.ID, "now() - interval '31 days'")
 	srv := stubLB(`[{"recording_mbid":"`+otherMbid+`","score":0.95}]`, `[]`, http.StatusOK)
 	defer srv.Close()
 	w := newTestWorker(f, srv.URL)
@@ -467,3 +466,147 @@ func TestUpsertArtistSimilar_PersistsUnmatchedToTable(t *testing.T) {

 	_ = inLibArtist
 }
+
+// recordFetch stamps a seed as answered by ListenBrainz at the given SQL time.
+func recordFetch(t *testing.T, f fixture, trackID pgtype.UUID, at string) {
+	t.Helper()
+	if _, err := f.pool.Exec(context.Background(),
+		`INSERT INTO track_similarity_fetches (track_id, fetched_at, returned) VALUES ($1, `+at+`, 0)`,
+		trackID); err != nil {
+		t.Fatalf("record fetch: %v", err)
+	}
+}
+
+// recordingLB answers similar-recordings with answer(mbid) and similar-artists
+// with an empty list, and reports which recording MBIDs it was asked about.
+func recordingLB(answer func(mbid string) string) (*httptest.Server, func() []string) {
+	var mu sync.Mutex
+	var asked []string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !strings.Contains(r.URL.Path, "/similar-recordings/") {
+			_, _ = w.Write([]byte(`[]`))
+			return
+		}
+		mbid := r.URL.Query().Get("recording_mbids")
+		mu.Lock()
+		asked = append(asked, mbid)
+		mu.Unlock()
+		_, _ = w.Write([]byte(answer(mbid)))
+	}))
+	return srv, func() []string {
+		mu.Lock()
+		defer mu.Unlock()
+		return append([]string(nil), asked...)
+	}
+}
+
+// The starvation #3879 measured: a seed whose answer matched nothing in the
+// library wrote nothing, stayed due, and was asked again every tick ahead of
+// everything behind it. Recording the fetch moves the queue on.
+func TestTickOnce_UnmatchedAnswerDoesNotHoldTheQueue(t *testing.T) {
+	f := newFixture(t)
+	mbids := []string{
+		"11111111-1111-1111-1111-111111111111",
+		"22222222-2222-2222-2222-222222222222",
+		"33333333-3333-3333-3333-333333333333",
+	}
+	for i := range mbids {
+		tr := seedTrack(t, f, fmt.Sprintf("S%d", i), &mbids[i])
+		markPlayed(t, f, tr.ID)
+	}
+	srv, asked := recordingLB(func(string) string {
+		return `[{"recording_mbid":"99999999-9999-9999-9999-999999999999","score":0.9}]`
+	})
+	defer srv.Close()
+	w := newTestWorker(f, srv.URL)
+	w.batch = 1
+	for i := 0; i < len(mbids); i++ {
+		if err := w.tickOnce(context.Background()); err != nil {
+			t.Fatalf("tick %d: %v", i, err)
+		}
+	}
+	seen := map[string]bool{}
+	for _, m := range asked() {
+		seen[m] = true
+	}
+	if len(seen) != len(mbids) {
+		t.Errorf("three ticks of one asked about %d distinct seeds (%v), want all %d",
+			len(seen), asked(), len(mbids))
+	}
+	var fetched int
+	_ = f.pool.QueryRow(context.Background(), `SELECT count(*) FROM track_similarity_fetches`).Scan(&fetched)
+	if fetched != len(mbids) {
+		t.Errorf("track_similarity_fetches = %d, want %d", fetched, len(mbids))
+	}
+}
+
+// The point of the cache (#5296): a recording LB named while it was not in the
+// library becomes an edge once it arrives, on the next tick's resolve, without
+// asking LB again. One edge per recording when the library holds two copies,
+// and none to a copy that is missing.
+func TestTickOnce_CachedAnswerResolvesWhenTheRecordingArrives(t *testing.T) {
+	f := newFixture(t)
+	ctx := context.Background()
+	seedMbid := "11111111-1111-1111-1111-111111111111"
+	laterMbid := "77777777-7777-7777-7777-777777777777"
+	goneMbid := "88888888-8888-8888-8888-888888888888"
+	seed := seedTrack(t, f, "Seed", &seedMbid)
+	markPlayed(t, f, seed.ID)
+	srv, asked := recordingLB(func(string) string {
+		return `[{"recording_mbid":"` + laterMbid + `","score":0.9},
+		         {"recording_mbid":"` + goneMbid + `","score":0.8}]`
+	})
+	defer srv.Close()
+	w := newTestWorker(f, srv.URL)
+	if err := w.tickOnce(ctx); err != nil {
+		t.Fatalf("tick 1: %v", err)
+	}
+	if got := countTrackSim(t, f, seed.ID); got != 0 {
+		t.Fatalf("edges before the recording is in the library = %d, want 0", got)
+	}
+	var cached int
+	_ = f.pool.QueryRow(ctx, `SELECT count(*) FROM listenbrainz_similar_recordings WHERE seed_track_id = $1`,
+		seed.ID).Scan(&cached)
+	if cached != 2 {
+		t.Fatalf("cached answer rows = %d, want 2 (both out of library)", cached)
+	}
+
+	// The recording arrives twice (an mp3 beside its flac); the other one
+	// arrives only as a missing row.
+	_ = seedTrack(t, f, "Later mp3", &laterMbid)
+	_ = seedTrack(t, f, "Later flac", &laterMbid)
+	gone := seedTrack(t, f, "Gone", &goneMbid)
+	if _, err := f.pool.Exec(ctx, `UPDATE tracks SET missing_since = now() WHERE id = $1`, gone.ID); err != nil {
+		t.Fatal(err)
+	}
+	if err := w.tickOnce(ctx); err != nil {
+		t.Fatalf("tick 2: %v", err)
+	}
+	if got := countTrackSim(t, f, seed.ID); got != 1 {
+		t.Errorf("edges after the recording arrived = %d, want 1 (one per recording, none to a missing copy)", got)
+	}
+	if n := len(asked()); n != 1 {
+		t.Errorf("LB asked %d times, want 1: the resolve needs no new fetch", n)
+	}
+}
+
+// A 4xx answer is permanent for that MBID; it is recorded like an empty one
+// so the seed waits its turn instead of being re-asked every hour.
+func TestTickOnce_PermanentErrorCountsAsFetched(t *testing.T) {
+	f := newFixture(t)
+	mbid := "11111111-1111-1111-1111-111111111111"
+	trackA := seedTrack(t, f, "A", &mbid)
+	markPlayed(t, f, trackA.ID)
+	srv := stubLB(``, ``, http.StatusBadRequest)
+	defer srv.Close()
+	w := newTestWorker(f, srv.URL)
+	if err := w.tickOnce(context.Background()); err != nil {
+		t.Fatalf("tickOnce: %v", err)
+	}
+	var tracks, artists int
+	_ = f.pool.QueryRow(context.Background(), `SELECT count(*) FROM track_similarity_fetches`).Scan(&tracks)
+	_ = f.pool.QueryRow(context.Background(), `SELECT count(*) FROM artist_similarity_fetches`).Scan(&artists)
+	if tracks != 1 || artists != 1 {
+		t.Errorf("fetches recorded: tracks=%d artists=%d, want 1 and 1", tracks, artists)
+	}
+}