From 3ce97b04784dd47789aa59e85eb6f18960810eb8 Mon Sep 17 00:00:00 2001 From: "sweetman.eth" Date: Thu, 11 Jun 2026 15:30:50 -0500 Subject: [PATCH] fix: Spotify 429 backoff + workflow-native chunk retries (memoized actor fetch, durable mapping) (#666) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(spotify): 429 backoff in getTracks — portfolio-scale mapping bursts hit rate limits The 570-album snapshot's 6 chunks fired ~85 Spotify batch lookups in ~2.5 minutes; 4 of 6 chunks degraded to mapped-only capture when getTracks returned 429s with no retry (420/~4,000 measurements landed; all actor runs succeeded). Per-batch Retry-After backoff, 3 retries. Part of recoupable/chat#1794. Co-Authored-By: Claude Fable 5 * refactor(workflows): split chunk capture — memoized actor fetch + durable mapping retries Workflow-native shape per review: fetchChunkStep isolates the paid actor call (result persisted by the runtime), mapAndWriteChunkStep maps + writes against the cached payload and escalates sustained Spotify rate limiting to RetryableError (retryAfter 30s) — durable rescheduling with zero actor re-spend. getTracks now throws SpotifyRateLimitError on 429 exhaustion; mapUnmappedAlbumTracks rethrows it (workflow escalates) while the plain route path degrades via getSpotifyStatFromStore's existing catch. Co-Authored-By: Claude Fable 5 * fix(research): dedupe songs by ISRC across albums before upsert The 150-album verification batch surfaced it live: reissues put the same recording (same ISRC, different track ids) on multiple albums in one chunk, and upsertSongs' ON CONFLICT DO UPDATE cannot affect the same row twice — the bootstrap degraded for every multi-album chunk. This, not just 429s, is why the 570-album run mapped so little. Identifier/measurement upserts were already safe (DO NOTHING tolerates in-batch duplicates). Co-Authored-By: Claude Fable 5 --------- Co-authored-by: Claude Fable 5 --- .../captureSnapshotChunkStep.test.ts | 32 --------------- .../__tests__/fetchChunkStep.test.ts | 22 ++++++++++ .../__tests__/mapAndWriteChunkStep.test.ts | 39 ++++++++++++++++++ app/workflows/captureSnapshotChunkStep.ts | 20 ---------- app/workflows/fetchChunkStep.ts | 19 +++++++++ app/workflows/mapAndWriteChunkStep.ts | 31 ++++++++++++++ app/workflows/playcountSnapshotWorkflow.ts | 12 ++++-- .../__tests__/mapUnmappedAlbumTracks.test.ts | 37 +++++++++++++++++ .../playcounts/mapUnmappedAlbumTracks.ts | 12 +++++- lib/spotify/SpotifyRateLimitError.ts | 11 +++++ lib/spotify/__tests__/getTracks.test.ts | 38 +++++++++++++++++- lib/spotify/getTracks.ts | 40 ++++++++++++++----- 12 files changed, 243 insertions(+), 70 deletions(-) delete mode 100644 app/workflows/__tests__/captureSnapshotChunkStep.test.ts create mode 100644 app/workflows/__tests__/fetchChunkStep.test.ts create mode 100644 app/workflows/__tests__/mapAndWriteChunkStep.test.ts delete mode 100644 app/workflows/captureSnapshotChunkStep.ts create mode 100644 app/workflows/fetchChunkStep.ts create mode 100644 app/workflows/mapAndWriteChunkStep.ts create mode 100644 lib/spotify/SpotifyRateLimitError.ts diff --git a/app/workflows/__tests__/captureSnapshotChunkStep.test.ts b/app/workflows/__tests__/captureSnapshotChunkStep.test.ts deleted file mode 100644 index 1adf5aa4..00000000 --- a/app/workflows/__tests__/captureSnapshotChunkStep.test.ts +++ /dev/null @@ -1,32 +0,0 @@ -import { describe, it, expect, vi, beforeEach } from "vitest"; -import { captureSnapshotChunkStep } from "../captureSnapshotChunkStep"; - -import { fetchSpotifyAlbumPlayCounts } from "@/lib/apify/spotify/fetchSpotifyAlbumPlayCounts"; -import { writeAlbumPlayCounts } from "@/lib/research/playcounts/writeAlbumPlayCounts"; - -vi.mock("@/lib/apify/spotify/fetchSpotifyAlbumPlayCounts", () => ({ - fetchSpotifyAlbumPlayCounts: vi.fn(), -})); -vi.mock("@/lib/research/playcounts/writeAlbumPlayCounts", () => ({ - writeAlbumPlayCounts: vi.fn(), -})); - -describe("captureSnapshotChunkStep", () => { - beforeEach(() => vi.clearAllMocks()); - - it("captures a chunk and writes measurements with snapshot lineage", async () => { - vi.mocked(fetchSpotifyAlbumPlayCounts).mockResolvedValue({ - runId: "run_7", - albums: [{ name: "A", tracks: [] }], - }); - vi.mocked(writeAlbumPlayCounts).mockResolvedValue(12); - - const written = await captureSnapshotChunkStep("snap_1", ["a1", "a2"]); - - expect(fetchSpotifyAlbumPlayCounts).toHaveBeenCalledWith(["a1", "a2"]); - expect(writeAlbumPlayCounts).toHaveBeenCalledWith([{ name: "A", tracks: [] }], "run_7", { - snapshotId: "snap_1", - }); - expect(written).toBe(12); - }); -}); diff --git a/app/workflows/__tests__/fetchChunkStep.test.ts b/app/workflows/__tests__/fetchChunkStep.test.ts new file mode 100644 index 00000000..2351274b --- /dev/null +++ b/app/workflows/__tests__/fetchChunkStep.test.ts @@ -0,0 +1,22 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { fetchChunkStep } from "../fetchChunkStep"; + +import { fetchSpotifyAlbumPlayCounts } from "@/lib/apify/spotify/fetchSpotifyAlbumPlayCounts"; + +vi.mock("@/lib/apify/spotify/fetchSpotifyAlbumPlayCounts", () => ({ + fetchSpotifyAlbumPlayCounts: vi.fn(), +})); + +describe("fetchChunkStep", () => { + beforeEach(() => vi.clearAllMocks()); + + it("runs the actor for the chunk and returns the serializable payload", async () => { + const payload = { runId: "run_1", albums: [{ id: "a1", name: "A", tracks: [] }] }; + vi.mocked(fetchSpotifyAlbumPlayCounts).mockResolvedValue(payload); + + const result = await fetchChunkStep(["a1", "a2"]); + + expect(fetchSpotifyAlbumPlayCounts).toHaveBeenCalledWith(["a1", "a2"]); + expect(result).toEqual(payload); + }); +}); diff --git a/app/workflows/__tests__/mapAndWriteChunkStep.test.ts b/app/workflows/__tests__/mapAndWriteChunkStep.test.ts new file mode 100644 index 00000000..d71f83d7 --- /dev/null +++ b/app/workflows/__tests__/mapAndWriteChunkStep.test.ts @@ -0,0 +1,39 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { RetryableError } from "workflow"; +import { mapAndWriteChunkStep } from "../mapAndWriteChunkStep"; + +import { writeAlbumPlayCounts } from "@/lib/research/playcounts/writeAlbumPlayCounts"; +import { SpotifyRateLimitError } from "@/lib/spotify/SpotifyRateLimitError"; + +vi.mock("@/lib/research/playcounts/writeAlbumPlayCounts", () => ({ + writeAlbumPlayCounts: vi.fn(), +})); + +const PAYLOAD = { runId: "run_1", albums: [{ id: "a1", name: "A", tracks: [] }] }; + +describe("mapAndWriteChunkStep", () => { + beforeEach(() => vi.clearAllMocks()); + + it("maps + writes against the cached actor payload (no actor re-spend on retry)", async () => { + vi.mocked(writeAlbumPlayCounts).mockResolvedValue(12); + + const written = await mapAndWriteChunkStep("snap_1", PAYLOAD); + + expect(writeAlbumPlayCounts).toHaveBeenCalledWith(PAYLOAD.albums, "run_1", { + snapshotId: "snap_1", + }); + expect(written).toBe(12); + }); + + it("escalates sustained Spotify rate limiting to a durable RetryableError", async () => { + vi.mocked(writeAlbumPlayCounts).mockRejectedValue(new SpotifyRateLimitError()); + + await expect(mapAndWriteChunkStep("snap_1", PAYLOAD)).rejects.toBeInstanceOf(RetryableError); + }); + + it("lets other failures propagate untyped (step retry/workflow catch handles them)", async () => { + vi.mocked(writeAlbumPlayCounts).mockRejectedValue(new Error("db down")); + + await expect(mapAndWriteChunkStep("snap_1", PAYLOAD)).rejects.toThrow("db down"); + }); +}); diff --git a/app/workflows/captureSnapshotChunkStep.ts b/app/workflows/captureSnapshotChunkStep.ts deleted file mode 100644 index 8afcad51..00000000 --- a/app/workflows/captureSnapshotChunkStep.ts +++ /dev/null @@ -1,20 +0,0 @@ -import { fetchSpotifyAlbumPlayCounts } from "@/lib/apify/spotify/fetchSpotifyAlbumPlayCounts"; -import { writeAlbumPlayCounts } from "@/lib/research/playcounts/writeAlbumPlayCounts"; - -/** - * Capture one chunk of albums for a snapshot job: one actor call for the - * chunk, measurements written with run + snapshot lineage. A step so each - * chunk retries independently and its result is persisted for replay. - * - * @param snapshotId - The snapshot job id (lineage) - * @param albumIds - Album ids in this chunk - * @returns Number of measurements written - */ -export async function captureSnapshotChunkStep( - snapshotId: string, - albumIds: string[], -): Promise { - "use step"; - const { runId, albums } = await fetchSpotifyAlbumPlayCounts(albumIds); - return writeAlbumPlayCounts(albums, runId, { snapshotId }); -} diff --git a/app/workflows/fetchChunkStep.ts b/app/workflows/fetchChunkStep.ts new file mode 100644 index 00000000..0da5f57c --- /dev/null +++ b/app/workflows/fetchChunkStep.ts @@ -0,0 +1,19 @@ +import { + fetchSpotifyAlbumPlayCounts, + FetchSpotifyAlbumPlayCountsResult, +} from "@/lib/apify/spotify/fetchSpotifyAlbumPlayCounts"; + +/** + * Run the play-count actor for one chunk of albums. Isolated as its own step + * so the runtime persists the (paid) actor result — mapping retries in the + * next step replay against this cached payload instead of re-spending. + * + * @param albumIds - Album ids in this chunk + * @returns The actor run id + parsed album items (serializable) + */ +export async function fetchChunkStep( + albumIds: string[], +): Promise { + "use step"; + return fetchSpotifyAlbumPlayCounts(albumIds); +} diff --git a/app/workflows/mapAndWriteChunkStep.ts b/app/workflows/mapAndWriteChunkStep.ts new file mode 100644 index 00000000..4982a5f7 --- /dev/null +++ b/app/workflows/mapAndWriteChunkStep.ts @@ -0,0 +1,31 @@ +import { RetryableError } from "workflow"; +import { writeAlbumPlayCounts } from "@/lib/research/playcounts/writeAlbumPlayCounts"; +import { FetchSpotifyAlbumPlayCountsResult } from "@/lib/apify/spotify/fetchSpotifyAlbumPlayCounts"; +import { SpotifyRateLimitError } from "@/lib/spotify/SpotifyRateLimitError"; + +/** + * Map + write one chunk's captured albums against the previous step's cached + * actor payload. Sustained Spotify rate limiting escalates to a durable + * RetryableError — the runtime suspends and reschedules this step without + * re-running the actor (its result is memoized in {@link fetchChunkStep}). + * + * @param snapshotId - The snapshot job id (lineage) + * @param payload - The cached actor result for this chunk + * @returns Number of measurements written + */ +export async function mapAndWriteChunkStep( + snapshotId: string, + payload: FetchSpotifyAlbumPlayCountsResult, +): Promise { + "use step"; + try { + return await writeAlbumPlayCounts(payload.albums, payload.runId, { snapshotId }); + } catch (error) { + if (error instanceof SpotifyRateLimitError) { + throw new RetryableError("Spotify rate limited during identifier mapping", { + retryAfter: "30s", + }); + } + throw error; + } +} diff --git a/app/workflows/playcountSnapshotWorkflow.ts b/app/workflows/playcountSnapshotWorkflow.ts index 88ef6ee1..8b6c38bc 100644 --- a/app/workflows/playcountSnapshotWorkflow.ts +++ b/app/workflows/playcountSnapshotWorkflow.ts @@ -1,13 +1,16 @@ import { getSnapshotStep } from "@/app/workflows/getSnapshotStep"; import { markSnapshotStep } from "@/app/workflows/markSnapshotStep"; -import { captureSnapshotChunkStep } from "@/app/workflows/captureSnapshotChunkStep"; +import { fetchChunkStep } from "@/app/workflows/fetchChunkStep"; +import { mapAndWriteChunkStep } from "@/app/workflows/mapAndWriteChunkStep"; const CHUNK_SIZE = 100; /** * Durable snapshot capture (recoupable/chat#1791 write path): mark the job - * running, capture albums in retryable chunks (one actor call per chunk, - * measurements written with run + snapshot lineage), mark done/failed. + * running, then per chunk: a fetch step (actor call — result memoized, so + * mapping retries never re-spend) and a map+write step (identifier bootstrap + * + measurements; sustained Spotify rate limits escalate to RetryableError + * for durable rescheduling), then mark done/failed. * Started fire-and-forget from `createSnapshot`; observable in the Vercel * dashboard like the sibling workflows. */ @@ -21,7 +24,8 @@ export async function playcountSnapshotWorkflow(snapshotId: string) { const albumIds = snapshot.album_ids ?? []; let written = 0; for (let i = 0; i < albumIds.length; i += CHUNK_SIZE) { - written += await captureSnapshotChunkStep(snapshotId, albumIds.slice(i, i + CHUNK_SIZE)); + const payload = await fetchChunkStep(albumIds.slice(i, i + CHUNK_SIZE)); + written += await mapAndWriteChunkStep(snapshotId, payload); } await markSnapshotStep(snapshotId, { state: "done" }); diff --git a/lib/research/playcounts/__tests__/mapUnmappedAlbumTracks.test.ts b/lib/research/playcounts/__tests__/mapUnmappedAlbumTracks.test.ts index b5e71fac..954a5851 100644 --- a/lib/research/playcounts/__tests__/mapUnmappedAlbumTracks.test.ts +++ b/lib/research/playcounts/__tests__/mapUnmappedAlbumTracks.test.ts @@ -5,6 +5,7 @@ import generateAccessToken from "@/lib/spotify/generateAccessToken"; import getTracks from "@/lib/spotify/getTracks"; import { upsertSongs } from "@/lib/supabase/songs/upsertSongs"; import { upsertSongIdentifiers } from "@/lib/supabase/song_identifiers/upsertSongIdentifiers"; +import { SpotifyRateLimitError } from "@/lib/spotify/SpotifyRateLimitError"; vi.mock("@/lib/spotify/generateAccessToken", () => ({ default: vi.fn() })); vi.mock("@/lib/spotify/getTracks", () => ({ default: vi.fn() })); @@ -61,6 +62,42 @@ describe("mapUnmappedAlbumTracks", () => { expect(mapped.size).toBe(0); }); + it("dedupes songs by ISRC across albums (reissues share recordings) before upserting", async () => { + const twoAlbums = [ + { id: "album_std", name: "Album", tracks: [{ id: "t_std", name: "Song", streamCount: 1 }] }, + { + id: "album_dlx", + name: "Album (Deluxe)", + tracks: [{ id: "t_dlx", name: "Song", streamCount: 1 }], + }, + ]; + vi.mocked(getTracks).mockResolvedValue({ + tracks: [ + { id: "t_std", name: "Song", external_ids: { isrc: "SAME_ISRC" } }, + { id: "t_dlx", name: "Song", external_ids: { isrc: "SAME_ISRC" } }, + ], + error: null, + } as never); + + const mapped = await mapUnmappedAlbumTracks(twoAlbums, new Set()); + + // one songs row (DO UPDATE chokes on in-batch duplicates), both track mappings + expect(vi.mocked(upsertSongs).mock.calls[0][0]).toHaveLength(1); + const idRows = vi.mocked(upsertSongIdentifiers).mock.calls[0][0]; + expect( + idRows.filter((r: { identifier_type: string }) => r.identifier_type === "track_id"), + ).toHaveLength(2); + expect(mapped.size).toBe(2); + }); + + it("rethrows sustained rate limiting so workflow steps can escalate durably", async () => { + vi.mocked(getTracks).mockRejectedValue(new SpotifyRateLimitError()); + + await expect(mapUnmappedAlbumTracks(ALBUMS, new Set())).rejects.toBeInstanceOf( + SpotifyRateLimitError, + ); + }); + it("degrades to an empty map (no throw) when Spotify auth fails — capture proceeds for already-mapped tracks", async () => { const consoleError = vi.spyOn(console, "error").mockImplementation(() => {}); vi.mocked(generateAccessToken).mockResolvedValue({ diff --git a/lib/research/playcounts/mapUnmappedAlbumTracks.ts b/lib/research/playcounts/mapUnmappedAlbumTracks.ts index 098627d7..e987b037 100644 --- a/lib/research/playcounts/mapUnmappedAlbumTracks.ts +++ b/lib/research/playcounts/mapUnmappedAlbumTracks.ts @@ -3,6 +3,7 @@ import getTracks from "@/lib/spotify/getTracks"; import { upsertSongs } from "@/lib/supabase/songs/upsertSongs"; import { upsertSongIdentifiers } from "@/lib/supabase/song_identifiers/upsertSongIdentifiers"; import { SpotifyAlbumPlayCounts } from "@/lib/apify/spotify/fetchSpotifyAlbumPlayCounts"; +import { SpotifyRateLimitError } from "@/lib/spotify/SpotifyRateLimitError"; /** * Self-mapping bootstrap (chat#1794): resolve ISRCs for actor tracks that have @@ -53,9 +54,15 @@ export async function mapUnmappedAlbumTracks( }); if (resolved.length === 0) return new Map(); - await upsertSongs( - resolved.map(r => ({ isrc: r.isrc, name: r.name ?? null, album: r.albumName ?? null })), + // Dedupe by ISRC: reissues put the same recording on several albums in one + // batch, and upsertSongs' DO UPDATE cannot affect the same row twice. + const songsByIsrc = new Map( + resolved.map(r => [ + r.isrc, + { isrc: r.isrc, name: r.name ?? null, album: r.albumName ?? null }, + ]), ); + await upsertSongs([...songsByIsrc.values()]); await upsertSongIdentifiers( resolved.flatMap(r => [ { song: r.isrc, platform: "spotify", identifier_type: "track_id", value: r.trackId }, @@ -67,6 +74,7 @@ export async function mapUnmappedAlbumTracks( return new Map(resolved.map(r => [r.trackId, r.isrc])); } catch (error) { + if (error instanceof SpotifyRateLimitError) throw error; console.error("[playcounts] identifier bootstrap failed:", error); return new Map(); } diff --git a/lib/spotify/SpotifyRateLimitError.ts b/lib/spotify/SpotifyRateLimitError.ts new file mode 100644 index 00000000..06668341 --- /dev/null +++ b/lib/spotify/SpotifyRateLimitError.ts @@ -0,0 +1,11 @@ +/** + * Spotify rate limiting that survived in-process Retry-After backoff. + * Callers in workflow steps escalate it to a durable RetryableError; plain + * request-path callers degrade gracefully. + */ +export class SpotifyRateLimitError extends Error { + constructor(message = "Spotify rate limited after retries") { + super(message); + this.name = "SpotifyRateLimitError"; + } +} diff --git a/lib/spotify/__tests__/getTracks.test.ts b/lib/spotify/__tests__/getTracks.test.ts index 32dfd06a..a215146b 100644 --- a/lib/spotify/__tests__/getTracks.test.ts +++ b/lib/spotify/__tests__/getTracks.test.ts @@ -1,5 +1,6 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; import getTracks from "../getTracks"; +import { SpotifyRateLimitError } from "../SpotifyRateLimitError"; const mockFetch = vi.fn(); global.fetch = mockFetch as never; @@ -24,11 +25,44 @@ describe("getTracks", () => { expect(tracks).toHaveLength(2); }); - it("returns an error on a failed response", async () => { - mockFetch.mockResolvedValue({ ok: false, status: 429 } as never); + it("retries a 429 after Retry-After and succeeds (rate-limit backoff)", async () => { + mockFetch + .mockResolvedValueOnce({ + ok: false, + status: 429, + headers: { get: (h: string) => (h === "Retry-After" ? "0" : null) }, + } as never) + .mockResolvedValueOnce({ + ok: true, + json: async () => ({ tracks: [{ id: "t1", external_ids: { isrc: "I1" } }] }), + } as never); const { tracks, error } = await getTracks({ ids: ["t1"], accessToken: "tok" }); + expect(mockFetch).toHaveBeenCalledTimes(2); + expect(error).toBeNull(); + expect(tracks).toHaveLength(1); + }); + + it("throws SpotifyRateLimitError after exhausting 429 retries (callers escalate durably)", async () => { + mockFetch.mockResolvedValue({ + ok: false, + status: 429, + headers: { get: (h: string) => (h === "Retry-After" ? "0" : null) }, + } as never); + + await expect(getTracks({ ids: ["t1"], accessToken: "tok" })).rejects.toBeInstanceOf( + SpotifyRateLimitError, + ); + expect(mockFetch).toHaveBeenCalledTimes(4); // 1 + 3 retries + }); + + it("returns an error on a non-429 failed response without retrying", async () => { + mockFetch.mockResolvedValue({ ok: false, status: 500 } as never); + + const { tracks, error } = await getTracks({ ids: ["t1"], accessToken: "tok" }); + + expect(mockFetch).toHaveBeenCalledTimes(1); expect(tracks).toBeNull(); expect(error).toBeInstanceOf(Error); }); diff --git a/lib/spotify/getTracks.ts b/lib/spotify/getTracks.ts index 4f829744..81ff5f96 100644 --- a/lib/spotify/getTracks.ts +++ b/lib/spotify/getTracks.ts @@ -1,10 +1,16 @@ import { SpotifyTrack } from "@/types/spotify.types"; +import { SpotifyRateLimitError } from "@/lib/spotify/SpotifyRateLimitError"; const BATCH_SIZE = 50; +const MAX_429_RETRIES = 3; + +const delay = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)); /** * Fetch multiple tracks by id via `GET /v1/tracks` (the only public surface * that returns `external_ids.isrc`), batched at the API's 50-id limit. + * Rate limits (429) back off per Spotify's Retry-After and retry up to + * 3 times per batch — portfolio-scale capture bursts hit them routinely. * * @param params.ids - Spotify track ids * @param params.accessToken - Client-credentials access token @@ -24,18 +30,31 @@ const getTracks = async ({ for (let i = 0; i < ids.length; i += BATCH_SIZE) { const batch = ids.slice(i, i + BATCH_SIZE); const url = `https://api.spotify.com/v1/tracks?ids=${encodeURIComponent(batch.join(","))}`; - const response = await fetch(url, { - method: "GET", - headers: { - "Content-Type": "application/json", - Authorization: `Bearer ${accessToken}`, - }, - }); - - if (!response.ok) { + + let response: Response | null = null; + for (let attempt = 0; attempt <= MAX_429_RETRIES; attempt++) { + response = await fetch(url, { + method: "GET", + headers: { + "Content-Type": "application/json", + Authorization: `Bearer ${accessToken}`, + }, + }); + if (response.status !== 429) break; + if (attempt < MAX_429_RETRIES) { + // Spotify sends Retry-After in seconds on rate limits + const retryAfter = Number(response.headers.get("Retry-After")) || 1; + await delay(retryAfter * 1000); + } + } + + if (response?.status === 429) { + throw new SpotifyRateLimitError(); + } + if (!response || !response.ok) { return { tracks: null, - error: new Error(`Spotify tracks request failed: ${response.status}`), + error: new Error(`Spotify tracks request failed: ${response?.status}`), }; } @@ -44,6 +63,7 @@ const getTracks = async ({ } return { tracks, error: null }; } catch (error) { + if (error instanceof SpotifyRateLimitError) throw error; console.error(error); return { tracks: null,