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,