diff --git a/app/workflows/__tests__/backfillTrackStep.test.ts b/app/workflows/__tests__/backfillTrackStep.test.ts index 6c211ed6..c923cb03 100644 --- a/app/workflows/__tests__/backfillTrackStep.test.ts +++ b/app/workflows/__tests__/backfillTrackStep.test.ts @@ -3,7 +3,6 @@ import { backfillTrackStep } from "../backfillTrackStep"; import { fetchSongstatsWithBackoff } from "@/lib/songstats/fetchSongstatsWithBackoff"; import { upsertSongMeasurements } from "@/lib/supabase/song_measurements/upsertSongMeasurements"; -import { insertSongstatsQuotaLedger } from "@/lib/supabase/songstats_quota_ledger/insertSongstatsQuotaLedger"; import { updateSongstatsBackfillQueue } from "@/lib/supabase/songstats_backfill_queue/updateSongstatsBackfillQueue"; vi.mock("@/lib/songstats/fetchSongstatsWithBackoff", () => ({ @@ -12,9 +11,6 @@ vi.mock("@/lib/songstats/fetchSongstatsWithBackoff", () => ({ vi.mock("@/lib/supabase/song_measurements/upsertSongMeasurements", () => ({ upsertSongMeasurements: vi.fn(), })); -vi.mock("@/lib/supabase/songstats_quota_ledger/insertSongstatsQuotaLedger", () => ({ - insertSongstatsQuotaLedger: vi.fn(), -})); vi.mock("@/lib/supabase/songstats_backfill_queue/updateSongstatsBackfillQueue", () => ({ updateSongstatsBackfillQueue: vi.fn(), })); @@ -28,7 +24,7 @@ describe("backfillTrackStep", () => { vi.mocked(upsertSongMeasurements).mockResolvedValue([] as never); }); - it("writes the historic series, records the spend, marks done on 200", async () => { + it("writes the historic series and marks done on 200", async () => { vi.mocked(fetchSongstatsWithBackoff).mockResolvedValue({ status: 200, attempts: 1, @@ -61,15 +57,11 @@ describe("backfillTrackStep", () => { raw_ref: "songstats-backfill", }, ]); - expect(insertSongstatsQuotaLedger).toHaveBeenCalledWith({ - hits: 1, - purpose: "backfill USA2P2015959", - }); expect(updateSongstatsBackfillQueue).toHaveBeenCalledWith(["q1"], { status: "done" }); - expect(result).toEqual({ ok: true, hitsSpent: 1 }); + expect(result).toEqual({ ok: true }); }); - it("DEFERS (pending, no quota hit, signals stop) when backoff is exhausted on 429", async () => { + it("DEFERS (pending, signals stop) when backoff is exhausted on 429", async () => { vi.mocked(fetchSongstatsWithBackoff).mockResolvedValue({ status: 429, attempts: 6, @@ -79,14 +71,12 @@ describe("backfillTrackStep", () => { const result = await backfillTrackStep(ROW); - // left pending for the next drain; NO ledger hit (Songstats consumed nothing) expect(updateSongstatsBackfillQueue).toHaveBeenCalledWith(["q1"], { status: "pending" }); - expect(insertSongstatsQuotaLedger).not.toHaveBeenCalled(); expect(upsertSongMeasurements).not.toHaveBeenCalled(); - expect(result).toEqual({ ok: false, hitsSpent: 0, deferred: true }); + expect(result).toEqual({ ok: false, deferred: true }); }); - it("marks a definitive 404 (no history) as done and records the spend", async () => { + it("marks a definitive 404 (no history) as done", async () => { vi.mocked(fetchSongstatsWithBackoff).mockResolvedValue({ status: 404, attempts: 1, @@ -97,14 +87,10 @@ describe("backfillTrackStep", () => { const result = await backfillTrackStep(ROW); expect(updateSongstatsBackfillQueue).toHaveBeenCalledWith(["q1"], { status: "done" }); - expect(insertSongstatsQuotaLedger).toHaveBeenCalledWith({ - hits: 1, - purpose: "backfill USA2P2015959 (no data 404)", - }); - expect(result).toEqual({ ok: false, hitsSpent: 1 }); + expect(result).toEqual({ ok: false }); }); - it("marks a permanent 4xx (403) as done (terminal) and records the spend", async () => { + it("marks a permanent 4xx (403) as done (terminal)", async () => { vi.mocked(fetchSongstatsWithBackoff).mockResolvedValue({ status: 403, attempts: 1, @@ -115,10 +101,6 @@ describe("backfillTrackStep", () => { const result = await backfillTrackStep(ROW); expect(updateSongstatsBackfillQueue).toHaveBeenCalledWith(["q1"], { status: "done" }); - expect(insertSongstatsQuotaLedger).toHaveBeenCalledWith({ - hits: 1, - purpose: "backfill USA2P2015959 (terminal 403)", - }); - expect(result).toEqual({ ok: false, hitsSpent: 1 }); + expect(result).toEqual({ ok: false }); }); }); diff --git a/app/workflows/__tests__/getBackfillBudgetStep.test.ts b/app/workflows/__tests__/getBackfillBudgetStep.test.ts deleted file mode 100644 index 6c3fb815..00000000 --- a/app/workflows/__tests__/getBackfillBudgetStep.test.ts +++ /dev/null @@ -1,32 +0,0 @@ -import { describe, it, expect, vi, beforeEach } from "vitest"; -import { getBackfillBudgetStep } from "../getBackfillBudgetStep"; - -import { selectSongstatsQuotaSpent } from "@/lib/supabase/songstats_quota_ledger/selectSongstatsQuotaSpent"; - -vi.mock("@/lib/supabase/songstats_quota_ledger/selectSongstatsQuotaSpent", () => ({ - selectSongstatsQuotaSpent: vi.fn(), -})); - -describe("getBackfillBudgetStep", () => { - beforeEach(() => { - vi.clearAllMocks(); - delete process.env.SONGSTATS_QUOTA_LIMIT; - delete process.env.SONGSTATS_QUOTA_RESERVE; - }); - - it("computes limit - reserve - spent over the rolling 30d window", async () => { - vi.mocked(selectSongstatsQuotaSpent).mockResolvedValue(300); - - const budget = await getBackfillBudgetStep(); - - expect(budget).toBe(1000 - 100 - 300); - const since = vi.mocked(selectSongstatsQuotaSpent).mock.calls[0][0]; - const ageDays = (Date.now() - new Date(since).getTime()) / 86400000; - expect(Math.round(ageDays)).toBe(30); - }); - - it("never returns negative", async () => { - vi.mocked(selectSongstatsQuotaSpent).mockResolvedValue(5000); - expect(await getBackfillBudgetStep()).toBe(0); - }); -}); diff --git a/app/workflows/__tests__/songstatsBackfillWorkflow.test.ts b/app/workflows/__tests__/songstatsBackfillWorkflow.test.ts index cc435558..b18c5aeb 100644 --- a/app/workflows/__tests__/songstatsBackfillWorkflow.test.ts +++ b/app/workflows/__tests__/songstatsBackfillWorkflow.test.ts @@ -1,12 +1,10 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; import { songstatsBackfillWorkflow } from "../songstatsBackfillWorkflow"; -import { getBackfillBudgetStep } from "../getBackfillBudgetStep"; import { claimBackfillRowsStep } from "../claimBackfillRowsStep"; import { backfillTrackStep } from "../backfillTrackStep"; import { releaseClaimedRowsStep } from "../releaseClaimedRowsStep"; -vi.mock("../getBackfillBudgetStep", () => ({ getBackfillBudgetStep: vi.fn() })); vi.mock("../claimBackfillRowsStep", () => ({ claimBackfillRowsStep: vi.fn() })); vi.mock("../backfillTrackStep", () => ({ backfillTrackStep: vi.fn() })); vi.mock("../releaseClaimedRowsStep", () => ({ releaseClaimedRowsStep: vi.fn() })); @@ -21,41 +19,39 @@ describe("songstatsBackfillWorkflow", () => { }); it("releases the rest of the claimed batch to pending when a track defers", async () => { - vi.mocked(getBackfillBudgetStep).mockResolvedValue(100); vi.mocked(claimBackfillRowsStep).mockResolvedValue([row("r1"), row("r2"), row("r3")]); vi.mocked(backfillTrackStep) - .mockResolvedValueOnce({ ok: true, hitsSpent: 1 }) // r1 - .mockResolvedValueOnce({ ok: false, hitsSpent: 0, deferred: true }); // r2 defers + .mockResolvedValueOnce({ ok: true }) // r1 + .mockResolvedValueOnce({ ok: false, deferred: true }); // r2 defers const result = await songstatsBackfillWorkflow(); // r2 is set pending by the step itself; the unprocessed remainder (r3) is released here expect(releaseClaimedRowsStep).toHaveBeenCalledWith(["r3"]); expect(backfillTrackStep).toHaveBeenCalledTimes(2); // stopped at the defer, never reached r3 - expect(result).toEqual({ backfilled: 1, failed: 0, deferred: true }); + expect(result).toEqual({ backfilled: 1, terminal: 0, deferred: true }); }); it("drains until the queue is empty and never releases when nothing defers", async () => { - vi.mocked(getBackfillBudgetStep).mockResolvedValue(100); vi.mocked(claimBackfillRowsStep) .mockResolvedValueOnce([row("a"), row("b")]) .mockResolvedValueOnce([]); // queue drained vi.mocked(backfillTrackStep) - .mockResolvedValueOnce({ ok: true, hitsSpent: 1 }) - .mockResolvedValueOnce({ ok: false, hitsSpent: 1 }); // terminal (e.g. 404) + .mockResolvedValueOnce({ ok: true }) + .mockResolvedValueOnce({ ok: false }); // terminal (e.g. 404) const result = await songstatsBackfillWorkflow(); expect(releaseClaimedRowsStep).not.toHaveBeenCalled(); - expect(result).toEqual({ backfilled: 1, failed: 1, deferred: false }); + expect(result).toEqual({ backfilled: 1, terminal: 1, deferred: false }); }); - it("does not drain when there is no budget", async () => { - vi.mocked(getBackfillBudgetStep).mockResolvedValue(0); + it("stops immediately when the first claim is empty", async () => { + vi.mocked(claimBackfillRowsStep).mockResolvedValue([]); const result = await songstatsBackfillWorkflow(); - expect(claimBackfillRowsStep).not.toHaveBeenCalled(); - expect(result).toEqual({ backfilled: 0, failed: 0, deferred: false }); + expect(backfillTrackStep).not.toHaveBeenCalled(); + expect(result).toEqual({ backfilled: 0, terminal: 0, deferred: false }); }); }); diff --git a/app/workflows/backfillTrackStep.ts b/app/workflows/backfillTrackStep.ts index fd394743..4a627ee8 100644 --- a/app/workflows/backfillTrackStep.ts +++ b/app/workflows/backfillTrackStep.ts @@ -1,6 +1,5 @@ import { fetchSongstatsWithBackoff } from "@/lib/songstats/fetchSongstatsWithBackoff"; import { upsertSongMeasurements } from "@/lib/supabase/song_measurements/upsertSongMeasurements"; -import { insertSongstatsQuotaLedger } from "@/lib/supabase/songstats_quota_ledger/insertSongstatsQuotaLedger"; import { updateSongstatsBackfillQueue } from "@/lib/supabase/songstats_backfill_queue/updateSongstatsBackfillQueue"; import { historicStatsPayloadSchema } from "@/lib/research/playcounts/historicStatsPayloadSchema"; import { Tables } from "@/types/database.types"; @@ -9,21 +8,22 @@ const METRIC = "platform_displayed_play_count"; /** * Backfill one claimed queue row, with bounded exponential backoff on Songstats' - * rate limit (Songstats is the rate authority — see chat#1797): + * rate limit (Songstats is the rate authority — see chat#1797; there is no local + * quota ledger): * - **200** → write each history point as a permanent `songstats` measurement, - * record the spend, mark `done`. + * mark `done`. * - **404 / other 4xx** → a real request with a definitive answer; terminal, so - * mark `done` (404 = no history) and record the spend. + * mark `done` (404 = no history). * - **backoff exhausted** (still 429/5xx after retries) → **defer**: leave the row - * `pending` for the next drain, consume no quota, and signal the workflow to - * stop (`deferred`) — Songstats is saturated right now. + * `pending` for the next drain and signal the workflow to stop (`deferred`) — + * Songstats is saturated right now. * * @param row - The claimed queue row (already in_progress) - * @returns ok + hitsSpent (0 when deferred) + `deferred` when Songstats is saturated + * @returns `ok` (true on a written backfill) + `deferred` when Songstats is saturated */ export async function backfillTrackStep( row: Tables<"songstats_backfill_queue">, -): Promise<{ ok: boolean; hitsSpent: number; deferred?: boolean }> { +): Promise<{ ok: boolean; deferred?: boolean }> { "use step"; const result = await fetchSongstatsWithBackoff("tracks/historic_stats", { isrc: row.song, @@ -32,12 +32,12 @@ export async function backfillTrackStep( if (result.retriesExhausted) { // Still retryable (429 throttle / 408 / gateway 5xx) past the backoff bound — - // leave it for the next run, spend nothing. + // leave it for the next run. console.log( `[backfill] ${row.song} deferred (retryable ${result.status} after ${result.attempts} tries)`, ); await updateSongstatsBackfillQueue([row.id], { status: "pending" }); - return { ok: false, hitsSpent: 0, deferred: true }; + return { ok: false, deferred: true }; } if (result.status !== 200) { @@ -45,12 +45,8 @@ export async function backfillTrackStep( console.log( `[backfill] ${row.song} done (${noData ? "no data 404" : `terminal ${result.status}`})`, ); - await insertSongstatsQuotaLedger({ - hits: 1, - purpose: `backfill ${row.song} (${noData ? "no data 404" : `terminal ${result.status}`})`, - }); await updateSongstatsBackfillQueue([row.id], { status: "done" }); - return { ok: false, hitsSpent: 1 }; + return { ok: false }; } const parsed = historicStatsPayloadSchema.safeParse(result.data); @@ -76,7 +72,6 @@ export async function backfillTrackStep( await upsertSongMeasurements(rows); console.log(`[backfill] ${row.song} done (${rows.length} points written)`); - await insertSongstatsQuotaLedger({ hits: 1, purpose: `backfill ${row.song}` }); await updateSongstatsBackfillQueue([row.id], { status: "done" }); - return { ok: true, hitsSpent: 1 }; + return { ok: true }; } diff --git a/app/workflows/getBackfillBudgetStep.ts b/app/workflows/getBackfillBudgetStep.ts deleted file mode 100644 index 866ee760..00000000 --- a/app/workflows/getBackfillBudgetStep.ts +++ /dev/null @@ -1,21 +0,0 @@ -import { selectSongstatsQuotaSpent } from "@/lib/supabase/songstats_quota_ledger/selectSongstatsQuotaSpent"; - -const DEFAULT_QUOTA_LIMIT = 1000; -const DEFAULT_QUOTA_RESERVE = 100; -const WINDOW_DAYS = 30; - -/** - * Remaining Songstats budget for this run: plan limit minus a reserve (kept - * for request-path fallbacks and manual research) minus hits spent in the - * rolling 30-day window. Never negative. - * - * @returns Hits the backfill worker may spend now - */ -export async function getBackfillBudgetStep(): Promise { - "use step"; - const limit = Number(process.env.SONGSTATS_QUOTA_LIMIT) || DEFAULT_QUOTA_LIMIT; - const reserve = Number(process.env.SONGSTATS_QUOTA_RESERVE) || DEFAULT_QUOTA_RESERVE; - const since = new Date(Date.now() - WINDOW_DAYS * 24 * 60 * 60 * 1000).toISOString(); - const spent = await selectSongstatsQuotaSpent(since); - return Math.max(0, limit - reserve - spent); -} diff --git a/app/workflows/songstatsBackfillWorkflow.ts b/app/workflows/songstatsBackfillWorkflow.ts index f94c0223..459d1887 100644 --- a/app/workflows/songstatsBackfillWorkflow.ts +++ b/app/workflows/songstatsBackfillWorkflow.ts @@ -1,4 +1,3 @@ -import { getBackfillBudgetStep } from "@/app/workflows/getBackfillBudgetStep"; import { claimBackfillRowsStep } from "@/app/workflows/claimBackfillRowsStep"; import { backfillTrackStep } from "@/app/workflows/backfillTrackStep"; import { releaseClaimedRowsStep } from "@/app/workflows/releaseClaimedRowsStep"; @@ -8,24 +7,22 @@ const BATCH_SIZE = 25; /** * Durable Songstats backfill drain (recoupable/chat#1791 write path): claim * value-ranked rows via the SKIP LOCKED RPC and backfill each track's historic - * series, with per-track exponential backoff handling Songstats' rate limit - * (chat#1797). **Stops as soon as a track defers** — Songstats still - * rate-limiting it past the backoff bound — releasing the rest of the claimed - * batch back to `pending` (so the next drain retries them immediately rather - * than waiting on stale-reclaim) instead of hammering a saturated API. Every - * successful hit converts into permanent owned data (fetch-once: captured - * history is never refetched). + * series. Per-track exponential backoff absorbs Songstats' rate limit; the run + * **stops as soon as a track defers** (still retryable past the backoff bound), + * releasing the rest of the claimed batch back to `pending` so the next drain + * retries them immediately rather than waiting on stale-reclaim. Otherwise it + * drains until the queue has no claimable `pending` rows. Every backfill + * converts into permanent owned data (fetch-once). */ export async function songstatsBackfillWorkflow() { "use workflow"; - let budget = await getBackfillBudgetStep(); let backfilled = 0; - let failed = 0; + let terminal = 0; let deferred = false; - drain: while (budget > 0) { - const rows = await claimBackfillRowsStep(Math.min(budget, BATCH_SIZE)); + drain: while (true) { + const rows = await claimBackfillRowsStep(BATCH_SIZE); if (rows.length === 0) break; console.log(`[songstats-backfill] claimed ${rows.length} rows`); @@ -39,16 +36,14 @@ export async function songstatsBackfillWorkflow() { await releaseClaimedRowsStep(rows.slice(i + 1).map(r => r.id)); break drain; } - budget -= result.hitsSpent; if (result.ok) backfilled += 1; - else failed += 1; - if (budget <= 0) break; + else terminal += 1; } } console.log( - `[songstats-backfill] done: ${backfilled} backfilled, ${failed} terminal` + + `[songstats-backfill] done: ${backfilled} backfilled, ${terminal} terminal` + (deferred ? ", deferred (rate-limited)" : ""), ); - return { backfilled, failed, deferred }; + return { backfilled, terminal, deferred }; } diff --git a/lib/supabase/songstats_quota_ledger/__tests__/insertSongstatsQuotaLedger.test.ts b/lib/supabase/songstats_quota_ledger/__tests__/insertSongstatsQuotaLedger.test.ts deleted file mode 100644 index 4f7d5c2a..00000000 --- a/lib/supabase/songstats_quota_ledger/__tests__/insertSongstatsQuotaLedger.test.ts +++ /dev/null @@ -1,32 +0,0 @@ -import { describe, it, expect, vi, beforeEach } from "vitest"; -import { insertSongstatsQuotaLedger } from "../insertSongstatsQuotaLedger"; -import supabase from "../../serverClient"; - -vi.mock("../../serverClient", () => { - const mockFrom = vi.fn(); - const mockRpc = vi.fn(); - return { default: { from: mockFrom, rpc: mockRpc } }; -}); - -describe("insertSongstatsQuotaLedger", () => { - beforeEach(() => vi.clearAllMocks()); - - it("records spent hits", async () => { - const insert = vi.fn().mockResolvedValue({ error: null }); - vi.mocked(supabase.from).mockReturnValue({ insert } as never); - - await insertSongstatsQuotaLedger({ hits: 1, purpose: "backfill USA2P2015959" }); - - expect(supabase.from).toHaveBeenCalledWith("songstats_quota_ledger"); - expect(insert).toHaveBeenCalledWith([{ hits: 1, purpose: "backfill USA2P2015959" }]); - }); - - it("throws on insert error (spend must be recorded)", async () => { - const insert = vi.fn().mockResolvedValue({ error: { message: "boom" } }); - vi.mocked(supabase.from).mockReturnValue({ insert } as never); - - await expect(insertSongstatsQuotaLedger({ hits: 1 })).rejects.toThrow( - "Failed to record songstats quota spend: boom", - ); - }); -}); diff --git a/lib/supabase/songstats_quota_ledger/__tests__/selectSongstatsQuotaSpent.test.ts b/lib/supabase/songstats_quota_ledger/__tests__/selectSongstatsQuotaSpent.test.ts deleted file mode 100644 index 6d59c626..00000000 --- a/lib/supabase/songstats_quota_ledger/__tests__/selectSongstatsQuotaSpent.test.ts +++ /dev/null @@ -1,35 +0,0 @@ -import { describe, it, expect, vi, beforeEach } from "vitest"; -import { selectSongstatsQuotaSpent } from "../selectSongstatsQuotaSpent"; -import supabase from "../../serverClient"; - -vi.mock("../../serverClient", () => { - const mockFrom = vi.fn(); - const mockRpc = vi.fn(); - return { default: { from: mockFrom, rpc: mockRpc } }; -}); - -describe("selectSongstatsQuotaSpent", () => { - beforeEach(() => vi.clearAllMocks()); - - it("sums hits spent since the window start", async () => { - const gte = vi.fn().mockResolvedValue({ data: [{ hits: 3 }, { hits: 7 }], error: null }); - const select = vi.fn().mockReturnValue({ gte }); - vi.mocked(supabase.from).mockReturnValue({ select } as never); - - const result = await selectSongstatsQuotaSpent("2026-05-12T00:00:00Z"); - - expect(supabase.from).toHaveBeenCalledWith("songstats_quota_ledger"); - expect(gte).toHaveBeenCalledWith("spent_at", "2026-05-12T00:00:00Z"); - expect(result).toBe(10); - }); - - it("treats errors as full quota spent (fail safe — do not overspend)", async () => { - const consoleError = vi.spyOn(console, "error").mockImplementation(() => {}); - const gte = vi.fn().mockResolvedValue({ data: null, error: { message: "boom" } }); - const select = vi.fn().mockReturnValue({ gte }); - vi.mocked(supabase.from).mockReturnValue({ select } as never); - - expect(await selectSongstatsQuotaSpent("2026-05-12T00:00:00Z")).toBe(Number.MAX_SAFE_INTEGER); - consoleError.mockRestore(); - }); -}); diff --git a/lib/supabase/songstats_quota_ledger/insertSongstatsQuotaLedger.ts b/lib/supabase/songstats_quota_ledger/insertSongstatsQuotaLedger.ts deleted file mode 100644 index 777c42a7..00000000 --- a/lib/supabase/songstats_quota_ledger/insertSongstatsQuotaLedger.ts +++ /dev/null @@ -1,19 +0,0 @@ -import supabase from "../serverClient"; -import { TablesInsert } from "@/types/database.types"; - -/** - * Record Songstats quota spend. Throws on failure — unrecorded spend would - * let the worker silently blow the rolling-window budget. - * - * @param entry - hits spent, optional purpose/account attribution - * @throws Error if the insert fails - */ -export async function insertSongstatsQuotaLedger( - entry: TablesInsert<"songstats_quota_ledger">, -): Promise { - const { error } = await supabase.from("songstats_quota_ledger").insert([entry]); - - if (error) { - throw new Error(`Failed to record songstats quota spend: ${error.message}`); - } -} diff --git a/lib/supabase/songstats_quota_ledger/selectSongstatsQuotaSpent.ts b/lib/supabase/songstats_quota_ledger/selectSongstatsQuotaSpent.ts deleted file mode 100644 index 0610a4b2..00000000 --- a/lib/supabase/songstats_quota_ledger/selectSongstatsQuotaSpent.ts +++ /dev/null @@ -1,23 +0,0 @@ -import supabase from "../serverClient"; - -/** - * Sum Songstats hits spent since a window start (the rolling 30-day quota - * check). Errors are treated as the full quota being spent — failing safe: - * the worker must never overspend because the ledger was unreadable. - * - * @param since - Inclusive ISO lower bound of the window - * @returns Total hits spent in the window - */ -export async function selectSongstatsQuotaSpent(since: string): Promise { - const { data, error } = await supabase - .from("songstats_quota_ledger") - .select("hits") - .gte("spent_at", since); - - if (error) { - console.error("Error reading songstats quota ledger:", error); - return Number.MAX_SAFE_INTEGER; - } - - return (data || []).reduce((sum, row) => sum + row.hits, 0); -}