diff --git a/lib/research/playcounts/__tests__/createSnapshot.test.ts b/lib/research/playcounts/__tests__/createSnapshot.test.ts index f3a0baed..867efb15 100644 --- a/lib/research/playcounts/__tests__/createSnapshot.test.ts +++ b/lib/research/playcounts/__tests__/createSnapshot.test.ts @@ -4,6 +4,7 @@ import { createSnapshot } from "../createSnapshot"; import { resolveSnapshotAlbums } from "../resolveSnapshotAlbums"; import { selectPlaycountSnapshots } from "@/lib/supabase/playcount_snapshots/selectPlaycountSnapshots"; import { insertPlaycountSnapshot } from "@/lib/supabase/playcount_snapshots/insertPlaycountSnapshot"; +import { deletePlaycountSnapshot } from "@/lib/supabase/playcount_snapshots/deletePlaycountSnapshot"; import { start } from "workflow/api"; vi.mock("../resolveSnapshotAlbums", () => ({ resolveSnapshotAlbums: vi.fn() })); @@ -13,6 +14,9 @@ vi.mock("@/lib/supabase/playcount_snapshots/selectPlaycountSnapshots", () => ({ vi.mock("@/lib/supabase/playcount_snapshots/insertPlaycountSnapshot", () => ({ insertPlaycountSnapshot: vi.fn(), })); +vi.mock("@/lib/supabase/playcount_snapshots/deletePlaycountSnapshot", () => ({ + deletePlaycountSnapshot: vi.fn(), +})); vi.mock("workflow/api", () => ({ start: vi.fn() })); vi.mock("@/app/workflows/playcountSnapshotWorkflow", () => ({ playcountSnapshotWorkflow: vi.fn(), @@ -180,4 +184,172 @@ describe("createSnapshot", () => { }); vi.useRealTimers(); }); + + // chat#1912 row 7: two simultaneous identical requests both insert before + // either can see the other. On re-read the loser must withdraw its claim and + // hand back the winner's, so exactly one capture is scraped and one row + // survives. + it("withdraws its own claim when a concurrent request won the race", async () => { + vi.mocked(resolveSnapshotAlbums).mockResolvedValue(["a1"]); + vi.mocked(insertPlaycountSnapshot).mockResolvedValue({ id: "mine" } as never); + const scope = { + album_ids: ["a1"], + platforms: ["spotify"], + schedule: "once", + state: "queued", + album_count: 1, + estimated_cost_usd: 0.003, + }; + vi.mocked(selectPlaycountSnapshots) + // pre-check: nobody has claimed it yet + .mockResolvedValueOnce([] as never) + // reconcile: the other request's row landed first + .mockResolvedValueOnce([ + { ...scope, id: "theirs", created_at: "2026-07-30T12:00:00.000Z" }, + { ...scope, id: "mine", created_at: "2026-07-30T12:00:01.000Z" }, + ] as never); + + const result = await createSnapshot({ + accountId: "acc_1", + body: { album_ids: ["a1"], platforms: ["spotify"], schedule: "once" }, + }); + + expect(deletePlaycountSnapshot).toHaveBeenCalledWith("mine"); + expect(start).not.toHaveBeenCalled(); + expect(result).toEqual({ + data: { + status: "success", + snapshot_id: "theirs", + state: "queued", + album_count: 1, + estimated_cost_usd: 0, + reused: true, + }, + }); + }); + + it("keeps its claim and scrapes when it won the race", async () => { + vi.mocked(resolveSnapshotAlbums).mockResolvedValue(["a1"]); + vi.mocked(insertPlaycountSnapshot).mockResolvedValue({ id: "mine" } as never); + const scope = { + album_ids: ["a1"], + platforms: ["spotify"], + schedule: "once", + state: "queued", + album_count: 1, + estimated_cost_usd: 0.003, + }; + vi.mocked(selectPlaycountSnapshots) + .mockResolvedValueOnce([] as never) + .mockResolvedValueOnce([ + { ...scope, id: "mine", created_at: "2026-07-30T12:00:00.000Z" }, + { ...scope, id: "theirs", created_at: "2026-07-30T12:00:01.000Z" }, + ] as never); + + const result = await createSnapshot({ + accountId: "acc_1", + body: { album_ids: ["a1"], platforms: ["spotify"], schedule: "once" }, + }); + + expect(deletePlaycountSnapshot).not.toHaveBeenCalled(); + expect(start).toHaveBeenCalledWith(expect.anything(), ["mine"]); + expect((result as { data: { snapshot_id: string } }).data.snapshot_id).toBe("mine"); + }); + + // Review finding (coderabbit + cubic P1, 2026-07-30). The reconcile used to + // count only `queued` claims, but findReusableSnapshot treats queued, + // running and done as reusable. A winner that advanced to `running` between + // the pre-check and the re-read was therefore invisible, and both requests + // scraped — the exact race this row exists to close. + it("defers to a winner that has already started running", async () => { + vi.mocked(resolveSnapshotAlbums).mockResolvedValue(["a1"]); + vi.mocked(insertPlaycountSnapshot).mockResolvedValue({ id: "mine" } as never); + const scope = { + album_ids: ["a1"], + platforms: ["spotify"], + schedule: "once", + album_count: 1, + estimated_cost_usd: 0.003, + }; + vi.mocked(selectPlaycountSnapshots) + .mockResolvedValueOnce([] as never) + .mockResolvedValueOnce([ + { ...scope, id: "theirs", state: "running", created_at: "2026-07-30T12:00:00.000Z" }, + { ...scope, id: "mine", state: "queued", created_at: "2026-07-30T12:00:01.000Z" }, + ] as never); + + const result = await createSnapshot({ + accountId: "acc_1", + body: { album_ids: ["a1"], platforms: ["spotify"], schedule: "once" }, + }); + + expect(deletePlaycountSnapshot).toHaveBeenCalledWith("mine"); + expect(start).not.toHaveBeenCalled(); + expect((result as { data: { snapshot_id: string } }).data.snapshot_id).toBe("theirs"); + }); + + // A failed capture is not a claim: nothing can be handed back from it, so a + // new request must go ahead and scrape. + it("does not defer to a failed claim", async () => { + vi.mocked(resolveSnapshotAlbums).mockResolvedValue(["a1"]); + vi.mocked(insertPlaycountSnapshot).mockResolvedValue({ id: "mine" } as never); + const scope = { + album_ids: ["a1"], + platforms: ["spotify"], + schedule: "once", + album_count: 1, + estimated_cost_usd: 0.003, + }; + vi.mocked(selectPlaycountSnapshots) + .mockResolvedValueOnce([] as never) + .mockResolvedValueOnce([ + { ...scope, id: "failed", state: "failed", created_at: "2026-07-30T12:00:00.000Z" }, + { ...scope, id: "mine", state: "queued", created_at: "2026-07-30T12:00:01.000Z" }, + ] as never); + + await createSnapshot({ + accountId: "acc_1", + body: { album_ids: ["a1"], platforms: ["spotify"], schedule: "once" }, + }); + + expect(deletePlaycountSnapshot).not.toHaveBeenCalled(); + expect(start).toHaveBeenCalledWith(expect.anything(), ["mine"]); + }); + + // Preview verification of 8 concurrent identical requests: exactly one row + // and one scrape survived (the point of this row), but four callers were + // handed the id of a mid-ranked claim that then withdrew itself, leaving + // them polling a snapshot that no longer exists. After withdrawing, re-read + // and hand back a claim that is still standing. + it("never hands back a claim that has itself withdrawn", async () => { + vi.mocked(resolveSnapshotAlbums).mockResolvedValue(["a1"]); + vi.mocked(insertPlaycountSnapshot).mockResolvedValue({ id: "mine" } as never); + const scope = { + album_ids: ["a1"], + platforms: ["spotify"], + schedule: "once", + album_count: 1, + estimated_cost_usd: 0.003, + state: "queued", + }; + vi.mocked(selectPlaycountSnapshots) + .mockResolvedValueOnce([] as never) + // reconcile: a mid-ranked claim looks canonical from here + .mockResolvedValueOnce([ + { ...scope, id: "midranked", created_at: "2026-07-30T12:00:01.000Z" }, + { ...scope, id: "mine", created_at: "2026-07-30T12:00:02.000Z" }, + ] as never) + // after withdrawing: midranked has withdrawn too, the true earliest stands + .mockResolvedValueOnce([ + { ...scope, id: "earliest", created_at: "2026-07-30T12:00:00.000Z" }, + ] as never); + + const result = await createSnapshot({ + accountId: "acc_1", + body: { album_ids: ["a1"], platforms: ["spotify"], schedule: "once" }, + }); + + expect(deletePlaycountSnapshot).toHaveBeenCalledWith("mine"); + expect((result as { data: { snapshot_id: string } }).data.snapshot_id).toBe("earliest"); + }); }); diff --git a/lib/research/playcounts/__tests__/pickCanonicalSnapshot.test.ts b/lib/research/playcounts/__tests__/pickCanonicalSnapshot.test.ts new file mode 100644 index 00000000..734dd004 --- /dev/null +++ b/lib/research/playcounts/__tests__/pickCanonicalSnapshot.test.ts @@ -0,0 +1,48 @@ +import { describe, it, expect } from "vitest"; +import { pickCanonicalSnapshot } from "@/lib/research/playcounts/pickCanonicalSnapshot"; +import type { Tables } from "@/types/database.types"; + +const row = (over: Partial>) => + ({ + id: "b", + created_at: "2026-07-30T12:00:00.000Z", + ...over, + }) as Tables<"playcount_snapshots">; + +describe("pickCanonicalSnapshot", () => { + // chat#1912 row 7. Two simultaneous identical requests both insert before + // either can see the other, so the pre-check cannot help. Both then re-read + // and must independently agree on the same winner, or they both scrape. + it("picks the earliest claim", () => { + const winner = row({ id: "late", created_at: "2026-07-30T12:00:05.000Z" }); + const early = row({ id: "early", created_at: "2026-07-30T12:00:01.000Z" }); + + expect(pickCanonicalSnapshot([winner, early])?.id).toBe("early"); + }); + + // Same-millisecond inserts are the whole point of this function, so the + // tie-break must be total and identical for every caller. + it("breaks an exact timestamp tie by id, deterministically", () => { + const a = row({ id: "aaa", created_at: "2026-07-30T12:00:00.000Z" }); + const b = row({ id: "bbb", created_at: "2026-07-30T12:00:00.000Z" }); + + expect(pickCanonicalSnapshot([a, b])?.id).toBe("aaa"); + // Order of the input must not change the answer, or the two racers disagree. + expect(pickCanonicalSnapshot([b, a])?.id).toBe("aaa"); + }); + + it("returns the only claim when there is no race", () => { + expect(pickCanonicalSnapshot([row({ id: "solo" })])?.id).toBe("solo"); + }); + + it("returns null for no claims", () => { + expect(pickCanonicalSnapshot([])).toBeNull(); + }); + + it("ignores claims with no timestamp rather than crowning them", () => { + const dated = row({ id: "dated", created_at: "2026-07-30T12:00:09.000Z" }); + const undatedRow = row({ id: "aaa", created_at: null }); + + expect(pickCanonicalSnapshot([undatedRow, dated])?.id).toBe("dated"); + }); +}); diff --git a/lib/research/playcounts/createSnapshot.ts b/lib/research/playcounts/createSnapshot.ts index 53cb4a1d..f330e94f 100644 --- a/lib/research/playcounts/createSnapshot.ts +++ b/lib/research/playcounts/createSnapshot.ts @@ -5,6 +5,10 @@ import { insertPlaycountSnapshot } from "@/lib/supabase/playcount_snapshots/inse import { playcountSnapshotWorkflow } from "@/app/workflows/playcountSnapshotWorkflow"; import { findReusableSnapshot } from "@/lib/research/playcounts/findReusableSnapshot"; import { buildReusedSnapshotResult } from "@/lib/research/playcounts/buildReusedSnapshotResult"; +import { pickCanonicalSnapshot } from "@/lib/research/playcounts/pickCanonicalSnapshot"; +import { sameScope } from "@/lib/research/playcounts/sameScope"; +import { isReusableSnapshotState } from "@/lib/research/playcounts/isReusableSnapshotState"; +import { deletePlaycountSnapshot } from "@/lib/supabase/playcount_snapshots/deletePlaycountSnapshot"; import { getMonthlySpendUsd } from "@/lib/research/playcounts/getMonthlySpendUsd"; import { CreateSnapshotBody } from "@/lib/research/playcounts/validateCreateSnapshotRequest"; @@ -87,6 +91,43 @@ export async function createSnapshot(params: { estimated_cost_usd: estimatedCostUsd, }); + // The insert is a claim, not yet a scrape. Two simultaneous identical + // requests both get here before either can see the other, so re-read and + // let the earliest claim win — the loser withdraws its row and hands back + // the winner's rather than starting a second capture (chat#1912 row 7). + const claims = await selectPlaycountSnapshots({ + account: params.accountId, + createdAfter: reuseCutoff.toISOString(), + }); + const canonical = pickCanonicalSnapshot( + claims.filter( + candidate => + candidate.id === row.id || + (sameScope(candidate, albumIds, params.body.platforms, params.body.schedule) && + isReusableSnapshotState(candidate.state)), + ), + ); + if (canonical && canonical.id !== row.id) { + await deletePlaycountSnapshot(row.id); + + // The claim that looked canonical from here may itself have been a loser + // that withdrew, which would hand this caller a snapshot id that no longer + // exists. Re-read after withdrawing so the id returned is one still + // standing (observed with 8 concurrent requests during verification). + const remaining = await selectPlaycountSnapshots({ + account: params.accountId, + createdAfter: reuseCutoff.toISOString(), + }); + const survivor = pickCanonicalSnapshot( + remaining.filter( + candidate => + sameScope(candidate, albumIds, params.body.platforms, params.body.schedule) && + isReusableSnapshotState(candidate.state), + ), + ); + return buildReusedSnapshotResult(survivor ?? canonical); + } + await start(playcountSnapshotWorkflow, [row.id]); return { diff --git a/lib/research/playcounts/findReusableSnapshot.ts b/lib/research/playcounts/findReusableSnapshot.ts index 8cd8ee4b..1aa7b9f8 100644 Binary files a/lib/research/playcounts/findReusableSnapshot.ts and b/lib/research/playcounts/findReusableSnapshot.ts differ diff --git a/lib/research/playcounts/isReusableSnapshotState.ts b/lib/research/playcounts/isReusableSnapshotState.ts new file mode 100644 index 00000000..57819f43 --- /dev/null +++ b/lib/research/playcounts/isReusableSnapshotState.ts @@ -0,0 +1,14 @@ +/** A capture that failed has nothing to hand back; every other live state does. */ +const REUSABLE_STATES = new Set(["queued", "running", "done"]); + +/** + * Whether a capture in this state can satisfy a new request for the same scope. + * + * Shared by the reuse pre-check and the post-insert race reconcile. They must + * agree: if the reconcile recognised only `queued`, a winner that advanced to + * `running` between the two reads would be invisible and both requests would + * scrape — the exact race this is meant to close (chat#1912 rows 4 and 7). + */ +export function isReusableSnapshotState(state: string | null): boolean { + return REUSABLE_STATES.has(state ?? ""); +} diff --git a/lib/research/playcounts/pickCanonicalSnapshot.ts b/lib/research/playcounts/pickCanonicalSnapshot.ts new file mode 100644 index 00000000..22d5e4c1 --- /dev/null +++ b/lib/research/playcounts/pickCanonicalSnapshot.ts @@ -0,0 +1,26 @@ +import type { Tables } from "@/types/database.types"; + +/** + * Chooses which of several competing claims on the same scope is the real one. + * + * Two simultaneous identical requests both insert before either can see the + * other, so the reuse pre-check cannot help (chat#1912 row 7). Both then re-read + * and call this, and because the ordering is total and independent of input + * order they reach the same answer without coordinating: earliest claim wins, + * ties broken by id. + * + * Rows with no timestamp cannot be ordered and are never crowned. + */ +export function pickCanonicalSnapshot( + claims: Tables<"playcount_snapshots">[], +): Tables<"playcount_snapshots"> | null { + const dated = claims.filter(row => !!row.created_at); + if (dated.length === 0) return null; + + return dated.reduce((winner, row) => { + const a = new Date(row.created_at!).getTime(); + const b = new Date(winner.created_at!).getTime(); + if (a !== b) return a < b ? row : winner; + return row.id < winner.id ? row : winner; + }); +} diff --git a/lib/research/playcounts/sameScope.ts b/lib/research/playcounts/sameScope.ts new file mode 100644 index 00000000..220549fc --- /dev/null +++ b/lib/research/playcounts/sameScope.ts @@ -0,0 +1,22 @@ +import type { Tables } from "@/types/database.types"; + +const sameSet = (a: string[], b: string[]) => + a.length === b.length && [...a].sort().join(" ") === [...b].sort().join(" "); + +/** + * Whether a snapshot row asks the same question as a request: same album set + * (order-insensitive), same platforms, same schedule. + * + * Shared by the reuse pre-check and the post-insert race reconcile so both + * agree on what "identical" means (chat#1912 rows 4 and 7). + */ +export function sameScope( + row: Tables<"playcount_snapshots">, + albumIds: string[], + platforms: string[], + schedule: string, +): boolean { + if (row.schedule !== schedule) return false; + if (!sameSet(row.album_ids ?? [], albumIds)) return false; + return sameSet(row.platforms ?? [], platforms); +} diff --git a/lib/supabase/playcount_snapshots/deletePlaycountSnapshot.ts b/lib/supabase/playcount_snapshots/deletePlaycountSnapshot.ts new file mode 100644 index 00000000..fb75017f --- /dev/null +++ b/lib/supabase/playcount_snapshots/deletePlaycountSnapshot.ts @@ -0,0 +1,19 @@ +import supabase from "../serverClient"; + +/** + * Delete a snapshot job row. + * + * Used to withdraw a losing claim when two simultaneous identical requests + * both inserted before either could see the other, so one capture is left + * rather than two (chat#1912 row 7). + * + * @param id - The snapshot id + * @throws Error if the delete fails + */ +export async function deletePlaycountSnapshot(id: string): Promise { + const { error } = await supabase.from("playcount_snapshots").delete().eq("id", id); + + if (error) { + throw new Error(`Failed to delete playcount snapshot: ${error.message}`); + } +}