kami/muzick
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

fix: give the worker a pg Pool and real transactions

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
kami
2026-07-30 23:56:29 +04:00
parent d0ca479d4f
commit ee43995e96
8 changed files with 186 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
workers/src/index.ts
+117 -88
View File
@@ -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<any>) => {
// 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<string, string[]>();
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);