From e2d84a6f916cc97fe179d3af3076df4ea5f7caaf Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Efe=20=C4=B0nal?=
 <87021466+efeesrefinal@users.noreply.github.com>
Date: Wed, 4 Nov 2026 16:06:16 +0300
Subject: [PATCH 1/6] Wake idle workers once a named queue is released

Since completion became fire-and-forget in 0.17.0, a worker asks for its
next job before the completion of a job in a named queue has been
committed, finds the queue still locked, and waits a full pollInterval.
Nothing else wakes it, so once inserts stop a backlog in a named queue
advances only once per poll.

Nudge the pool after completing or failing jobs that belong to a named
queue, the same way a jobs:insert notification does.

Fixes #650
---
 __tests__/main.runTaskList.test.ts | 40 ++++++++++++++++++++++++++++++
 src/main.ts                        | 40 +++++++++++++++++++++++++++---
 2 files changed, 77 insertions(+), 3 deletions(-)

diff --git a/__tests__/main.runTaskList.test.ts b/__tests__/main.runTaskList.test.ts
index c07b9d59..1bf5842e 100644
--- a/__tests__/main.runTaskList.test.ts
+++ b/__tests__/main.runTaskList.test.ts
@@ -243,3 +243,43 @@ test("jobs in different named queues run in parallel when the local queue is ena
       Object.values(jobPromisesById).forEach((p) => p.resolve());
     }
   }));
