diff --git a/.changeset/postgres-queue-error-metadata.md b/.changeset/postgres-queue-error-metadata.md new file mode 100644 index 0000000000..e254101511 --- /dev/null +++ b/.changeset/postgres-queue-error-metadata.md @@ -0,0 +1,5 @@ +--- +'@workflow/world-postgres': patch +--- + +Preserve error names, messages, stacks, and nested causes in Graphile Worker log metadata. diff --git a/packages/world-postgres/src/queue-logging.test.ts b/packages/world-postgres/src/queue-logging.test.ts new file mode 100644 index 0000000000..17fc681fd5 --- /dev/null +++ b/packages/world-postgres/src/queue-logging.test.ts @@ -0,0 +1,61 @@ +import { describe, expect, it } from 'vitest'; +import { serializeGraphileMeta } from './queue.js'; + +describe('serializeGraphileMeta', () => { + it('expands the non-enumerable Error fields JSON.stringify drops', () => { + const error = new Error('delivery failed'); + expect(JSON.parse(JSON.stringify({ error }))).toEqual({ error: {} }); + + const meta = JSON.parse(serializeGraphileMeta({ error, jobId: '7' })); + expect(meta).toMatchObject({ + jobId: '7', + error: { + name: 'Error', + message: 'delivery failed', + stack: expect.stringContaining('Error: delivery failed'), + }, + }); + expect(meta.error).not.toHaveProperty('cause'); + }); + + it('keeps enumerable diagnostics and recurses through cause and AggregateError', () => { + const socket = Object.assign(new Error('other side closed'), { + code: 'UND_ERR_SOCKET', + }); + const fetchFailed = new TypeError('fetch failed', { cause: socket }); + const aggregate = new AggregateError([fetchFailed], 'all attempts failed'); + + const meta = JSON.parse(serializeGraphileMeta({ error: aggregate })); + expect(meta.error).toMatchObject({ + name: 'AggregateError', + message: 'all attempts failed', + errors: [ + { + name: 'TypeError', + message: 'fetch failed', + cause: { + name: 'Error', + message: 'other side closed', + code: 'UND_ERR_SOCKET', + stack: expect.stringContaining('other side closed'), + }, + }, + ], + }); + }); + + it('does not throw on a cyclic cause chain', () => { + const outer = new Error('outer'); + const inner = new Error('inner', { cause: outer }); + outer.cause = inner; + + const meta = JSON.parse(serializeGraphileMeta({ error: outer })); + expect(meta.error).toMatchObject({ + message: 'outer', + cause: { + message: 'inner', + cause: { name: 'Error', message: 'outer', repeated: true }, + }, + }); + }); +}); diff --git a/packages/world-postgres/src/queue.ts b/packages/world-postgres/src/queue.ts index 898960fa7c..6a7db567c1 100644 --- a/packages/world-postgres/src/queue.ts +++ b/packages/world-postgres/src/queue.ts @@ -33,6 +33,40 @@ import { z } from 'zod/v4'; import type { PostgresWorldConfig } from './config.js'; import { MessageData } from './message.js'; +/** + * Serialize Graphile Worker log metadata. `JSON.stringify` alone renders an + * `Error` as `{}` because `name`, `message`, `stack`, and `cause` are + * non-enumerable, which is how a failed delivery used to log `"error": {}`. + * Errors are expanded to those fields plus their enumerable properties (such as + * a transport `code`), recursively through `cause` and `AggregateError.errors`. + * An error that has already been expanded is replaced with a marker: a cyclic + * cause chain would otherwise make `JSON.stringify` throw from inside the + * logger, and Graphile has no fallback for a logger that throws. + */ +export function serializeGraphileMeta(meta: unknown): string { + const seen = new WeakSet(); + const expandError = (error: Error): Record => { + if (seen.has(error)) { + return { name: error.name, message: error.message, repeated: true }; + } + seen.add(error); + const expanded: Record = { + ...error, + name: error.name, + message: error.message, + stack: error.stack, + }; + if (error.cause !== undefined) expanded.cause = error.cause; + if (error instanceof AggregateError) expanded.errors = error.errors; + return expanded; + }; + return JSON.stringify( + meta, + (_key, value) => (value instanceof Error ? expandError(value) : value), + 2 + ); +} + function createGraphileLogger() { const isJsonMode = () => process.env.WORKFLOW_JSON_MODE === '1'; const isVerbose = () => Boolean(process.env.DEBUG); @@ -43,7 +77,7 @@ function createGraphileLogger() { const pipe = level === 'error' ? process.stderr : process.stdout; if (meta) { pipe.write( - `[Graphile Worker] ${message} ${JSON.stringify(meta, null, 2)}\n` + `[Graphile Worker] ${message} ${serializeGraphileMeta(meta)}\n` ); } else { pipe.write(`[Graphile Worker] ${message}\n`); diff --git a/packages/world-postgres/test/fixtures/failing-queue.mjs b/packages/world-postgres/test/fixtures/failing-queue.mjs new file mode 100644 index 0000000000..ade8178711 --- /dev/null +++ b/packages/world-postgres/test/fixtures/failing-queue.mjs @@ -0,0 +1,55 @@ +import assert from 'node:assert/strict'; +import { randomUUID } from 'node:crypto'; +import { createServer } from 'node:http'; +import { setTimeout as sleep } from 'node:timers/promises'; +import { getQueueTopicPrefix } from '@workflow/world'; +import { Pool } from 'pg'; +import { createQueue } from '../../dist/queue.js'; + +assert(process.env.WORKFLOW_POSTGRES_URL, 'WORKFLOW_POSTGRES_URL is required'); +const connectionString = process.env.WORKFLOW_POSTGRES_URL; +const pool = new Pool({ connectionString, max: 4 }); +const server = createServer(async (request, response) => { + await request.toArray(); + response.writeHead(503, { 'content-type': 'text/plain' }); + response.end('test queue delivery failure'); +}); +await new Promise((resolve, reject) => { + server.once('error', reject); + server.listen(0, '127.0.0.1', resolve); +}); +const address = server.address(); +assert(address && typeof address !== 'string', 'Expected a TCP server'); +process.env.WORKFLOW_LOCAL_BASE_URL = `http://127.0.0.1:${address.port}`; +const queue = createQueue( + { connectionString, queueConcurrency: 1, applicationManagedShutdown: true }, + pool +); +try { + const { messageId } = await queue.queue( + `${getQueueTopicPrefix('workflow')}test`, + { runId: `run_${randomUUID()}` } + ); + const deadline = Date.now() + 10_000; + for (;;) { + const jobs = await pool.query( + 'SELECT attempts, last_error, locked_at FROM graphile_worker._private_jobs WHERE key = $1', + [messageId] + ); + const job = jobs.rows[0]; + if (job && job.last_error !== null && job.locked_at === null) { + assert.equal(job.attempts, 1); + break; + } + assert(Date.now() < deadline, 'Queue delivery did not fail'); + await sleep(20); + } +} finally { + await queue.close(); + await pool.query('TRUNCATE graphile_worker._private_jobs'); + await pool.end(); + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + server.closeAllConnections(); + }); +} diff --git a/packages/world-postgres/test/queue-logging.test.ts b/packages/world-postgres/test/queue-logging.test.ts new file mode 100644 index 0000000000..bc27fd9dc4 --- /dev/null +++ b/packages/world-postgres/test/queue-logging.test.ts @@ -0,0 +1,70 @@ +import { execFile } from 'node:child_process'; +import { fileURLToPath } from 'node:url'; +import { promisify } from 'node:util'; +import { PostgreSqlContainer } from '@testcontainers/postgresql'; +import { afterAll, beforeAll, describe, expect, test } from 'vitest'; + +const run = promisify(execFile); + +/** + * What Graphile Worker writes to stderr when a delivery fails, captured from a + * real worker in a subprocess (`fixtures/failing-queue.mjs`, which imports the + * built `dist/queue.js`). The serializer itself is unit-tested in + * `src/queue-logging.test.ts`; this pins down that Graphile hands the logger + * the `Error` in `meta` and that the two output controls still hold. + */ +describe('Postgres queue error logs (integration)', () => { + if (process.platform === 'win32') { + test.skip('skipped on Windows since it relies on a docker container', () => {}); + return; + } + + let container: Awaited>; + + beforeAll(async () => { + container = await new PostgreSqlContainer('postgres:15-alpine').start(); + }, 120_000); + + afterAll(async () => { + await container.stop(); + }); + + async function failDelivery(jsonMode: boolean) { + return await run( + process.execPath, + [fileURLToPath(new URL('./fixtures/failing-queue.mjs', import.meta.url))], + { + env: { + ...process.env, + DEBUG: '', + WORKFLOW_JSON_MODE: jsonMode ? '1' : '0', + WORKFLOW_POSTGRES_URL: container.getConnectionUri(), + }, + timeout: 15_000, + } + ); + } + + test('preserves the delivery error and stack in worker metadata', async () => { + const { stdout, stderr } = await failDelivery(false); + expect(stdout).toBe(''); + expect(stderr).toContain('[Graphile Worker] Failed task'); + const metadataStart = stderr.indexOf('{\n'); + expect(metadataStart).toBeGreaterThan(-1); + const metadata = JSON.parse(stderr.slice(metadataStart)); + expect(metadata.error).toMatchObject({ + name: 'Error', + message: + '[postgres world] Queue execution failed (503): test queue delivery failure', + stack: expect.stringContaining( + 'Error: [postgres world] Queue execution failed (503): test queue delivery failure' + ), + }); + }); + + test('keeps worker output suppressed in CLI JSON mode', async () => { + const { stdout, stderr } = await failDelivery(true); + expect(stdout).toBe(''); + expect(stderr).toBe(''); + }); +});