fix: single-flight Spotify token + 401-refresh-retry to survive cold-cache fan-out (closes task/spotify-token-resilience)
Source: task/spotify-token-resilience (plan/steward-linker-quality) — cold-cache fan-out fired up to 100 concurrent token POSTs (429 storm) and search didn't refresh-retry on a 401. - service/spotify.js (new): makeSpotifyToken() single-flights the token fetch via an in-flight-promise cache so N concurrent cold/expired lookups share ONE POST /api/token; clears the promise on settle (success or failure) so a rejected fetch never poisons the cache. makeSpotifySearch() refreshes-and-retries once on a 401 and honors a bounded (capped) Retry-After on a 429, else throws (resolve-core nulls the candidate, as before). - service/resolver.js: wires the factories with the real fetch + Date.now; value-cache + (expires_in-60)*1000 expiry math unchanged; external behavior (result shapes, USE_SPOTIFY degradation, /health, /resolve) identical. - service/spotify.test.js (new): hermetic node:test coverage — single-flight (100 concurrent -> 1 fetch), re-fetch after expiry, failure clears in-flight, 401 refresh-retry + persistent-401 throw, bounded 429 backoff. Zero new deps. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
+31
-22
@@ -18,6 +18,7 @@ import { readFileSync } from "node:fs";
|
|||||||
import { pickArtist, pickAlbum } from "./lib.js";
|
import { pickArtist, pickAlbum } from "./lib.js";
|
||||||
import { buildClassifyMessages, parseClassification, DEFAULT_MODEL } from "./llm.js";
|
import { buildClassifyMessages, parseClassification, DEFAULT_MODEL } from "./llm.js";
|
||||||
import { resolveAll as resolveAllCore } from "./resolve-core.js";
|
import { resolveAll as resolveAllCore } from "./resolve-core.js";
|
||||||
|
import { makeSpotifyToken, makeSpotifySearch } from "./spotify.js";
|
||||||
|
|
||||||
// --- minimal .env loader (no dependency) ------------------------------------
|
// --- minimal .env loader (no dependency) ------------------------------------
|
||||||
try {
|
try {
|
||||||
@@ -42,38 +43,46 @@ const USE_LLM = Boolean(OLLAMA_URL);
|
|||||||
const spCache = new Map();
|
const spCache = new Map();
|
||||||
|
|
||||||
// --- Spotify (optional): direct-artist-link lookup --------------------------
|
// --- Spotify (optional): direct-artist-link lookup --------------------------
|
||||||
let spToken = null;
|
// The single-flight token getter: a cold/expired-cache fan-out (up to MAX_CANDIDATES concurrent
|
||||||
let spTokenExp = 0;
|
// resolves) shares ONE POST /api/token instead of stampeding the rate-limited token endpoint.
|
||||||
async function spotifyToken() {
|
// fetchToken does the real POST; the cache + (expires_in-60)*1000 expiry math live in the factory.
|
||||||
if (spToken && Date.now() < spTokenExp) return spToken;
|
const spTokens = makeSpotifyToken({
|
||||||
const auth = Buffer.from(`${SPOTIFY_ID}:${SPOTIFY_SECRET}`).toString("base64");
|
fetchToken: async () => {
|
||||||
const r = await fetch("https://accounts.spotify.com/api/token", {
|
const auth = Buffer.from(`${SPOTIFY_ID}:${SPOTIFY_SECRET}`).toString("base64");
|
||||||
method: "POST",
|
const r = await fetch("https://accounts.spotify.com/api/token", {
|
||||||
headers: { authorization: `Basic ${auth}`, "content-type": "application/x-www-form-urlencoded" },
|
method: "POST",
|
||||||
body: "grant_type=client_credentials",
|
headers: { authorization: `Basic ${auth}`, "content-type": "application/x-www-form-urlencoded" },
|
||||||
|
body: "grant_type=client_credentials",
|
||||||
|
signal: AbortSignal.timeout(5000), // fail fast on a hung upstream (caught per-candidate -> null)
|
||||||
|
});
|
||||||
|
if (!r.ok) throw new Error(`spotify token http ${r.status}`);
|
||||||
|
return r.json();
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
// One bare Spotify search (no token mgmt): returns { ok, status, headers, body } so the
|
||||||
|
// search-with-retry layer can branch on 401/429 without an exception masking the status.
|
||||||
|
async function doSpotifySearch(name, token) {
|
||||||
|
const url = `https://api.spotify.com/v1/search?type=artist,album&limit=5&q=${encodeURIComponent(name)}`;
|
||||||
|
const r = await fetch(url, {
|
||||||
|
headers: { authorization: `Bearer ${token}` },
|
||||||
signal: AbortSignal.timeout(5000), // fail fast on a hung upstream (caught per-candidate -> null)
|
signal: AbortSignal.timeout(5000), // fail fast on a hung upstream (caught per-candidate -> null)
|
||||||
});
|
});
|
||||||
if (!r.ok) throw new Error(`spotify token http ${r.status}`);
|
// 401/429 carry no usable body for our purposes; only parse JSON on a real hit.
|
||||||
const j = await r.json();
|
const body = r.ok ? await r.json() : null;
|
||||||
spToken = j.access_token;
|
return { ok: r.ok, status: r.status, headers: r.headers, body };
|
||||||
spTokenExp = Date.now() + (j.expires_in - 60) * 1000;
|
|
||||||
return spToken;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 401 -> invalidate + refresh + retry once; 429 -> bounded Retry-After + retry once; else throw.
|
||||||
|
const spSearch = makeSpotifySearch({ doSearch: doSpotifySearch, tokens: spTokens });
|
||||||
|
|
||||||
// The injected `search` for resolve-core: one type=artist,album Spotify search per candidate,
|
// The injected `search` for resolve-core: one type=artist,album Spotify search per candidate,
|
||||||
// returning the exact-name-match hits as { artist, album } where each is the external Spotify url
|
// returning the exact-name-match hits as { artist, album } where each is the external Spotify url
|
||||||
// (string) or null. No creds -> both null (nothing links). resolve-core's pickDirect applies the
|
// (string) or null. No creds -> both null (nothing links). resolve-core's pickDirect applies the
|
||||||
// kindHint tie-break and shapes the link; the cache (spCache) is owned by resolve-core too.
|
// kindHint tie-break and shapes the link; the cache (spCache) is owned by resolve-core too.
|
||||||
async function search(name) {
|
async function search(name) {
|
||||||
if (!USE_SPOTIFY) return { artist: null, album: null };
|
if (!USE_SPOTIFY) return { artist: null, album: null };
|
||||||
const token = await spotifyToken();
|
const j = await spSearch(name);
|
||||||
const url = `https://api.spotify.com/v1/search?type=artist,album&limit=5&q=${encodeURIComponent(name)}`;
|
|
||||||
const r = await fetch(url, {
|
|
||||||
headers: { authorization: `Bearer ${token}` },
|
|
||||||
signal: AbortSignal.timeout(5000), // fail fast on a hung upstream (caught per-candidate -> null)
|
|
||||||
});
|
|
||||||
if (!r.ok) throw new Error(`spotify search http ${r.status}`);
|
|
||||||
const j = await r.json();
|
|
||||||
return {
|
return {
|
||||||
artist: pickArtist(name, j.artists?.items)?.external_urls?.spotify ?? null,
|
artist: pickArtist(name, j.artists?.items)?.external_urls?.spotify ?? null,
|
||||||
album: pickAlbum(name, j.albums?.items)?.external_urls?.spotify ?? null,
|
album: pickAlbum(name, j.albums?.items)?.external_urls?.spotify ?? null,
|
||||||
|
|||||||
@@ -0,0 +1,79 @@
|
|||||||
|
// Testable Spotify client-credentials token + search helpers, decoupled from resolver.js's HTTP
|
||||||
|
// side effects so `node --test` can drive them with injected fakes (a counting fetch + a fake
|
||||||
|
// clock). The single-flight token getter coalesces a cold-cache fan-out — up to MAX_CANDIDATES
|
||||||
|
// concurrent resolves share ONE POST /api/token instead of stampeding the rate-limited endpoint.
|
||||||
|
|
||||||
|
// Single-flight token getter factory. `fetchToken()` performs the real POST /api/token and
|
||||||
|
// returns { access_token, expires_in } (resolver.js wires the live fetch); `now` is the clock
|
||||||
|
// (Date.now, injectable for tests). Returns { getToken, invalidate }:
|
||||||
|
// getToken() -> the cached access_token while it is fresh; on a cold/expired cache, the FIRST
|
||||||
|
// caller starts the fetch and every concurrent caller awaits the SAME in-flight promise, so
|
||||||
|
// N concurrent cold lookups make exactly one fetch. The in-flight promise is cleared once it
|
||||||
|
// settles (success OR failure) so the next cold window re-fetches and a rejected fetch never
|
||||||
|
// poisons the cache.
|
||||||
|
// invalidate() -> drop the cached value so the next getToken re-fetches (used by the 401 path).
|
||||||
|
// Expiry math is (expires_in - 60)*1000 ms — a 60s safety margin before Spotify's stated expiry.
|
||||||
|
export function makeSpotifyToken({ fetchToken, now = Date.now }) {
|
||||||
|
let token = null;
|
||||||
|
let exp = 0;
|
||||||
|
let inflight = null;
|
||||||
|
|
||||||
|
async function getToken() {
|
||||||
|
if (token && now() < exp) return token; // warm hit — value cache, no promise
|
||||||
|
if (inflight) return inflight; // a cold fetch is already running — coalesce onto it
|
||||||
|
inflight = (async () => {
|
||||||
|
const j = await fetchToken();
|
||||||
|
token = j.access_token;
|
||||||
|
exp = now() + (j.expires_in - 60) * 1000;
|
||||||
|
return token;
|
||||||
|
})().finally(() => { inflight = null; }); // clear on success OR failure
|
||||||
|
return inflight;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Force the next getToken to re-fetch (the cached value is rejected mid-window, e.g. a 401).
|
||||||
|
function invalidate() {
|
||||||
|
token = null;
|
||||||
|
exp = 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
return { getToken, invalidate };
|
||||||
|
}
|
||||||
|
|
||||||
|
// Search-with-retry factory. `doSearch(name, token)` performs ONE Spotify search and returns
|
||||||
|
// { ok, status, body } — never throws for an HTTP error (the caller decides). `tokens` is a
|
||||||
|
// makeSpotifyToken() handle; `sleep(ms)` is the backoff delay (injectable, zero in tests).
|
||||||
|
// Returns search(name) -> the parsed JSON body, or throws so resolve-core's per-candidate catch
|
||||||
|
// nulls that candidate (today's graceful degradation). Behavior:
|
||||||
|
// * 401 -> invalidate the token, fetch a fresh one, retry ONCE; a persistent 401 throws.
|
||||||
|
// * 429 -> honor a bounded Retry-After (capped) once, then retry ONCE; a persistent 429 throws.
|
||||||
|
// * other !ok -> throw (caller nulls the candidate, as today).
|
||||||
|
// Retries are bounded to one — never unbounded.
|
||||||
|
export function makeSpotifySearch({ doSearch, tokens, sleep = (ms) => new Promise((r) => setTimeout(r, ms)), maxRetryAfterMs = 2000 }) {
|
||||||
|
return async function search(name) {
|
||||||
|
let token = await tokens.getToken();
|
||||||
|
let r = await doSearch(name, token);
|
||||||
|
|
||||||
|
if (r.status === 401) {
|
||||||
|
tokens.invalidate(); // the token was rejected mid-window — force a fresh one
|
||||||
|
token = await tokens.getToken();
|
||||||
|
r = await doSearch(name, token);
|
||||||
|
} else if (r.status === 429) {
|
||||||
|
const wait = retryAfterMs(r.headers, maxRetryAfterMs);
|
||||||
|
if (wait != null) await sleep(wait);
|
||||||
|
r = await doSearch(name, token);
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!r.ok) throw new Error(`spotify search http ${r.status}`);
|
||||||
|
return r.body;
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
// Parse a bounded Retry-After (seconds) from response headers into ms, capped at `maxMs`.
|
||||||
|
// `headers` is a fetch Headers (or any object with a .get). Returns null when absent/garbage.
|
||||||
|
export function retryAfterMs(headers, maxMs) {
|
||||||
|
const raw = headers?.get?.("retry-after");
|
||||||
|
if (raw == null || raw === "") return null; // absent -> no wait (Number(null/"") is 0, guard it)
|
||||||
|
const secs = Number(raw);
|
||||||
|
if (!Number.isFinite(secs) || secs < 0) return null;
|
||||||
|
return Math.min(secs * 1000, maxMs);
|
||||||
|
}
|
||||||
@@ -0,0 +1,177 @@
|
|||||||
|
// Unit tests for the Spotify token single-flight + search-with-retry helpers (run with
|
||||||
|
// `node --test`). Zero-dep, hermetic: a counting fake fetchToken/doSearch + an injected clock,
|
||||||
|
// no network and no real timers (sleep is injected as a zero-delay stub).
|
||||||
|
import { test } from "node:test";
|
||||||
|
import assert from "node:assert/strict";
|
||||||
|
import { makeSpotifyToken, makeSpotifySearch, retryAfterMs } from "./spotify.js";
|
||||||
|
|
||||||
|
// A controllable clock: starts at 0, advance() bumps it.
|
||||||
|
function fakeClock(start = 0) {
|
||||||
|
let t = start;
|
||||||
|
return { now: () => t, advance: (ms) => { t += ms; } };
|
||||||
|
}
|
||||||
|
|
||||||
|
// A fetchToken that counts calls and yields a token + expires_in; optionally defers via a gate
|
||||||
|
// so concurrent callers actually overlap on the in-flight promise before it settles.
|
||||||
|
function countingFetchToken({ expiresIn = 3600, token = "tok", gate = null } = {}) {
|
||||||
|
const state = { calls: 0 };
|
||||||
|
const fetchToken = async () => {
|
||||||
|
state.calls += 1;
|
||||||
|
const n = state.calls;
|
||||||
|
if (gate) await gate; // hold all in-flight calls open until the test releases the gate
|
||||||
|
return { access_token: `${token}-${n}`, expires_in: expiresIn };
|
||||||
|
};
|
||||||
|
return { fetchToken, state };
|
||||||
|
}
|
||||||
|
|
||||||
|
test("single-flight: N concurrent cold getToken share exactly one fetch", async () => {
|
||||||
|
let release;
|
||||||
|
const gate = new Promise((r) => { release = r; });
|
||||||
|
const { fetchToken, state } = countingFetchToken({ gate });
|
||||||
|
const clock = fakeClock();
|
||||||
|
const { getToken } = makeSpotifyToken({ fetchToken, now: clock.now });
|
||||||
|
|
||||||
|
// Fire 100 concurrent cold lookups; they all enter while the fetch is gated open.
|
||||||
|
const pending = Array.from({ length: 100 }, () => getToken());
|
||||||
|
release(); // let the single in-flight fetch resolve
|
||||||
|
const tokens = await Promise.all(pending);
|
||||||
|
|
||||||
|
assert.equal(state.calls, 1, "exactly one POST /api/token for 100 concurrent cold callers");
|
||||||
|
assert.ok(tokens.every((t) => t === "tok-1"), "every caller got the same token");
|
||||||
|
});
|
||||||
|
|
||||||
|
test("warm hit serves the cached value without any fetch", async () => {
|
||||||
|
const { fetchToken, state } = countingFetchToken();
|
||||||
|
const clock = fakeClock();
|
||||||
|
const { getToken } = makeSpotifyToken({ fetchToken, now: clock.now });
|
||||||
|
|
||||||
|
const a = await getToken();
|
||||||
|
const b = await getToken(); // still fresh -> no new fetch
|
||||||
|
assert.equal(a, "tok-1");
|
||||||
|
assert.equal(b, "tok-1");
|
||||||
|
assert.equal(state.calls, 1);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("re-fetch after expiry: advancing the clock past exp triggers exactly one more fetch", async () => {
|
||||||
|
const { fetchToken, state } = countingFetchToken({ expiresIn: 3600 });
|
||||||
|
const clock = fakeClock();
|
||||||
|
const { getToken } = makeSpotifyToken({ fetchToken, now: clock.now });
|
||||||
|
|
||||||
|
await getToken();
|
||||||
|
assert.equal(state.calls, 1);
|
||||||
|
|
||||||
|
// exp = now + (3600-60)*1000 = 3_540_000 ms. Advance just past it.
|
||||||
|
clock.advance(3_540_001);
|
||||||
|
const t2 = await getToken();
|
||||||
|
assert.equal(state.calls, 2, "one more fetch after expiry");
|
||||||
|
assert.equal(t2, "tok-2", "the fresh token is served");
|
||||||
|
});
|
||||||
|
|
||||||
|
test("failure clears the in-flight promise: a rejected fetch does not poison the cache", async () => {
|
||||||
|
let failNext = true;
|
||||||
|
const state = { calls: 0 };
|
||||||
|
const fetchToken = async () => {
|
||||||
|
state.calls += 1;
|
||||||
|
if (failNext) { failNext = false; throw new Error("token http 503"); }
|
||||||
|
return { access_token: "tok-ok", expires_in: 3600 };
|
||||||
|
};
|
||||||
|
const clock = fakeClock();
|
||||||
|
const { getToken } = makeSpotifyToken({ fetchToken, now: clock.now });
|
||||||
|
|
||||||
|
await assert.rejects(getToken(), /token http 503/);
|
||||||
|
// The poisoned promise must be cleared -> the next call retries and succeeds.
|
||||||
|
const t = await getToken();
|
||||||
|
assert.equal(t, "tok-ok");
|
||||||
|
assert.equal(state.calls, 2, "the rejected fetch did not stick — a fresh fetch ran");
|
||||||
|
});
|
||||||
|
|
||||||
|
// --- search-with-retry --------------------------------------------------------
|
||||||
|
|
||||||
|
// A doSearch that returns a scripted sequence of { ok, status, headers?, body? } responses,
|
||||||
|
// counting calls and recording which token each call saw.
|
||||||
|
function scriptedSearch(responses) {
|
||||||
|
const state = { calls: 0, tokensSeen: [] };
|
||||||
|
const doSearch = async (_name, token) => {
|
||||||
|
state.tokensSeen.push(token);
|
||||||
|
const r = responses[state.calls] ?? responses[responses.length - 1];
|
||||||
|
state.calls += 1;
|
||||||
|
return r;
|
||||||
|
};
|
||||||
|
return { doSearch, state };
|
||||||
|
}
|
||||||
|
|
||||||
|
const okResp = (body) => ({ ok: true, status: 200, body });
|
||||||
|
const errResp = (status, headers) => ({ ok: false, status, headers, body: null });
|
||||||
|
const hdr = (obj) => ({ get: (k) => obj[k.toLowerCase()] ?? null });
|
||||||
|
|
||||||
|
test("401 retry: a stale-token 401 invalidates, refreshes, and retries once with a fresh token", async () => {
|
||||||
|
const { fetchToken, state: tokState } = countingFetchToken();
|
||||||
|
const tokens = makeSpotifyToken({ fetchToken, now: fakeClock().now });
|
||||||
|
const { doSearch, state } = scriptedSearch([errResp(401), okResp({ hit: true })]);
|
||||||
|
|
||||||
|
const search = makeSpotifySearch({ doSearch, tokens, sleep: async () => {} });
|
||||||
|
const body = await search("Tool");
|
||||||
|
|
||||||
|
assert.deepEqual(body, { hit: true });
|
||||||
|
assert.equal(state.calls, 2, "exactly one retry after the 401");
|
||||||
|
assert.equal(tokState.calls, 2, "the token was invalidated and re-fetched");
|
||||||
|
assert.equal(state.tokensSeen[0], "tok-1");
|
||||||
|
assert.equal(state.tokensSeen[1], "tok-2", "the retry used the FRESH token");
|
||||||
|
});
|
||||||
|
|
||||||
|
test("persistent 401: a second 401 throws (no infinite retry), so resolve-core nulls the candidate", async () => {
|
||||||
|
const { fetchToken } = countingFetchToken();
|
||||||
|
const tokens = makeSpotifyToken({ fetchToken, now: fakeClock().now });
|
||||||
|
const { doSearch, state } = scriptedSearch([errResp(401), errResp(401)]);
|
||||||
|
|
||||||
|
const search = makeSpotifySearch({ doSearch, tokens, sleep: async () => {} });
|
||||||
|
await assert.rejects(search("Tool"), /spotify search http 401/);
|
||||||
|
assert.equal(state.calls, 2, "retried exactly once, then gave up");
|
||||||
|
});
|
||||||
|
|
||||||
|
test("429 backoff: a Retry-After 429 waits (bounded) then retries once and succeeds", async () => {
|
||||||
|
const { fetchToken } = countingFetchToken();
|
||||||
|
const tokens = makeSpotifyToken({ fetchToken, now: fakeClock().now });
|
||||||
|
const { doSearch, state } = scriptedSearch([errResp(429, hdr({ "retry-after": "1" })), okResp({ hit: true })]);
|
||||||
|
|
||||||
|
let slept = null;
|
||||||
|
const search = makeSpotifySearch({ doSearch, tokens, sleep: async (ms) => { slept = ms; } });
|
||||||
|
const body = await search("Tool");
|
||||||
|
|
||||||
|
assert.deepEqual(body, { hit: true });
|
||||||
|
assert.equal(state.calls, 2, "one retry after the 429");
|
||||||
|
assert.equal(slept, 1000, "honored Retry-After: 1 second");
|
||||||
|
});
|
||||||
|
|
||||||
|
test("429 retry-after is capped, and a persistent 429 throws without unbounded retry", async () => {
|
||||||
|
const { fetchToken } = countingFetchToken();
|
||||||
|
const tokens = makeSpotifyToken({ fetchToken, now: fakeClock().now });
|
||||||
|
// Retry-After: 9999s would be huge — must cap at maxRetryAfterMs (2000 by default).
|
||||||
|
const { doSearch, state } = scriptedSearch([errResp(429, hdr({ "retry-after": "9999" })), errResp(429)]);
|
||||||
|
|
||||||
|
let slept = null;
|
||||||
|
const search = makeSpotifySearch({ doSearch, tokens, sleep: async (ms) => { slept = ms; } });
|
||||||
|
await assert.rejects(search("Tool"), /spotify search http 429/);
|
||||||
|
assert.equal(slept, 2000, "the wait was capped at maxRetryAfterMs");
|
||||||
|
assert.equal(state.calls, 2, "retried exactly once, then gave up");
|
||||||
|
});
|
||||||
|
|
||||||
|
test("happy path: a 200 on the first try returns the body with no retry", async () => {
|
||||||
|
const { fetchToken } = countingFetchToken();
|
||||||
|
const tokens = makeSpotifyToken({ fetchToken, now: fakeClock().now });
|
||||||
|
const { doSearch, state } = scriptedSearch([okResp({ artists: { items: [] } })]);
|
||||||
|
|
||||||
|
const search = makeSpotifySearch({ doSearch, tokens, sleep: async () => {} });
|
||||||
|
const body = await search("Tool");
|
||||||
|
assert.deepEqual(body, { artists: { items: [] } });
|
||||||
|
assert.equal(state.calls, 1, "no retry on a clean 200");
|
||||||
|
});
|
||||||
|
|
||||||
|
test("retryAfterMs parses, caps, and rejects garbage", () => {
|
||||||
|
assert.equal(retryAfterMs(hdr({ "retry-after": "2" }), 5000), 2000);
|
||||||
|
assert.equal(retryAfterMs(hdr({ "retry-after": "100" }), 5000), 5000, "capped at maxMs");
|
||||||
|
assert.equal(retryAfterMs(hdr({ "retry-after": "junk" }), 5000), null);
|
||||||
|
assert.equal(retryAfterMs(hdr({ "retry-after": "-1" }), 5000), null);
|
||||||
|
assert.equal(retryAfterMs(hdr({}), 5000), null, "absent header -> null");
|
||||||
|
assert.equal(retryAfterMs(null, 5000), null, "no headers object -> null");
|
||||||
|
});
|
||||||
Reference in New Issue
Block a user