Wake idle workers once a named queue is released

.changeset/quick-queues-wake.md

@@ -0,0 +1,6 @@
+---
+"graphile-worker": patch
+---
+
+Fix a bug (introduced 0.17.0) where jobs in named queues may drain as slowly as
+once per `pollInterval`.

__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,
@@ -243,3 +244,44 @@ 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", { worker: { localQueue: { size: 5 } } }],
+])(
+  "a backlog in a named queue drains without waiting for a poll per job once inserts stop (%s)",
+  (_description, testPreset) =>
+    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`,
+      );
+
+      const preset: GraphileConfig.Preset = {
+        extends: [WorkerPreset, testPreset],
+        worker: {
+          concurrentJobs: 3,
+          pollInterval: 60_000,
+        },
+      };
+
+      const workerPool = runTaskList({ preset }, tasks, pgPool);
+      try {
+        // Each poll would take a minute; we only wait a second so no polls.
+        await sleepUntil(() => completedCount >= 5, 1000);
+      } finally {
+        await workerPool.gracefulShutdown();
+      }
+      await expectJobCount(pgPool, 0);
+    }),
+);

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",

src/main.ts

@@ -785,6 +785,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 {
@@ -1257,7 +1260,7 @@ export function _runTaskList(
             batchCompleteJobs(
               compiledSharedOptions,
               withPgClient, // batch handles retries and adds backpressure
-              workerPool.id,
+              workerPool,
               jobs,
             ),
           (error, jobs) => {
@@ -1289,7 +1292,7 @@ export function _runTaskList(
             batchCompleteJobs(
               compiledSharedOptions,
               withPgClient.withRetries,
-              workerPool.id,
+              workerPool,
               [job],
             ),
         }
@@ -1304,7 +1307,7 @@ export function _runTaskList(
             batchFailJobs(
               compiledSharedOptions,
               withPgClient, // batch handles retries and adds backpressure
-              workerPool.id,
+              workerPool,
               specs,
             ),
           (error, specs) => {
@@ -1336,7 +1339,7 @@ export function _runTaskList(
             batchFailJobs(
               compiledSharedOptions,
               withPgClient.withRetries,
-              workerPool.id,
+              workerPool,
               [spec],
             ),
         }

src/sql/completeJobs.ts

@@ -1,14 +1,15 @@
-import type { DbJob, WithPgClient } from "../interfaces.ts";
+import type { DbJob, WithPgClient, WorkerPool } from "../interfaces.ts";
 import type { CompiledSharedOptions } from "../lib.ts";
 
 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);
+  }
 }

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(

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();