+
+test.each([
+  ["without the local queue", {}],
+  ["with the local queue", { localQueue: { size: 5 } }],
+])(
+  "a backlog in a named queue drains without waiting for a poll per job once inserts stop (%s)",
+  (_description, workerPreset) =>
+    withPgPool(async (pgPool) => {
+      await reset(pgPool, options);
+
+      let completedCount = 0;
+      const tasks: TaskList = {
+        job1: () => {
+          completedCount++;
+        },
+      };
+
+      // One statement, so no `jobs:insert` notification arrives once the
+      // pool has started.
+      await pgPool.query(
+        `select ${ESCAPED_GRAPHILE_WORKER_SCHEMA}.add_job('job1', json_build_object('id', n::text), 'serial') from generate_series(1, 5) n`,
+      );
+
+      // Each job that has to wait for a poll would take a minute.
+      const workerPool = runTaskList(
+        {
+          concurrency: 3,
+          preset: { worker: { ...workerPreset, pollInterval: 60_000 } },
+        },
+        tasks,
+        pgPool,
+      );
+      try {
+        await sleepUntil(() => completedCount >= 5);
+      } finally {
+        await workerPool.gracefulShutdown();
+      }
+      await expectJobCount(pgPool, 0);
+    }),
+);
diff --git a/src/main.ts b/src/main.ts
index 60b06787..de2feb8a 100644
--- a/src/main.ts
+++ b/src/main.ts
@@ -13,6 +13,7 @@ import {
 } from "./helpers.ts";
 import type {
   CompleteJobFunction,
+  DbJob,
   EnhancedWithPgClient,
   FailJobFunction,
   GetJobFunction,
@@ -1248,6 +1249,37 @@ export function _runTaskList(
         return jobs[0];
       };
 
+  /**
+   * Completing or failing a job that belongs to a named queue unlocks that
+   * queue, so the next job in it can now be fetched. Workers only learn about
+   * new work from `jobs:insert` notifications and from polling, and the worker
+   * that ran the job asked for its next job before the completion was
+   * committed (completion does not block the worker). Without this nudge, a
+   * backlog in a named queue only advances once per `pollInterval` after
+   * inserts stop.
+   */
+  const nudgeAfterQueuedJobsReleased = (jobs: ReadonlyArray<DbJob>) => {
+    if (!workerPool._active || workerPool._shuttingDown) {
+      return;
+    }
+    let releasedQueueCount = 0;
+    for (const job of jobs) {
+      if (job.job_queue_id != null) {
+        releasedQueueCount++;
+      }
+    }
+    if (releasedQueueCount > 0) {
+      try {
+        workerPool.nudge(releasedQueueCount);
+      } catch (e) {
+        logger.error(
+          `Failed to nudge workers after releasing named queues: ${coerceError(e)}`,
+          { error: e },
+        );
+      }
+    }
+  };
+
   const { release: releaseCompleteJob, fn: completeJob } = (
     completeJobBatchDelay >= 0
       ? batch(
@@ -1259,7 +1291,7 @@ export function _runTaskList(
               withPgClient, // batch handles retries and adds backpressure
               workerPool.id,
               jobs,
-            ),
+            ).then(() => nudgeAfterQueuedJobsReleased(jobs)),
           (error, jobs) => {
             safeEmit(ctx, "pool:fatalError", {
               ctx,
@@ -1291,7 +1323,7 @@ export function _runTaskList(
               withPgClient.withRetries,
               workerPool.id,
               [job],
-            ),
+            ).then(() => nudgeAfterQueuedJobsReleased([job])),
         }
   ) as { release: (() => void) | null; fn: CompleteJobFunction };
 
@@ -1306,6 +1338,8 @@ export function _runTaskList(
               withPgClient, // batch handles retries and adds backpressure
               workerPool.id,
               specs,
+            ).then(() =>
+              nudgeAfterQueuedJobsReleased(specs.map((spec) => spec.job)),
             ),
           (error, specs) => {
             safeEmit(ctx, "pool:fatalError", {
@@ -1338,7 +1372,7 @@ export function _runTaskList(
               withPgClient.withRetries,
               workerPool.id,
               [spec],
-            ),
+            ).then(() => nudgeAfterQueuedJobsReleased([spec.job])),
         }
   ) as { release: (() => void) | null; fn: FailJobFunction };
 

From 1a1602380fb1ae10340f76fadae76f94573f3a5f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Efe=20=C4=B0nal?=
 <87021466+efeesrefinal@users.noreply.github.com>
Date: Wed, 4 Nov 2026 16:06:30 +0300
Subject: [PATCH 2/6] Add changeset

---
 .changeset/quick-queues-wake.md | 7 +++++++
 1 file changed, 7 insertions(+)
 create mode 100644 .changeset/quick-queues-wake.md

diff --git a/.changeset/quick-queues-wake.md b/.changeset/quick-queues-wake.md
new file mode 100644
index 00000000..84c988de
--- /dev/null
+++ b/.changeset/quick-queues-wake.md
@@ -0,0 +1,7 @@
+---
+"graphile-worker": patch
+---
+
+Fix a regression from 0.17.0 where a backlog in a named queue drained only once
+per `pollInterval` after inserts stopped: idle workers are now woken as soon as
+completing or failing a job releases its named queue.

From d875b7559f596036d16a0e2c35450777c97c9bf3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Efe=20=C4=B0nal?=
 <87021466+efeesrefinal@users.noreply.github.com>
Date: Wed, 4 Nov 2026 18:19:29 +0300
Subject: [PATCH 3/6] Guard nudges centrally

Check that the pool is active and not shutting down in WorkerPool.nudge
rather than in each caller, and have a worker ignore a nudge once it is
no longer active instead of asserting, since a released worker can
still be in the pool until its promise settles.
---
 src/main.ts   | 15 ++++-----------
 src/worker.ts |  6 +++++-
 2 files changed, 9 insertions(+), 12 deletions(-)

diff --git a/src/main.ts b/src/main.ts
index de2feb8a..128257fa 100644
--- a/src/main.ts
+++ b/src/main.ts
@@ -786,6 +786,9 @@ export function _runTaskList(
       return concurrency === 1 ? (this._workers[0] ?? null) : null;
     },
     nudge(this: WorkerPool, count: number) {
+      if (!this._active || this._shuttingDown) {
+        return;
+      }
       if (localQueue) {
         localQueue.pulse(count);
       } else {
@@ -1259,9 +1262,6 @@ export function _runTaskList(
    * inserts stop.
    */
   const nudgeAfterQueuedJobsReleased = (jobs: ReadonlyArray<DbJob>) => {
-    if (!workerPool._active || workerPool._shuttingDown) {
-      return;
-    }
     let releasedQueueCount = 0;
     for (const job of jobs) {
       if (job.job_queue_id != null) {
@@ -1269,14 +1269,7 @@ export function _runTaskList(
       }
     }
     if (releasedQueueCount > 0) {
-      try {
-        workerPool.nudge(releasedQueueCount);
-      } catch (e) {
-        logger.error(
-          `Failed to nudge workers after releasing named queues: ${coerceError(e)}`,
-          { error: e },
-        );
-      }
+      workerPool.nudge(releasedQueueCount);
     }
   };
 
diff --git a/src/worker.ts b/src/worker.ts
index 70034227..4eaf568a 100644
--- a/src/worker.ts
+++ b/src/worker.ts
@@ -107,7 +107,11 @@ export function makeNewWorker(
   };
 
   const nudge = () => {
-    assert.ok(active, "nudge called after worker terminated");
+    if (!active) {
+      // A released worker can still be in the pool until its promise settles
+      logger.debug("Ignoring nudge; worker is no longer active");
+      return false;
+    }
     if (doNextTimer) {
       // Must be idle; call early
       doNext();

From 359b600cc8a508afb901baaac88fa73e2484b7f1 Mon Sep 17 00:00:00 2001
From: Benjie Gillam <benjie@jemjie.com>
Date: Wed, 4 Nov 2026 16:40:19 +0000
Subject: [PATCH 4/6] Benjie's approach

---
 .changeset/quick-queues-wake.md    |  5 ++--
 __tests__/main.runTaskList.test.ts | 24 ++++++++++---------
 src/main.ts                        | 37 ++++++------------------------
 src/sql/completeJobs.ts            | 10 ++++++--
 src/sql/failJobs.ts                |  9 +++++++-
 5 files changed, 38 insertions(+), 47 deletions(-)

diff --git a/.changeset/quick-queues-wake.md b/.changeset/quick-queues-wake.md
index 84c988de..ab0ad48a 100644
--- a/.changeset/quick-queues-wake.md
+++ b/.changeset/quick-queues-wake.md
@@ -2,6 +2,5 @@
 "graphile-worker": patch
 ---
 
-Fix a regression from 0.17.0 where a backlog in a named queue drained only once
-per `pollInterval` after inserts stopped: idle workers are now woken as soon as
-completing or failing a job releases its named queue.
+Fix a bug (introduced 0.17.0) where jobs in named queues may drain as slowly as
+once per `pollInterval`.
diff --git a/__tests__/main.runTaskList.test.ts b/__tests__/main.runTaskList.test.ts
index 1bf5842e..95c61600 100644
--- a/__tests__/main.runTaskList.test.ts
+++ b/__tests__/main.runTaskList.test.ts
@@ -4,6 +4,7 @@ import type { Pool } from "pg";
 
 import type { Deferred } from "../src/deferred.ts";
 import deferred from "../src/deferred.ts";
+import { WorkerPreset } from "../src/index.ts";
 import type {
   Job,
   Task,
@@ -246,10 +247,10 @@ test("jobs in different named queues run in parallel when the local queue is ena
 
 test.each([
   ["without the local queue", {}],
-  ["with the local queue", { localQueue: { size: 5 } }],
+  ["with the local queue", { worker: { localQueue: { size: 5 } } }],
 ])(
   "a backlog in a named queue drains without waiting for a poll per job once inserts stop (%s)",
-  (_description, workerPreset) =>
+  (_description, testPreset) =>
     withPgPool(async (pgPool) => {
       await reset(pgPool, options);
 
@@ -266,17 +267,18 @@ test.each([
         `select ${ESCAPED_GRAPHILE_WORKER_SCHEMA}.add_job('job1', json_build_object('id', n::text), 'serial') from generate_series(1, 5) n`,
       );
 
-      // Each job that has to wait for a poll would take a minute.
-      const workerPool = runTaskList(
-        {
-          concurrency: 3,
-          preset: { worker: { ...workerPreset, pollInterval: 60_000 } },
+      const preset: GraphileConfig.Preset = {
+        extends: [WorkerPreset, testPreset],
+        worker: {
+          concurrentJobs: 3,
+          pollInterval: 60_000,
         },
-        tasks,
-        pgPool,
-      );
+      };
+
+      const workerPool = runTaskList({ preset }, tasks, pgPool);
       try {
-        await sleepUntil(() => completedCount >= 5);
+        // Each poll would take a minute; we only wait a second so no polls.
+        await sleepUntil(() => completedCount >= 5, 1000);
       } finally {
         await workerPool.gracefulShutdown();
       }
diff --git a/src/main.ts b/src/main.ts
index 128257fa..d28932e0 100644
--- a/src/main.ts
+++ b/src/main.ts
@@ -1252,27 +1252,6 @@ export function _runTaskList(
         return jobs[0];
       };
 
-  /**
-   * Completing or failing a job that belongs to a named queue unlocks that
-   * queue, so the next job in it can now be fetched. Workers only learn about
-   * new work from `jobs:insert` notifications and from polling, and the worker
-   * that ran the job asked for its next job before the completion was
-   * committed (completion does not block the worker). Without this nudge, a
-   * backlog in a named queue only advances once per `pollInterval` after
-   * inserts stop.
-   */
-  const nudgeAfterQueuedJobsReleased = (jobs: ReadonlyArray<DbJob>) => {
-    let releasedQueueCount = 0;
-    for (const job of jobs) {
-      if (job.job_queue_id != null) {
-        releasedQueueCount++;
-      }
-    }
-    if (releasedQueueCount > 0) {
-      workerPool.nudge(releasedQueueCount);
-    }
-  };
-
   const { release: releaseCompleteJob, fn: completeJob } = (
     completeJobBatchDelay >= 0
       ? batch(
@@ -1282,9 +1261,9 @@ export function _runTaskList(
             batchCompleteJobs(
               compiledSharedOptions,
               withPgClient, // batch handles retries and adds backpressure
-              workerPool.id,
+              workerPool,
               jobs,
-            ).then(() => nudgeAfterQueuedJobsReleased(jobs)),
+            ),
           (error, jobs) => {
             safeEmit(ctx, "pool:fatalError", {
               ctx,
@@ -1314,9 +1293,9 @@ export function _runTaskList(
             batchCompleteJobs(
               compiledSharedOptions,
               withPgClient.withRetries,
-              workerPool.id,
+              workerPool,
               [job],
-            ).then(() => nudgeAfterQueuedJobsReleased([job])),
+            ),
         }
   ) as { release: (() => void) | null; fn: CompleteJobFunction };
 
@@ -1329,10 +1308,8 @@ export function _runTaskList(
             batchFailJobs(
               compiledSharedOptions,
               withPgClient, // batch handles retries and adds backpressure
-              workerPool.id,
+              workerPool,
               specs,
-            ).then(() =>
-              nudgeAfterQueuedJobsReleased(specs.map((spec) => spec.job)),
             ),
           (error, specs) => {
             safeEmit(ctx, "pool:fatalError", {
@@ -1363,9 +1340,9 @@ export function _runTaskList(
             batchFailJobs(
               compiledSharedOptions,
               withPgClient.withRetries,
-              workerPool.id,
+              workerPool,
               [spec],
-            ).then(() => nudgeAfterQueuedJobsReleased([spec.job])),
+            ),
         }
   ) as { release: (() => void) | null; fn: FailJobFunction };
 
diff --git a/src/sql/completeJobs.ts b/src/sql/completeJobs.ts
index c56fc8d7..28043dbd 100644
--- a/src/sql/completeJobs.ts
+++ b/src/sql/completeJobs.ts
@@ -1,4 +1,4 @@
-import type { DbJob, WithPgClient } from "../interfaces.ts";
+import type { DbJob, WithPgClient, WorkerPool } from "../interfaces.ts";
 import type { CompiledSharedOptions } from "../lib.ts";
 
 const manualPrepare = false;
@@ -6,9 +6,10 @@ const manualPrepare = false;
 export async function batchCompleteJobs(
   compiledSharedOptions: CompiledSharedOptions,
   withPgClient: WithPgClient,
-  poolId: string,
+  workerPool: WorkerPool,
   jobs: ReadonlyArray<DbJob>,
 ): Promise<void> {
+  const poolId = workerPool.id;
   const {
     escapedWorkerSchema,
     workerSchema,
@@ -83,4 +84,9 @@ where id = n`,
       );
     }
   }
+
+  // Avoid slow drain of named queues by nudging now lock is released
+  if (jobIdsWithQueue.length > 0) {
+    workerPool.nudge(jobIdsWithQueue.length);
+  }
 }
diff --git a/src/sql/failJobs.ts b/src/sql/failJobs.ts
index 281de35a..bfe8ccc7 100644
--- a/src/sql/failJobs.ts
+++ b/src/sql/failJobs.ts
@@ -2,6 +2,7 @@ import type {
   DbJob,
   EnhancedWithPgClient,
   WithPgClient,
+  WorkerPool,
 } from "../interfaces.ts";
 import type { CompiledSharedOptions } from "../lib.ts";
 interface Spec {
@@ -13,9 +14,10 @@ interface Spec {
 export async function batchFailJobs(
   compiledSharedOptions: CompiledSharedOptions,
   withPgClient: WithPgClient,
-  poolId: string,
+  workerPool: WorkerPool,
   specs: ReadonlyArray<Spec>,
 ): Promise<void> {
+  const poolId = workerPool.id;
   const {
     escapedWorkerSchema,
     workerSchema,
@@ -96,6 +98,11 @@ where id = (el->>'jobId')::bigint and locked_by = $1::text;`,
       }),
     );
   }
+
+  // Avoid slow drain of named queues by nudging now lock is released
+  if (specsWithQueues.length > 0) {
+    workerPool.nudge(specsWithQueues.length);
+  }
 }
 
 export async function failJobs(

From 4fc9ddc6a1878386bb1dbf81537221e850322057 Mon Sep 17 00:00:00 2001
From: Benjie Gillam <benjie@jemjie.com>
Date: Wed, 4 Nov 2026 16:41:38 +0000
Subject: [PATCH 5/6] Ignore changesets

---
 package.json | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/package.json b/package.json
index 28e7a548..62a020a3 100644
--- a/package.json
+++ b/package.json
@@ -16,7 +16,7 @@
     "test": "yarn prepack && yarn depcheck && yarn test:setupdb && yarn test:only",
     "test:setupdb": "./scripts/setup_template_db.sh",
     "test:only": "NO_LOG_SUCCESS=1 node --experimental-vm-modules node_modules/jest/bin/jest.js",
-    "depcheck": "depcheck --ignores='graphile-worker,faktory-worker,@google-cloud/tasks,bullmq,jest-environment-node,@docusaurus/*,@fortawesome/*,@mdx-js/*,@types/jest,clsx,eslint_d,graphile,juice,postcss-nested,prism-react-renderer,react,react-dom,svgo,@types/*,@tsconfig/*,@typescript-eslint/*,eslint-*'",
+    "depcheck": "depcheck --ignores='graphile-worker,faktory-worker,@google-cloud/tasks,bullmq,jest-environment-node,@docusaurus/*,@fortawesome/*,@mdx-js/*,@types/jest,clsx,eslint_d,graphile,juice,postcss-nested,prism-react-renderer,react,react-dom,svgo,@types/*,@tsconfig/*,@typescript-eslint/*,eslint-*,@changesets/*'",
     "db:dump": "./scripts/dump_db",
     "perfTest": "cd perfTest && node ./run.js",
     "towerDefence": "cd towerDefence && node ./run.mjs",

From 147f45867ef80a8c43513ef03ad0a3b9c31bd053 Mon Sep 17 00:00:00 2001
From: Benjie Gillam <benjie@jemjie.com>
Date: Wed, 4 Nov 2026 16:48:25 +0000
Subject: [PATCH 6/6] Lint

---
 src/main.ts | 1 -
 1 file changed, 1 deletion(-)

diff --git a/src/main.ts b/src/main.ts
index d28932e0..7d530041 100644
--- a/src/main.ts
+++ b/src/main.ts
@@ -13,7 +13,6 @@ import {
 } from "./helpers.ts";
 import type {
   CompleteJobFunction,
-  DbJob,
   EnhancedWithPgClient,
   FailJobFunction,
   GetJobFunction,