Fix localQueue running jobs from the same named queue concurrently

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
 

__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<string, Deferred> = {};
+    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());
+    }
+  }));

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,