BullMQ with postgres has a race condition when doing job.log

BullMQ Bug Report

Version

6.3.4

Platform

NodeJS

What happened?

Using the PostgreSQL backend with a sandboxed (external file) processor, any job that calls job.log() more than once in quick succession fails with a unique constraint violation on job_log_pkey, and log lines are silently lost.

Two things combine to cause this:

1. add_log.sql is a non-atomic read-modify-write.

dist/*/postgres/commands/add_log.sql picks the next per-job ordinal with a bare sub-select:

INSERT INTO job_log (queue, job_id, idx, row)
VALUES (
  $1, $2,
  COALESCE(
    (SELECT MAX(idx) + 1 FROM job_log WHERE queue = $1 AND job_id = $2),
    0
  ),
  $3
)
RETURNING idx;

There is no lock, no sequence and no ON CONFLICT. Two overlapping calls for the same (queue, job_id) both read the same MAX(idx), and the second insert violates PRIMARY KEY (queue, job_id, idx) declared in 0001_schema.sql.

2. The sandbox does not serialise addLog calls.

In classes/sandbox.js, msgHandler is an async function attached with child.on('message', msgHandler). EventEmitter invokes it once per IPC message and discards the returned promise, so the await job.log(msg.value) in the ParentCommand.Log branch runs concurrently with the handler for the next message. Nothing awaits the previous DB write before starting the next.

Awaiting in the processor does not avoid this. The child's job.log() is await send({ cmd: ParentCommand.Log, ... }), and send is asyncSend in utils/index.js, which resolves on the IPC write callback — not on the parent's commit. So a processor that awaits every job.log() sequentially still hands the parent a burst of messages it then processes in parallel. The reproduction below is fully sequential in the child and still fails.

This is a regression relative to Bull/Redis, where addLog was an atomic RPUSH with no client-computed index. It is not specific to custom job IDs — it reproduces with generated IDs.

The Redis backend appears unaffected, since the index is assigned server-side.

Suggested fix: make the index assignment atomic in SQL — e.g. take a pg_advisory_xact_lock(hashtext($1 || ':' || $2)) before the sub-select, or give job_log a per-job sequence — rather than relying on callers not to overlap. Serialising msgHandler in the sandbox would fix the sandboxed path but would leave direct concurrent job.log() calls from an in-process processor exposed.

How to reproduce.

Requires only a PostgreSQL instance (13+). No Redis, no other dependencies.

processor.js:

export default async function (job) {
	for (let i = 0; i < 20; i++) {
		await job.log(`line ${i}`);
	}
	return 'ok';
}

repro.js:

import { Queue, Worker, createPostgresBackend } from 'bullmq';
import { fileURLToPath } from 'url';
import { dirname, join } from 'path';

const __dirname = dirname(fileURLToPath(import.meta.url));
const connection = { connectionString: process.env.PG_URL, schema: 'bullmq_repro', migrate: true };

const queue = new Queue('repro', { connection }, createPostgresBackend);
const worker = new Worker('repro', join(__dirname, 'processor.js'),
	{ connection, concurrency: 1 }, createPostgresBackend);

worker.on('error', err => console.error('WORKER ERROR:', err.message));

const done = new Promise((resolve, reject) => {
	worker.on('completed', resolve);
	worker.on('failed', (job, err) => reject(err));
});

await queue.add('j', {});
const job = await done;
console.log('logs stored:', (await queue.getJobLogs(job.id)).logs.length, 'of 20 expected');
await queue.obliterate({ force: true });
await worker.close();
await queue.close();

Run with PG_URL=postgresql://user:pass@host/db node repro.js.

Expected: 20 log lines stored, no errors. Actual: duplicate key value violates unique constraint "job_log_pkey" on the worker's error event, and fewer than 20 lines stored.

Relevant log output

This field uses render: shell, so GitHub adds the code fence itself. Paste the lines below without backticks.

error: duplicate key value violates unique constraint "job_log_pkey"
    at /app/node_modules/pg-pool/index.js:45:11
    at process.processTicksAndRejections (node:internal/process/task_queues:95:5)
    at async PostgresQueueBackend.query (/app/node_modules/bullmq/dist/cjs/postgres/postgres-queue-backend.js:364:20)
    at async PostgresQueueBackend.addLog (/app/node_modules/bullmq/dist/cjs/postgres/postgres-queue-backend.js:953:25)
    at async Child.msgHandler (/app/node_modules/bullmq/dist/cjs/classes/sandbox.js:40:41)

Code of Conduct

Tick: I agree to follow this project's Code of Conduct