From ee43995e9610d4b69cf7bd4ded1bbcd801e094e2 Mon Sep 17 00:00:00 2001 From: kami Date: Thu, 30 Jul 2026 23:56:29 +0400 Subject: [PATCH] fix: give the worker a pg Pool and real transactions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The worker ran every job through a single pg Client while BullMQ was configured with concurrency: 10. A Client is one connection with one protocol stream and no queueing: ten concurrent jobs interleave on it, and any BEGIN/COMMIT is shared by all of them, so an unrelated job's failure can roll back another's work and a rollback can discard a third's committed intent. Switched to a Pool, added a small withTransaction(pool, fn) helper that takes a dedicated connection per transaction, and threaded a Queryable interface through the services so they accept either a pool or a pooled client. Both reprocess_artists merge blocks — the artist merge and the duplicate-album merge — now run inside withTransaction; previously a failure partway through left artists merged and their tracks unmoved. integrity.service and cleanup.service get only the constructor type change here so this commit compiles; their own fixes follow in the next two commits. cleanup.service's BEGIN/COMMIT-on-a-Pool is therefore still wrong at this commit and is replaced wholesale by the hard-delete commit. REVIEW-2026-07-30.md finding 4 (and the concurrency note in finding 3). Co-Authored-By: Claude Opus 5 --- workers/src/audio-features.service.ts | 4 +- workers/src/cleanup.service.ts | 4 +- workers/src/db.ts | 57 +++++++ workers/src/enrichment.service.ts | 4 +- workers/src/index.ts | 205 +++++++++++++++----------- workers/src/integrity.service.ts | 4 +- workers/src/mb-spine-writer.ts | 4 +- workers/src/scanner.service.ts | 4 +- 8 files changed, 186 insertions(+), 100 deletions(-) create mode 100644 workers/src/db.ts diff --git a/workers/src/audio-features.service.ts b/workers/src/audio-features.service.ts index 9d07659..ceb67ad 100644 --- a/workers/src/audio-features.service.ts +++ b/workers/src/audio-features.service.ts @@ -15,7 +15,7 @@ */ import { spawn } from 'child_process'; import mm from 'music-metadata'; -import { Client as PgClient } from 'pg'; +import type { Queryable } from './db.js'; // Lazy WASM singleton — heavy to load (~2.4 MB), so we initialise once and // reuse across all enrichment jobs within the same worker process. @@ -76,7 +76,7 @@ export interface AudioFeatures { } export class AudioFeaturesService { - constructor(private pgClient: PgClient) {} + constructor(private pgClient: Queryable) {} async ensureSchema(): Promise { await this.pgClient.query( diff --git a/workers/src/cleanup.service.ts b/workers/src/cleanup.service.ts index 25c60da..ff45d69 100644 --- a/workers/src/cleanup.service.ts +++ b/workers/src/cleanup.service.ts @@ -1,4 +1,4 @@ -import { Client as PgClient } from 'pg'; +import type { Pool } from 'pg'; import { unlink } from 'fs/promises'; const NTFY_URL = process.env.NTFY_URL || ''; @@ -19,7 +19,7 @@ async function sendNtfy(title: string, message: string): Promise { } export class CleanupSweepService { - constructor(private pgClient: PgClient) {} + constructor(private pgClient: Pool) {} async runSweep(): Promise<{ warned: number; deleted: number }> { const warned = await this.advanceToWarned(); diff --git a/workers/src/db.ts b/workers/src/db.ts new file mode 100644 index 0000000..c84858e --- /dev/null +++ b/workers/src/db.ts @@ -0,0 +1,57 @@ +import type { Pool, PoolClient, QueryResult, QueryResultRow } from 'pg'; + +/** + * The subset of pg's API the worker services actually need: a `query()` that + * takes SQL plus positional params. + * + * Services are typed against this rather than against `Client` so they work + * unchanged whether they are handed a `Pool` (the worker process — see + * index.ts), a `PoolClient` checked out for a transaction, or a plain `Client` + * (the one-off maintenance scripts in ./scripts, which are single-threaded and + * own their connection). + * + * IMPORTANT: a `Queryable` gives NO transaction guarantees. When a `Pool` is + * behind it, consecutive `query()` calls may land on different connections, so + * bare `BEGIN`/`COMMIT` must never be issued through it — check out a dedicated + * client with `withTransaction()` instead. + */ +export interface Queryable { + query( + sql: string, + params?: any[] + ): Promise>; +} + +/** + * Run `fn` inside a transaction on a dedicated pooled connection, committing on + * success and rolling back on any throw. The client is always released. + * + * This is the only correct way to run a transaction in the worker: the process + * runs BullMQ with `concurrency: 10`, so issuing `BEGIN` on a shared connection + * would enrol another job's unrelated queries in this transaction — and discard + * them on `ROLLBACK`. (The backend documents the same hazard as its reason for + * using a `Pool`; see backend/src/app.ts.) + */ +export async function withTransaction( + pool: Pool, + fn: (client: PoolClient) => Promise +): Promise { + const client = await pool.connect(); + try { + await client.query('BEGIN'); + const result = await fn(client); + await client.query('COMMIT'); + return result; + } catch (err) { + try { + await client.query('ROLLBACK'); + } catch (rollbackErr) { + // A failed ROLLBACK means the connection is unusable; log and move on — + // release() below discards it rather than returning it to the pool. + console.error('[DB] ROLLBACK failed:', rollbackErr); + } + throw err; + } finally { + client.release(); + } +} diff --git a/workers/src/enrichment.service.ts b/workers/src/enrichment.service.ts index 8a76d1b..34b221e 100644 --- a/workers/src/enrichment.service.ts +++ b/workers/src/enrichment.service.ts @@ -1,4 +1,4 @@ -import { Client as PgClient } from 'pg'; +import type { Queryable } from './db.js'; import { MusicBrainzClient, LastFmClient, @@ -65,7 +65,7 @@ export class EnrichmentService { private readonly deezer = new DeezerClient(); private readonly audioFeatures: AudioFeaturesService; - constructor(private pgClient: PgClient) { + constructor(private pgClient: Queryable) { this.audioFeatures = new AudioFeaturesService(pgClient); } diff --git a/workers/src/index.ts b/workers/src/index.ts index bf7c24b..2c92b0b 100644 --- a/workers/src/index.ts +++ b/workers/src/index.ts @@ -1,12 +1,13 @@ import { Worker, Job } from 'bullmq'; import { connection, QUEUE_NAME, queue } from './queue.js'; import { MetadataRefreshJob, AudioAnalysisJob, LibraryScanJob, ArtistSimilarityJob, ArtistImageJob, AlbumCoverJob, ReprocessArtistsJob } from './types.js'; -import { Client as PgClient } from 'pg'; +import { Pool } from 'pg'; import { ScannerService } from './scanner.service.js'; import { IntegrityService } from './integrity.service.js'; import { EnrichmentService } from './enrichment.service.js'; import { AudioFeaturesService } from './audio-features.service.js'; import { CleanupSweepService } from './cleanup.service.js'; +import { withTransaction } from './db.js'; // Cron for the periodic integrity sweep (default: daily at 03:00). Configurable // via INTEGRITY_SWEEP_CRON. MUSIC_DIR (consumed by IntegrityService) controls @@ -21,23 +22,40 @@ const VIBE_REAP_CRON = process.env.VIBE_REAP_CRON || '0 * * * *'; // Worker concurrency - how many jobs to process in parallel const WORKER_CONCURRENCY = parseInt(process.env.WORKER_CONCURRENCY || '10', 10); -const pgClient = new PgClient({ +// A Pool, not a single Client. The worker processes jobs with +// `concurrency: 10` on one event loop, so a shared Client would multiplex every +// job's queries onto one connection: a `BEGIN` issued by one job (see +// cleanup.service.ts) would enrol other jobs' unrelated queries in that +// transaction and discard them on `ROLLBACK`. Pool gives each transaction its +// own connection. Same reasoning as backend/src/app.ts. +// +// Sized at least as large as the job concurrency so concurrent jobs never +// serialise waiting for a connection. +const pgPool = new Pool({ connectionString: process.env.DATABASE_URL, + max: Math.max(WORKER_CONCURRENCY + 2, 10), +}); + +pgPool.on('error', (err) => { + // Idle-client errors would otherwise be an unhandled 'error' event and crash + // the worker; the pool discards the client itself. + console.error('[DB] Idle pool client error:', err); }); async function initWorker() { - await pgClient.connect(); + // Fail fast if the database is unreachable at boot (Pool is lazy otherwise). + await pgPool.query('SELECT 1'); console.log('Worker connected to PostgreSQL'); - const scannerService = new ScannerService(pgClient, queue); - const enrichmentService = new EnrichmentService(pgClient); - const audioFeaturesService = new AudioFeaturesService(pgClient); + const scannerService = new ScannerService(pgPool, queue); + const enrichmentService = new EnrichmentService(pgPool); + const audioFeaturesService = new AudioFeaturesService(pgPool); await audioFeaturesService.ensureSchema(); // Self-provision the enrichment schema additions (artist_similar + reused // columns) at startup, alongside the integrity table. await enrichmentService.ensureSchema(); - await new IntegrityService(pgClient, queue).ensureSchema(); + await new IntegrityService(pgPool, queue).ensureSchema(); const worker = new Worker(QUEUE_NAME, async (job: Job) => { // console.log(`Processing job: ${job.name} (ID: ${job.id})`); @@ -93,7 +111,7 @@ async function initWorker() { // Periodic self-healing pass: detect corrupt/missing track metadata, // auto-fix via rescan + SQL strip, flag the rest for manual review. console.log('[Integrity] Starting integrity sweep'); - const integrityService = new IntegrityService(pgClient, queue); + const integrityService = new IntegrityService(pgPool, queue); const summary = await integrityService.runSweep(); console.log( `[Integrity] Sweep complete: detected=${summary.detected} fixed=${summary.fixed} needsReview=${summary.needsReview}` @@ -102,7 +120,7 @@ async function initWorker() { } case 'cleanup_sweep': { console.log('[Cleanup] Starting dislike cleanup sweep'); - const cleanupService = new CleanupSweepService(pgClient); + const cleanupService = new CleanupSweepService(pgPool); const result = await cleanupService.runSweep(); console.log(`[Cleanup] Sweep complete: warned=${result.warned} deleted=${result.deleted}`); break; @@ -110,7 +128,7 @@ async function initWorker() { case 'vibe_reap': { // Stale-session reaper (Invariant B). Transitions ACTIVE batches with no // interaction for 24h to RESOLVED. Runs hourly; the SQL is idempotent. - const reaped = await pgClient.query( + const reaped = await pgPool.query( `UPDATE recommendation_batch SET status = 'RESOLVED' WHERE status = 'ACTIVE' @@ -152,7 +170,7 @@ async function initWorker() { if (!err?.message?.includes('already exists')) throw err; } - const tracksRes = await pgClient.query( + const tracksRes = await pgPool.query( `SELECT t.id, t.title, t.artist, al.title AS album, t.duration, t.play_count, t.source_type FROM tracks t LEFT JOIN albums al ON al.id = t.album_id @@ -161,7 +179,7 @@ async function initWorker() { const tracks = tracksRes.rows; const genreMap = new Map(); - const genreRes = await pgClient.query( + const genreRes = await pgPool.query( `SELECT tg.track_id, g.name FROM track_genre tg JOIN genre g ON g.id = tg.genre_id` ); for (const row of genreRes.rows) { @@ -199,7 +217,7 @@ async function initWorker() { const offset = payload.offset ?? 0; console.log(`[ReprocessArtists] Starting artist reprocessing (batch=${batchSize}, offset=${offset})`); - const artistsRes = await pgClient.query( + const artistsRes = await pgPool.query( `SELECT id, name, canonical_name, mbid FROM artists ORDER BY id LIMIT $1 OFFSET $2`, [batchSize, offset] ); @@ -215,43 +233,50 @@ async function initWorker() { // Merge if resolved to a different artist (duplicate detected) if (result.artistId !== artist.id) { merged++; - // Move track links to the keeper, skipping any (track, role) the - // keeper already has, then drop the loser's — a plain UPDATE would - // violate track_artists_pkey when both are on the same track. - await pgClient.query( - `INSERT INTO track_artists (track_id, artist_id, role) - SELECT track_id, $1, role FROM track_artists WHERE artist_id = $2 - ON CONFLICT (track_id, artist_id, role) DO NOTHING`, - [result.artistId, artist.id] - ); - await pgClient.query( - `DELETE FROM track_artists WHERE artist_id = $1`, - [artist.id] - ); - // Fold albums into the keeper. Move tracks of any same-title album - // to the keeper's matching album first (UNIQUE(artist_id,title) and - // ON DELETE CASCADE mean a blind UPDATE could collide or, worse, - // cascade-delete tracks when the loser artist is removed). - const dupAlbums = await pgClient.query( - `SELECT l.id AS loser_id, k.id AS keeper_id - FROM albums l JOIN albums k - ON k.artist_id = $1 AND lower(k.title) = lower(l.title) - WHERE l.artist_id = $2`, - [result.artistId, artist.id] - ); - for (const { loser_id, keeper_id } of dupAlbums.rows) { - await pgClient.query(`UPDATE tracks SET album_id = $1 WHERE album_id = $2`, [keeper_id, loser_id]); - await pgClient.query(`DELETE FROM albums WHERE id = $1`, [loser_id]); - } - // Remaining (non-colliding) albums move over cleanly. - await pgClient.query( - `UPDATE albums SET artist_id = $1 WHERE artist_id = $2`, - [result.artistId, artist.id] - ); - await pgClient.query( - `DELETE FROM artists WHERE id = $1`, - [artist.id] - ); + // Whole merge is one transaction on one dedicated connection: the + // intermediate states (track links moved but albums not yet, or + // vice versa) must never be visible, and a failure part-way must + // not leave an artist half-merged. ON DELETE CASCADE makes a + // partial merge destructive. + await withTransaction(pgPool, async (client) => { + // Move track links to the keeper, skipping any (track, role) the + // keeper already has, then drop the loser's — a plain UPDATE would + // violate track_artists_pkey when both are on the same track. + await client.query( + `INSERT INTO track_artists (track_id, artist_id, role) + SELECT track_id, $1, role FROM track_artists WHERE artist_id = $2 + ON CONFLICT (track_id, artist_id, role) DO NOTHING`, + [result.artistId, artist.id] + ); + await client.query( + `DELETE FROM track_artists WHERE artist_id = $1`, + [artist.id] + ); + // Fold albums into the keeper. Move tracks of any same-title album + // to the keeper's matching album first (UNIQUE(artist_id,title) and + // ON DELETE CASCADE mean a blind UPDATE could collide or, worse, + // cascade-delete tracks when the loser artist is removed). + const dupAlbums = await client.query( + `SELECT l.id AS loser_id, k.id AS keeper_id + FROM albums l JOIN albums k + ON k.artist_id = $1 AND lower(k.title) = lower(l.title) + WHERE l.artist_id = $2`, + [result.artistId, artist.id] + ); + for (const { loser_id, keeper_id } of dupAlbums.rows) { + await client.query(`UPDATE tracks SET album_id = $1 WHERE album_id = $2`, [keeper_id, loser_id]); + await client.query(`DELETE FROM albums WHERE id = $1`, [loser_id]); + } + // Remaining (non-colliding) albums move over cleanly. + await client.query( + `UPDATE albums SET artist_id = $1 WHERE artist_id = $2`, + [result.artistId, artist.id] + ); + await client.query( + `DELETE FROM artists WHERE id = $1`, + [artist.id] + ); + }); console.log(`[ReprocessArtists] Merged "${artist.name}" (${artist.id}) -> "${result.canonicalName}" (${result.artistId})`); } else { // Update the existing artist record with new canonical_name and/or mbid @@ -274,7 +299,7 @@ async function initWorker() { if (updates.length > 0) { updates.push(`updated_at = CURRENT_TIMESTAMP`); params.push(artist.id); - await pgClient.query( + await pgPool.query( `UPDATE artists SET ${updates.join(', ')} WHERE id = $${paramIdx}`, params ); @@ -313,7 +338,7 @@ async function initWorker() { // Final batch - run deduplication pass to merge artists with same normalized_name console.log(`[ReprocessArtists] All batches complete, running deduplication pass...`); - const dupRes = await pgClient.query( + const dupRes = await pgPool.query( `SELECT normalized_name, array_agg(id ORDER BY CASE WHEN mbid IS NOT NULL THEN 0 ELSE 1 END, CASE WHEN canonical_name IS NOT NULL THEN 0 ELSE 1 END, @@ -333,41 +358,45 @@ async function initWorker() { for (const mergeId of mergeIds) { if (mergeId === keepId) continue; try { - // First, handle track_artists conflicts: if both artists are on same track, - // keep the 'main' role, or merge roles. Use ON CONFLICT DO NOTHING to skip duplicates. - await pgClient.query( - `INSERT INTO track_artists (track_id, artist_id, role) - SELECT track_id, $1, role FROM track_artists WHERE artist_id = $2 - ON CONFLICT (track_id, artist_id, role) DO NOTHING`, - [keepId, mergeId] - ); - // Then delete the old track_artists entries - await pgClient.query( - `DELETE FROM track_artists WHERE artist_id = $1`, - [mergeId] - ); + // One transaction per merge, on a dedicated connection: a + // partial merge is destructive (ON DELETE CASCADE). + await withTransaction(pgPool, async (client) => { + // First, handle track_artists conflicts: if both artists are on same track, + // keep the 'main' role, or merge roles. Use ON CONFLICT DO NOTHING to skip duplicates. + await client.query( + `INSERT INTO track_artists (track_id, artist_id, role) + SELECT track_id, $1, role FROM track_artists WHERE artist_id = $2 + ON CONFLICT (track_id, artist_id, role) DO NOTHING`, + [keepId, mergeId] + ); + // Then delete the old track_artists entries + await client.query( + `DELETE FROM track_artists WHERE artist_id = $1`, + [mergeId] + ); - // Fold same-title albums (move tracks) before reassigning the rest, - // to avoid UNIQUE(artist_id,title) collisions / cascade deletes. - const dupAlbums = await pgClient.query( - `SELECT l.id AS loser_id, k.id AS keeper_id - FROM albums l JOIN albums k - ON k.artist_id = $1 AND lower(k.title) = lower(l.title) - WHERE l.artist_id = $2`, - [keepId, mergeId] - ); - for (const { loser_id, keeper_id } of dupAlbums.rows) { - await pgClient.query(`UPDATE tracks SET album_id = $1 WHERE album_id = $2`, [keeper_id, loser_id]); - await pgClient.query(`DELETE FROM albums WHERE id = $1`, [loser_id]); - } - await pgClient.query( - `UPDATE albums SET artist_id = $1 WHERE artist_id = $2`, - [keepId, mergeId] - ); - await pgClient.query( - `DELETE FROM artists WHERE id = $1`, - [mergeId] - ); + // Fold same-title albums (move tracks) before reassigning the rest, + // to avoid UNIQUE(artist_id,title) collisions / cascade deletes. + const dupAlbums = await client.query( + `SELECT l.id AS loser_id, k.id AS keeper_id + FROM albums l JOIN albums k + ON k.artist_id = $1 AND lower(k.title) = lower(l.title) + WHERE l.artist_id = $2`, + [keepId, mergeId] + ); + for (const { loser_id, keeper_id } of dupAlbums.rows) { + await client.query(`UPDATE tracks SET album_id = $1 WHERE album_id = $2`, [keeper_id, loser_id]); + await client.query(`DELETE FROM albums WHERE id = $1`, [loser_id]); + } + await client.query( + `UPDATE albums SET artist_id = $1 WHERE artist_id = $2`, + [keepId, mergeId] + ); + await client.query( + `DELETE FROM artists WHERE id = $1`, + [mergeId] + ); + }); dedupMerged++; console.log(`[ReprocessArtists] Dedup merged ${mergeId} -> ${keepId} (normalized: ${row.normalized_name})`); } catch (err) { @@ -437,7 +466,7 @@ async function initWorker() { // deleting data/postgres, or first deploy). On subsequent restarts the library // is already populated, so an unconditional scan would be wasteful. const startupDir = process.env.MUSIC_DIR || '/music'; - const trackCount = await pgClient.query('SELECT COUNT(*)::int AS n FROM tracks'); + const trackCount = await pgPool.query('SELECT COUNT(*)::int AS n FROM tracks'); if (trackCount.rows[0].n === 0) { await queue.add('scan_library', { directory: startupDir }, { removeOnComplete: { age: 86400, count: 100 }, @@ -460,7 +489,7 @@ async function initWorker() { try { await worker.close(); await queue.close(); - await pgClient.end(); + await pgPool.end(); console.log('[Shutdown] Clean shutdown complete.'); } catch (err) { console.error('[Shutdown] Error during shutdown:', err); diff --git a/workers/src/integrity.service.ts b/workers/src/integrity.service.ts index c9b8cb4..9866c03 100644 --- a/workers/src/integrity.service.ts +++ b/workers/src/integrity.service.ts @@ -1,5 +1,5 @@ import fs from 'fs/promises'; -import { Client as PgClient } from 'pg'; +import type { Queryable } from './db.js'; import { Queue } from 'bullmq'; import { ScannerService } from './scanner.service.js'; @@ -36,7 +36,7 @@ export interface SweepSummary { export class IntegrityService { private scanner: ScannerService; - constructor(private pgClient: PgClient, private queue: Queue) { + constructor(private pgClient: Queryable, private queue: Queue) { this.scanner = new ScannerService(pgClient, queue); } diff --git a/workers/src/mb-spine-writer.ts b/workers/src/mb-spine-writer.ts index 37a311c..8508e77 100644 --- a/workers/src/mb-spine-writer.ts +++ b/workers/src/mb-spine-writer.ts @@ -1,4 +1,4 @@ -import { Client as PgClient } from 'pg'; +import type { Queryable } from './db.js'; import { MusicBrainzClient } from './integrations/musicbrainz.client.js'; import { normalizeForMatching } from './utils/fuzzy-match.js'; @@ -9,7 +9,7 @@ function generateSortName(name: string): string { } export class MbSpineWriter { - constructor(private pgClient: PgClient) {} + constructor(private pgClient: Queryable) {} /** * Fetch full artist-credit for a recording MBID and write claims. diff --git a/workers/src/scanner.service.ts b/workers/src/scanner.service.ts index 2e565db..40bcc61 100644 --- a/workers/src/scanner.service.ts +++ b/workers/src/scanner.service.ts @@ -3,7 +3,7 @@ import { createReadStream } from 'fs'; import { createHash } from 'crypto'; import path from 'path'; import mm from 'music-metadata'; -import { Client as PgClient } from 'pg'; +import type { Queryable } from './db.js'; import { Queue } from 'bullmq'; import { MetadataRefreshJob, ArtistSimilarityJob, ArtistImageJob, AlbumCoverJob } from './types.js'; import { splitArtistNames, parseArtists } from './utils/artist-names.js'; @@ -80,7 +80,7 @@ export class ScannerService { private enqueuedArtists = new Set(); private enqueuedAlbums = new Set(); - constructor(private pgClient: PgClient, private queue: Queue) {} + constructor(private pgClient: Queryable, private queue: Queue) {} async scanDirectory(directory: string) { console.log(`[Scanner] Starting scan in: ${directory}`);