From 32f4258e9208cc89b4e1386c66691c05ce525424 Mon Sep 17 00:00:00 2001 From: Tasos Tsournos <48796551+anavalo@users.noreply.github.com> Date: Thu, 6 Aug 2026 21:19:26 +0300 Subject: [PATCH] Fix localQueue running jobs from the same named queue concurrently With batching enabled (batchSize > 1), a single job fetch could lock multiple jobs belonging to the same named queue: the queue eligibility subquery locks each available queue row once, but every job in that queue then passes the membership check, so `limit batchSize` could take several of them. The local queue would hand them to workers concurrently, and per-queue concurrency became min(localQueue.size, concurrentJobs) instead of 1. Batch fetches now keep only the first job per named queue and discard the rest; the discarded rows' locks release at the end of the statement and the queue row remains locked until the kept job completes, so the serial execution guarantee holds again. The batchSize = 1 query is unchanged. --- RELEASE_NOTES.md | 4 ++ __tests__/main.runTaskList.test.ts | 103 +++++++++++++++++++++++++++++ src/sql/getJobs.ts | 32 ++++++++- 3 files changed, 137 insertions(+), 2 deletions(-) diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index bebe8a13..cdabbaa9 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -46,6 +46,10 @@ Read more: yargs, use erasable syntax only for type-stripping support. - `LogLevel` export is now type only - a string union rather than a TypeScript const enum. +- Fix issue where enabling `localQueue` could cause jobs from the same named + queue to run concurrently (violating the serial execution guarantee for named + queues): a single batch fetch could lock multiple jobs belonging to one named + queue. Batch fetches now return at most one job per named queue (#621). ## v0.17.3 diff --git a/__tests__/main.runTaskList.test.ts b/__tests__/main.runTaskList.test.ts index c02ff2d2..c07b9d59 100644 --- a/__tests__/main.runTaskList.test.ts +++ b/__tests__/main.runTaskList.test.ts @@ -140,3 +140,106 @@ test("gracefulShutdown", async () => const [job] = jobs; expect(job.last_error).toBeTruthy(); })); + +test("jobs in the same named queue run serially even when the local queue is enabled", () => + withPgPool(async (pgPool) => { + await reset(pgPool, options); + + const jobPromises: Deferred[] = []; + try { + const job1: Task<"job1"> = jest.fn(() => { + const jobPromise = deferred(); + jobPromises.push(jobPromise); + return jobPromise; + }); + const tasks: TaskList = { + job1, + }; + + // A backlog of 5 jobs in the same named queue, before the pool starts + for (let i = 0; i < 5; i++) { + await addJob(pgPool, i); + } + + const workerPool = runTaskList( + { + concurrency: 4, + preset: { worker: { localQueue: { size: 5 }, pollInterval: 10 } }, + }, + tasks, + pgPool, + ); + + for (let i = 0; i < 5; i++) { + await sleepUntil(() => jobPromises.length >= i + 1); + // Give the pool a chance to (incorrectly) hand more jobs from the + // same queue to the other, idle workers + await sleep(50); + expect(jobPromises).toHaveLength(i + 1); + + // Complete this job, on to the next one + jobPromises[i].resolve(); + } + + await workerPool.gracefulShutdown(); + await expectJobCount(pgPool, 0); + } finally { + jobPromises.forEach((p) => p.resolve()); + } + })); + +test("jobs in different named queues run in parallel when the local queue is enabled", () => + withPgPool(async (pgPool) => { + await reset(pgPool, options); + + const started: string[] = []; + const jobPromisesById: Record = {}; + try { + const job1: Task<"job1"> = jest.fn(({ id }) => { + const jobPromise = deferred(); + jobPromisesById[id] = jobPromise; + started.push(id); + return jobPromise; + }); + const tasks: TaskList = { + job1, + }; + + const addJobToQueue = (id: string, queueName: string) => + pgPool.query( + `select ${ESCAPED_GRAPHILE_WORKER_SCHEMA}.add_job('job1', json_build_object('id', $1::text), $2::text)`, + [id, queueName], + ); + await addJobToQueue("a1", "queue_a"); + await addJobToQueue("b1", "queue_b"); + await addJobToQueue("a2", "queue_a"); + await addJobToQueue("b2", "queue_b"); + + const workerPool = runTaskList( + { + concurrency: 4, + preset: { worker: { localQueue: { size: 5 }, pollInterval: 10 } }, + }, + tasks, + pgPool, + ); + + // The first job of each queue should run concurrently... + await sleepUntil(() => started.length >= 2); + await sleep(50); + expect([...started].sort()).toEqual(["a1", "b1"]); + + // ...but each queue's second job must wait for its first to complete + jobPromisesById["a1"].resolve(); + jobPromisesById["b1"].resolve(); + await sleepUntil(() => started.length >= 4); + expect([...started].sort()).toEqual(["a1", "a2", "b1", "b2"]); + jobPromisesById["a2"].resolve(); + jobPromisesById["b2"].resolve(); + + await workerPool.gracefulShutdown(); + await expectJobCount(pgPool, 0); + } finally { + Object.values(jobPromisesById).forEach((p) => p.resolve()); + } + })); diff --git a/src/sql/getJobs.ts b/src/sql/getJobs.ts index a087078e..3a76cc86 100644 --- a/src/sql/getJobs.ts +++ b/src/sql/getJobs.ts @@ -155,8 +155,36 @@ q as ( where job_queues.id = j.job_queue_id )`; + /** + * A batch (`batchSize > 1`) may select multiple jobs from the same named + * queue: `queueClause` locks each eligible queue row only once, but then + * every job belonging to that queue passes the eligibility check, so a + * single statement can lock several jobs from one queue. Those jobs would + * then run concurrently, breaking the serial execution guarantee for named + * queues (and once the first of them completes, `completeJobs`/`returnJobs` + * would unlock the queue while its siblings are still running, letting + * other pools claim yet more jobs from the same queue). + * + * To prevent this, keep only the first job per named queue and discard the + * rest. The discarded rows' `for update` locks release when the + * transaction ends and their `locked_at`/`locked_by` were never set, so + * they remain available; they cannot be picked up early because their + * queue stays locked (via `q` below) until the job we did keep completes. + * Jobs that aren't in a named queue are unaffected. + */ + const dedupeClause = + batchSize > 1 + ? `, +j as ( + select distinct on (job_queue_id, case when job_queue_id is null then id end) + job_queue_id, priority, run_at, id + from j_raw + order by job_queue_id, case when job_queue_id is null then id end, priority asc, run_at asc +)` + : ""; + const text = `\ -with j as ( +with ${batchSize > 1 ? "j_raw" : "j"} as ( select jobs.job_queue_id, jobs.priority, jobs.run_at, jobs.id from ${escapedWorkerSchema}._private_jobs as jobs where jobs.is_available = true @@ -168,7 +196,7 @@ with j as ( limit ${batchSize} for update skip locked -)${updateQueue} +)${dedupeClause}${updateQueue} update ${escapedWorkerSchema}._private_jobs as jobs set attempts = jobs.attempts + 1,