diff --git a/workers/src/integrations/http.ts b/workers/src/integrations/http.ts index 1c20597..4bed9b9 100644 --- a/workers/src/integrations/http.ts +++ b/workers/src/integrations/http.ts @@ -134,9 +134,24 @@ export interface RequestJsonOptions { timeoutMs?: number; } -// Per-host timestamp of the last request start, used to enforce the rate limit. -// Module-scoped so all clients sharing a host coordinate automatically. -const lastRequestAt = new Map(); +// Per-host rate-limit state. Module-scoped so all clients sharing a host +// coordinate automatically. +// +// `tail` is the promise chain for a host: every throttle() call links onto the +// host's current tail, so slot acquisition is strictly serialized. Without this +// chain the old implementation was a TOCTOU race — under the worker's +// `concurrency: 10`, ten jobs read the same `lastRequestAt`, slept the same +// duration and fired in the same tick, i.e. ~10 req/s against MusicBrainz's +// 1 req/s policy. Serialization is deliberately PER HOST (not global) so a slow +// MusicBrainz queue cannot starve Last.fm, Discogs, cover art, etc. +interface HostLimiter { + /** Timestamp of the last granted request slot. */ + lastRequestAt: number; + /** Tail of the serialization chain; resolves when the previous waiter is done. */ + tail: Promise; +} + +const hostLimiters = new Map(); const delay = (ms: number): Promise => new Promise((resolve) => setTimeout(resolve, ms)); @@ -149,17 +164,43 @@ function hostOf(url: string): string { } } -/** Block until at least `minIntervalMs` has elapsed since this host's last request. */ -async function throttle(host: string, minIntervalMs: number): Promise { - const now = Date.now(); - const last = lastRequestAt.get(host); - if (last !== undefined) { - const wait = minIntervalMs - (now - last); - if (wait > 0) await delay(wait); +/** + * Acquire this host's next request slot, blocking until at least + * `minIntervalMs` has elapsed since the previously granted slot. + * + * Concurrent callers for the same host queue strictly in arrival order: each + * one appends to `limiter.tail` and only computes its wait once every earlier + * waiter has already claimed its timestamp, so N concurrent callers are spaced + * `minIntervalMs` apart rather than all firing at once. + */ +function throttle(host: string, minIntervalMs: number): Promise { + let limiter = hostLimiters.get(host); + if (!limiter) { + limiter = { lastRequestAt: 0, tail: Promise.resolve() }; + hostLimiters.set(host, limiter); } - lastRequestAt.set(host, Date.now()); + const lim = limiter; + + // Link onto the tail. The critical section (read lastRequestAt → sleep → + // write lastRequestAt) runs only after the previous waiter finished it. + const slot = lim.tail.then(async () => { + const wait = minIntervalMs - (Date.now() - lim.lastRequestAt); + if (wait > 0) await delay(wait); + lim.lastRequestAt = Date.now(); + }); + + // The next caller waits for this slot. Swallow rejections on the chain itself + // so one failure can never poison the queue for subsequent requests. + lim.tail = slot.catch(() => {}); + return slot; } +/** + * Test-only handle on the per-host throttle, so its serialization can be + * asserted without issuing real network requests. Not used in production code. + */ +export const __throttleForTest = throttle; + /** Parse a Retry-After header (delta-seconds or HTTP date) into ms, or null. */ function parseRetryAfter(value: string | null): number | null { if (!value) return null; diff --git a/workers/src/integrations/musicbrainz.client.ts b/workers/src/integrations/musicbrainz.client.ts index f6bf1fb..91e370c 100644 --- a/workers/src/integrations/musicbrainz.client.ts +++ b/workers/src/integrations/musicbrainz.client.ts @@ -206,6 +206,35 @@ interface MbReleaseGroupSearchResponse { 'release-groups'?: MbReleaseGroupSearch[]; } +/** + * Log a failed MusicBrainz call. + * + * Every public method swallows errors and returns null/[] so enrichment never + * breaks. That makes a *rate-limited* MusicBrainz indistinguishable from "no + * data exists for your library" — the job still reports success. So throttling + * (429) and MB's own 503 "your requests are exceeding the allowable rate limit" + * are escalated to console.error with an explicit, greppable message, while + * ordinary failures stay at warn. + */ +function logMbFailure(op: string, err: unknown): void { + const e = err as { status?: number; message?: string; body?: string }; + const status = typeof e?.status === 'number' ? e.status : undefined; + const msg = e?.message ?? String(err); + const rateLimited = + status === 429 || + (status === 503 && /rate limit/i.test(`${e?.body ?? ''} ${msg}`)); + + if (rateLimited) { + console.error( + `[MusicBrainz] RATE LIMITED (HTTP ${status}) on ${op} — results are ` + + `INCOMPLETE, not empty. Enrichment will report success with missing ` + + `data. Check the 1 req/s throttle and MUSICBRAINZ_CONTACT. ${msg}` + ); + return; + } + console.warn(`[MusicBrainz] ${op} failed:`, msg); +} + export class MusicBrainzClient { private readonly cfg: MusicBrainzConfig; private readonly userAgent: string; @@ -278,7 +307,7 @@ export class MusicBrainzClient { score, }; } catch (err) { - console.warn('[MusicBrainz] lookupRecording failed:', (err as Error).message); + logMbFailure('lookupRecording', err); return null; } } @@ -313,7 +342,7 @@ export class MusicBrainzClient { disambiguation: best.disambiguation ?? null, }; } catch (err) { - console.warn('[MusicBrainz] searchArtist failed:', (err as Error).message); + logMbFailure('searchArtist', err); return null; } } @@ -365,7 +394,7 @@ export class MusicBrainzClient { score, }; } catch (err) { - console.warn('[MusicBrainz] searchReleaseGroup failed:', (err as Error).message); + logMbFailure('searchReleaseGroup', err); return null; } } @@ -399,7 +428,7 @@ export class MusicBrainzClient { .map(([name, count]) => ({ name, weight: count / maxCount })) .sort((a, b) => b.weight - a.weight); } catch (err) { - console.warn('[MusicBrainz] getArtistTags failed:', (err as Error).message); + logMbFailure('getArtistTags', err); return []; } } @@ -434,7 +463,7 @@ export class MusicBrainzClient { artistCredit: credit, }; } catch (err) { - console.warn('[MusicBrainz] getRecording failed:', (err as Error).message); + logMbFailure('getRecording', err); return null; } } @@ -469,7 +498,7 @@ export class MusicBrainzClient { artistCredit: credit, }; } catch (err) { - console.warn('[MusicBrainz] getReleaseGroup failed:', (err as Error).message); + logMbFailure('getReleaseGroup', err); return null; } } @@ -512,7 +541,7 @@ export class MusicBrainzClient { })) .filter(r => r.targetMbid !== ''); } catch (err) { - console.warn('[MusicBrainz] getArtistRelations failed:', (err as Error).message); + logMbFailure('getArtistRelations', err); return []; } }