diff --git a/.changeset/quick-queues-wake.md b/.changeset/quick-queues-wake.md new file mode 100644 index 00000000..ab0ad48a --- /dev/null +++ b/.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`. diff --git a/__tests__/main.runTaskList.test.ts b/__tests__/main.runTaskList.test.ts index c07b9d59..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, @@ -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); + }), +); 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", diff --git a/src/main.ts b/src/main.ts index 60b06787..7d530041 100644 --- a/src/main.ts +++ b/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], ), } 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, ): Promise { + 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, ): Promise { + 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( 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();