From 11201b5e52a64f533b62a99c1d01bfdab8e608cc Mon Sep 17 00:00:00 2001 From: Christopher Pruijsen Date: Mon, 12 Oct 2026 14:29:35 +0100 Subject: [PATCH] Handle errors from unbatched completeJob/failJob The default (unbatched) completeJob/failJob functions returned promises that nobody awaited or attached a rejection handler to, so a non-retryable error recording a job's outcome became an unhandled rejection. Attach the same error handler as the batched path so the failure is logged with the job id, emitted as pool:fatalError, and the pool shuts down gracefully. --- .changeset/proud-melons-repair.md | 9 + __tests__/main.unbatchedOutcomeErrors.test.ts | 161 ++++++++++++++++++ src/main.ts | 101 ++++++----- 3 files changed, 227 insertions(+), 44 deletions(-) create mode 100644 .changeset/proud-melons-repair.md create mode 100644 __tests__/main.unbatchedOutcomeErrors.test.ts diff --git a/.changeset/proud-melons-repair.md b/.changeset/proud-melons-repair.md new file mode 100644 index 00000000..fe42eea5 --- /dev/null +++ b/.changeset/proud-melons-repair.md @@ -0,0 +1,9 @@ +--- +"graphile-worker": patch +--- + +Fix unhandled rejection when recording a job's outcome fails with the default +unbatched `completeJob`/`failJob` (`completeJobBatchDelay`/`failJobBatchDelay` +of `-1`). Failures are now handled like the batched path: emitted as +`pool:fatalError`, logged with the job id, and the worker pool shuts down +gracefully with the error propagated to `workerPool.promise`. diff --git a/__tests__/main.unbatchedOutcomeErrors.test.ts b/__tests__/main.unbatchedOutcomeErrors.test.ts new file mode 100644 index 00000000..6fb3a924 --- /dev/null +++ b/__tests__/main.unbatchedOutcomeErrors.test.ts @@ -0,0 +1,161 @@ +import { EventEmitter } from "events"; +import type { Pool, PoolClient } from "pg"; + +import type { + TaskList, + WorkerEventMap, + WorkerSharedOptions, +} from "../src/interfaces.ts"; +import { Logger } from "../src/logger.ts"; +import { runTaskList } from "../src/main.ts"; +import { + ESCAPED_GRAPHILE_WORKER_SCHEMA, + reset, + sleepUntil, + withPgPool, +} from "./helpers.ts"; + +const addJob = (pgPool: Pool) => + pgPool.query( + `select ${ESCAPED_GRAPHILE_WORKER_SCHEMA}.add_job('job1', json_build_object('id', '1'), 'serial')`, + ); + +const options: WorkerSharedOptions = {}; + +/** + * Intercepts clients acquired from `pgPool` and makes the first query whose + * text satisfies `shouldFail` reject with an error that `withRetries` will + * not retry (one without a `code`), like a dropped connection. + */ +function failNextMatchingQuery( + pgPool: Pool, + shouldFail: (text: string) => boolean, +) { + const originalConnect = pgPool.connect; + let triggered = false; + const wrapped = new WeakSet(); + + const wrapClient = (client: PoolClient) => { + if (wrapped.has(client)) { + return client; + } + wrapped.add(client); + const originalQuery = client.query as (...args: any[]) => any; + client.query = ((config: any, ...rest: any[]) => { + const text = + typeof config === "string" ? config : (config?.text as string); + if (!triggered && typeof text === "string" && shouldFail(text)) { + triggered = true; + return Promise.reject(new Error("Connection terminated unexpectedly")); + } + return originalQuery.call(client, config, ...rest); + }) as any; + return client; + }; + + pgPool.connect = (( + cb?: ( + err: Error | undefined, + client?: PoolClient, + release?: (releaseErr?: Error | boolean) => void, + ) => void, + ) => { + const promise = (originalConnect as () => Promise) + .call(pgPool) + .then(wrapClient); + if (typeof cb === "function") { + promise.then( + (client) => cb(undefined, client, (err) => client.release(err)), + (err: Error) => cb(err), + ); + return; + } + return promise; + }) as any; + return () => { + pgPool.connect = originalConnect; + }; +} + +test.each([ + [ + "completeJob", + false, + (text: string) => + /delete\s+from/i.test(text) && text.includes("_private_jobs"), + ], + [ + "failJob", + true, + (text: string) => + text.includes("last_error") && text.includes("_private_jobs"), + ], +])( + "unbatched %s failing to record the job outcome is handled like the batched path", + (action, taskShouldFail, shouldFail) => + withPgPool(async (pgPool) => { + await reset(pgPool, options); + + const loggedErrors: string[] = []; + const logger = new Logger(() => (level, message) => { + if (level === "error") { + loggedErrors.push(message); + } + }); + + const events = new EventEmitter(); + const fatalErrors: Array = []; + events.on("pool:fatalError", (event) => { + fatalErrors.push(event); + }); + + const restore = failNextMatchingQuery(pgPool, shouldFail); + + const tasks: TaskList = { + job1() { + if (taskShouldFail) { + throw new Error("Task failed"); + } + }, + }; + + const workerPool = runTaskList( + { concurrency: 1, events, logger }, + tasks, + pgPool, + ); + // Attach handlers immediately so that a rejection of workerPool.promise + // is itself handled. + const finished = workerPool.promise.then( + () => ({ resolved: true as const }), + (error: unknown) => ({ resolved: false as const, error }), + ); + try { + await addJob(pgPool); + await sleepUntil(() => fatalErrors.length > 0); + + expect(fatalErrors).toHaveLength(1); + expect(fatalErrors[0].action).toEqual(action); + expect(fatalErrors[0].error).toBeInstanceOf(Error); + expect( + loggedErrors.some((message) => + message.includes( + action === "completeJob" + ? "Failed to complete jobs '1'" + : "Failed to fail jobs '1'", + ), + ), + ).toBe(true); + await expect(finished).resolves.toEqual({ + resolved: false, + error: expect.objectContaining({ + message: "Connection terminated unexpectedly", + }), + }); + } finally { + await Promise.resolve(workerPool.gracefulShutdown()).catch(() => {}); + await finished; + restore(); + } + }), +); diff --git a/src/main.ts b/src/main.ts index 60b06787..d1364770 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,31 @@ export function _runTaskList( return jobs[0]; }; + const handleCompleteJobsError = ( + error: unknown, + jobs: ReadonlyArray, + ) => { + safeEmit(ctx, "pool:fatalError", { + ctx, + error, + workerPool, + action: "completeJob", + }); + logger.error( + `Failed to complete jobs '${jobs + .map((j) => j.id) + .join("', '")}':\n${String(error)}`, + { fatalError: error, jobs }, + ); + if (!workerPool._shuttingDown) { + // This is the reason for shutdown + _finErrors.push(coerceError(error)); + workerPool.gracefulShutdown( + `Could not completeJobs; queue is in an inconsistent state; aborting.`, + ); + } + }; + const { release: releaseCompleteJob, fn: completeJob } = ( completeJobBatchDelay >= 0 ? batch( @@ -1260,41 +1286,47 @@ export function _runTaskList( workerPool.id, jobs, ), - (error, jobs) => { - safeEmit(ctx, "pool:fatalError", { - ctx, - error, - workerPool, - action: "completeJob", - }); - logger.error( - `Failed to complete jobs '${jobs - .map((j) => j.id) - .join("', '")}':\n${String(error)}`, - { fatalError: error, jobs }, - ); - if (!workerPool._shuttingDown) { - // This is the reason for shutdown - _finErrors.push(coerceError(error)); - workerPool.gracefulShutdown( - `Could not completeJobs; queue is in an inconsistent state; aborting.`, - ); - } - }, + handleCompleteJobsError, BATCH_RETRY_OPTIONS, ) : { release: null, + // Handle errors the same way as when batching fn: (job) => batchCompleteJobs( compiledSharedOptions, withPgClient.withRetries, workerPool.id, [job], - ), + ).then(null, (error) => handleCompleteJobsError(error, [job])), } ) as { release: (() => void) | null; fn: CompleteJobFunction }; + const handleFailJobsError = ( + error: unknown, + specs: ReadonlyArray[0]>, + ) => { + safeEmit(ctx, "pool:fatalError", { + ctx, + error, + workerPool, + action: "failJob", + }); + logger.error( + `Failed to fail jobs '${specs + .map((spec) => spec.job.id) + .join("', '")}':\n${String(error)}`, + { fatalError: error, specs }, + ); + if (!workerPool._shuttingDown) { + // This is the reason for shutdown + _finErrors.push(coerceError(error)); + workerPool.gracefulShutdown( + `Could not failJobs; queue is in an inconsistent state; aborting.`, + ); + } + }; + const { release: releaseFailJob, fn: failJob } = ( failJobBatchDelay >= 0 ? batch( @@ -1307,38 +1339,19 @@ export function _runTaskList( workerPool.id, specs, ), - (error, specs) => { - safeEmit(ctx, "pool:fatalError", { - ctx, - error, - workerPool, - action: "failJob", - }); - logger.error( - `Failed to fail jobs '${specs - .map((spec) => spec.job.id) - .join("', '")}':\n${String(error)}`, - { fatalError: error, specs }, - ); - if (!workerPool._shuttingDown) { - // This is the reason for shutdown - _finErrors.push(coerceError(error)); - workerPool.gracefulShutdown( - `Could not failJobs; queue is in an inconsistent state; aborting.`, - ); - } - }, + handleFailJobsError, BATCH_RETRY_OPTIONS, ) : { release: null, + // Handle errors the same way as when batching fn: (spec) => batchFailJobs( compiledSharedOptions, withPgClient.withRetries, workerPool.id, [spec], - ), + ).then(null, (error) => handleFailJobsError(error, [spec])), } ) as { release: (() => void) | null; fn: FailJobFunction };