fix(world-postgres): preserve queue error metadata

.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.

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 },
+      },
+    });
+  });
+});

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<object>();
+  const expandError = (error: Error): Record<string, unknown> => {
+    if (seen.has(error)) {
+      return { name: error.name, message: error.message, repeated: true };
+    }
+    seen.add(error);
+    const expanded: Record<string, unknown> = {
+      ...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`);

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

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<ReturnType<PostgreSqlContainer['start']>>;
+
+  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('');
+  });
+});