[world-postgres] Keep a dropped connection in the World's own pool from ending the process

.changeset/world-postgres-pool-error-listener.md

@@ -0,0 +1,5 @@
+---
+"@workflow/world-postgres": patch
+---
+
+Keep a checked-out client in the World's own `pg` pool from ending the process when the database drops its connection.

docs/content/docs/v5/configuration/worlds.mdx

@@ -146,6 +146,7 @@ The Postgres World is a self-hosted durable backend for long-running server proc
 - Environment variable: none
 - Default: new `pg.Pool`
 - Existing `pg.Pool` to use instead of constructing one from `connectionString`.
+- The World attaches no `'error'` listeners to a pool you pass. Listen on the pool and on each client it connects, or a connection the server ends will end the process. See [passing a `pg.Pool`](/worlds/postgres#programmatic-configuration).
 
 ### `jobPrefix`
 

docs/content/worlds/v5/postgres.mdx

@@ -320,14 +320,31 @@ export default createWorld({
 });
 ```
 
-You can also pass an existing `pg.Pool` as `pool` instead of a connection string. On a pool it creates, the World logs and drops an idle connection that the database closes, for example on a restart. A pool you pass needs its own `pool.on("error", ...)` listener. Without one, Node.js throws the error as an uncaught exception and the process exits.
-
 Options passed to `createWorld()` take precedence over the environment variables above. Export the World from a module and point `WORKFLOW_TARGET_WORLD` at that module:
 
 ```bash title=".env"
 WORKFLOW_TARGET_WORLD="./my-world.ts"
 ```
 
+You can also pass an existing `pg.Pool` as `pool` instead of a connection string. The World then leaves that pool's `'error'` events to you. A connection the server ends, in a restart, a failover or an idle timeout, emits one, and an `'error'` event with no listener ends the process. Listen on each client the pool connects as well as on the pool: pg-pool listens only on idle clients, and the World holds a checked-out client across a transaction's statements.
+
+{/*@skip-typecheck: requires pg, which the docs type-check does not install*/}
+
+```typescript title="my-world.ts" lineNumbers
+import { createWorld } from "@workflow/world-postgres";
+import { Pool } from "pg";
+
+const pool = new Pool({ connectionString: process.env.DATABASE_URL });
+pool.on("connect", (client) =>
+  client.on("error", (error) => console.error(error.message))
+);
+pool.on("error", () => {}); // An idle client's error, logged just above
+
+export default createWorld({ pool });
+```
+
+A pool the World creates from `connectionString` already has these listeners.
+
 ### Queue consumers other than the Workflow SDK
 
 The worker delivers every job over HTTP, and by default to the Workflow SDK's flow route, `/.well-known/workflow/v1/flow`. A library that consumes messages of its own on the same World, such as a durable agent harness, uses its own queue prefix (`__{namespace}_wkf_workflow_`) and registers where those go: `world.registerRoute(prefix, "/api/agent/queue")` to deliver them to its own route, or `world.registerHandler(prefix, handler)` to deliver them inside the process that took the job, without HTTP. A job keeps the prefix it was enqueued under and the route its producer had registered, so a worker in another process delivers it to the same place. Register before calling `start()`, which re-enqueues each active run to the queue namespace it recorded at creation (`executionContext.queueNamespace`).

packages/world-postgres/README.md

@@ -151,9 +151,17 @@ const world = createWorld({
 // Or pass an existing pg.Pool (shared with your app Drizzle, etc.); `world.close()` will not end it.
 import { Pool } from "pg";
 const pool = new Pool({ connectionString: process.env.DATABASE_URL });
+// The World leaves a pool you pass to you: listen for errors, or a connection
+// the server ends (a restart, a failover, an idle timeout) ends the process.
+pool.on("connect", (client) =>
+  client.on("error", (error) => console.error(error.message))
+);
+pool.on("error", () => {}); // An idle client's error, logged just above
 const worldFromPool = createWorld({ pool });
 ```
 
+A connection the server ends emits an `'error'` event, and an `'error'` event with no listener ends the process. When `createWorld()` creates the pool, it listens on the pool and on every client it connects, and logs one `[world-postgres] Pooled PostgreSQL connection lost` line per lost connection. It doesn't add listeners to a pool you pass, so add both yourself, as above: pg-pool listens only on idle clients, and the World holds a checked-out client across a transaction's statements.
+
 ### Application-managed shutdown
 
 By default, Graphile Worker responds automatically when the application is asked to shut down. If your application already coordinates shutdown, set `WORKFLOW_POSTGRES_APPLICATION_MANAGED_SHUTDOWN=1` when selecting the package with `WORKFLOW_TARGET_WORLD`, or set `applicationManagedShutdown: true` when calling `createWorld()` directly. Await `world.close()` from your shutdown path so Graphile Worker cannot terminate the process as soon as its queue stops, before your application finishes closing dependent resources:
@@ -181,7 +189,7 @@ An aborted HTTP request does not guarantee that its server-side handler stopped,
 | ------------------ | --------- | -------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------- |
 | `connectionString` | `string`  | `process.env.WORKFLOW_POSTGRES_URL`, `process.env.DATABASE_URL`, or `'postgres://world:world@localhost:5432/world'` | Used only when `pool` is omitted, to construct an internal pool                                      |
 | `maxPoolSize`      | `number`  | `process.env.WORKFLOW_POSTGRES_MAX_POOL_SIZE` or `pg.Pool` default (`10`)              | Optional. Sets the internal `pg.Pool` max size when `createWorld()` creates the pool                |
-| `pool`             | `pg.Pool` | Not applicable                                                                         | Optional. When set, used for Drizzle, Graphile Worker, and stream writes. `world.close()` does not end it. |
+| `pool`             | `pg.Pool` | Not applicable                                                                         | Optional. When set, used for Drizzle, Graphile Worker, and stream writes. `world.close()` does not end it, and the World adds no `'error'` listeners to it (see above). |
 | `jobPrefix`        | `string`  | `process.env.WORKFLOW_POSTGRES_JOB_PREFIX`                                             | Optional prefix for queue job names                                                                  |
 | `queueConcurrency` | `number`  | `50`                                                                                   | Number of concurrent active step executions per process. Must be high enough to cover any parent→child workflow polling in flight because each `Run#returnValue` await holds a worker slot until the child run terminates. |
 | `pollInterval`     | `number`  | `500`                                                                                  | Milliseconds between idle job fetches per worker. |

packages/world-postgres/src/connection-errors.ts

@@ -0,0 +1,49 @@
+import type { Pool, PoolClient } from 'pg';
+
+/**
+ * Keep a connection the server drops from ending the process.
+ *
+ * A restart or failover, `idle_session_timeout`, `pg_terminate_backend`, or a
+ * proxy closing idle sockets can end a connection at any time, and `pg`
+ * reports it as an `'error'` event on the client. pg-pool forwards an idle
+ * client's error to the pool, but detaches its own listener while a client is
+ * checked out, and Drizzle transactions and the invocation mailbox hold
+ * clients across statements. An `EventEmitter` with no `'error'` listener
+ * throws, so either case is an uncaught exception. Graphile Worker attaches
+ * listeners like these only while its runner is up, and a World that is only
+ * read from never starts one.
+ *
+ * Logging is all that is left to do: pg-pool discards a failed client when it
+ * is released, and a checked-out client's next query rejects, so its caller
+ * still sees the error.
+ *
+ * Only for a pool the World creates. A pool the caller passes in is theirs to
+ * listen on.
+ */
+export function handleConnectionErrors(
+  pool: Pool,
+  isClosing: () => boolean
+): void {
+  const log = (error: Error & { code?: string }) => {
+    // Expected while the World shuts its pool down.
+    if (isClosing()) return;
+    // Never the error itself: pg-pool sets `error.client`, whose connection
+    // parameters include the password.
+    console.warn(
+      `[world-postgres] Pooled PostgreSQL connection lost${error.code ? ` (${error.code})` : ''}: ${error.message}`
+    );
+  };
+  pool.on('connect', (client: PoolClient) => {
+    let lost = false;
+    client.on('error', (error) => {
+      // A client nobody has ended reports one drop twice: the server's error,
+      // then "Connection terminated unexpectedly" as the socket closes.
+      if (lost) return;
+      lost = true;
+      log(error);
+    });
+  });
+  // pg-pool re-emits an idle client's error here after the listener above has
+  // logged it, so this one only keeps the pool from throwing.
+  pool.on('error', () => {});
+}

packages/world-postgres/src/index.ts

@@ -2,6 +2,7 @@ import type { Queue, Storage, World } from '@workflow/world';
 import { mintedSpecVersion, reenqueueActiveRuns } from '@workflow/world';
 import { Pool } from 'pg';
 import type { PostgresWorldConfig } from './config.js';
+import { handleConnectionErrors } from './connection-errors.js';
 import { createClient, type Drizzle } from './drizzle/index.js';
 import { createQueue } from './queue.js';
 import {
@@ -62,7 +63,9 @@ export function createWorld(
   } {
   const maxPoolSize =
     config.maxPoolSize ?? getPositiveIntEnv('WORKFLOW_POSTGRES_MAX_POOL_SIZE');
-  const pool = config.pool || createOwnedPool(config, maxPoolSize);
+  let closing = false;
+  const pool =
+    config.pool || createOwnedPool(config, maxPoolSize, () => closing);
 
   const drizzle = createClient(pool);
   const queue = createQueue(config, pool);
@@ -104,6 +107,7 @@ export function createWorld(
       );
     },
     async close() {
+      closing = true;
       await queue.close();
       await streamer.close();
       await runStatusListener.close();
@@ -117,27 +121,26 @@ export function createWorld(
 /**
  * The pool the World creates when the caller passes none.
  *
- * It listens for `error`: pg-pool re-emits an idle client's connection error
- * (a database restart, `pg_terminate_backend`) on the pool, and an
- * EventEmitter with no `error` listener throws it as an uncaught exception.
+ * It listens for connection errors on the pool and on every client it
+ * connects (see `handleConnectionErrors`): a database restart or
+ * `pg_terminate_backend` emits `error` on an idle client, which pg-pool
+ * re-emits on the pool, and on a checked-out client, which pg-pool does not
+ * listen on at all. With no listener either is an uncaught exception.
  * Graphile Worker listens only while the queue runs, and `pool.end()`
  * resolves before its idle clients have closed, so without this a backend
- * terminated during or after `close()` ends the process. The dropped client is
- * already out of the pool, and the next query opens a new one.
+ * terminated during or after `close()` ends the process. The dropped client
+ * leaves the pool, and the next query opens a new one.
  */
 function createOwnedPool(
   config: PostgresWorldConfig,
-  maxPoolSize: number | undefined
+  maxPoolSize: number | undefined,
+  isClosing: () => boolean
 ): Pool {
   const pool = new Pool({
     connectionString: config.connectionString || getDefaultConnectionString(),
     ...(maxPoolSize !== undefined ? { max: maxPoolSize } : {}),
   });
-  pool.on('error', (error) => {
-    console.warn(
-      `[world-postgres] idle database connection closed: ${error.message}`
-    );
-  });
+  handleConnectionErrors(pool, isClosing);
   return pool;
 }
 

packages/world-postgres/src/reenqueue.test.ts

@@ -371,6 +371,23 @@ describe('re-enqueue active runs on start', () => {
     expect(internalPool?.end).toHaveBeenCalledOnce();
   });
 
+  it('listens for connection errors on a pool it creates, not on a caller-owned one', () => {
+    createWorld({ connectionString: 'postgres://test' });
+    const internalPool = vi.mocked(Pool).mock.results.at(-1)?.value;
+    expect(internalPool?.on).toHaveBeenCalledWith(
+      'connect',
+      expect.any(Function)
+    );
+    expect(internalPool?.on).toHaveBeenCalledWith(
+      'error',
+      expect.any(Function)
+    );
+
+    const callerPool = { ...pool, on: vi.fn() };
+    createWorld({ pool: callerPool });
+    expect(callerPool.on).not.toHaveBeenCalled();
+  });
+
   it('does not close a caller-owned pool', async () => {
     const world = createWorld({ pool });
     await world.start();

packages/world-postgres/test/connection-errors.test.ts

@@ -0,0 +1,87 @@
+import { execFile, execSync } from 'node:child_process';
+import { randomUUID } from 'node:crypto';
+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);
+
+/**
+ * A connection the server drops must not end the process. Each case runs in a
+ * subprocess (`fixtures/dropped-connections.mjs`, which imports the built
+ * `dist/`), because a missing `'error'` listener surfaces as an uncaught
+ * exception: the process exits non-zero instead of printing its result.
+ */
+describe('Postgres dropped connections (integration)', () => {
+  if (process.platform === 'win32') {
+    test.skip('skipped on Windows since it relies on a docker container', () => {});
+    return;
+  }
+
+  // Unique, so the test can tell a leaked password from any other text.
+  const password = randomUUID();
+  let container: Awaited<ReturnType<PostgreSqlContainer['start']>>;
+
+  beforeAll(async () => {
+    container = await new PostgreSqlContainer('postgres:15-alpine')
+      .withPassword(password)
+      .start();
+    execSync('pnpm db:push', {
+      stdio: 'ignore',
+      cwd: process.cwd(),
+      env: {
+        ...process.env,
+        WORKFLOW_POSTGRES_URL: container.getConnectionUri(),
+      },
+    });
+  }, 120_000);
+
+  afterAll(async () => {
+    await container.stop();
+  });
+
+  async function dropConnections(mode: 'world' | 'checked-out') {
+    const { stdout, stderr } = await run(
+      process.execPath,
+      [
+        fileURLToPath(
+          new URL('./fixtures/dropped-connections.mjs', import.meta.url)
+        ),
+        mode,
+      ],
+      {
+        env: {
+          ...process.env,
+          WORKFLOW_POSTGRES_URL: container.getConnectionUri(),
+        },
+        timeout: 15_000,
+      }
+    );
+    return { result: JSON.parse(stdout), stderr };
+  }
+
+  test("survives losing idle connections in a World's own pool", async () => {
+    const { result, stderr } = await dropConnections('world');
+    // The pool's read connections, plus the streamer's LISTEN client.
+    expect(result.dropped).toBeGreaterThanOrEqual(2);
+    expect(result.readAfterDrop).toBe(true);
+    expect(stderr).toContain(
+      '[world-postgres] Pooled PostgreSQL connection lost (57P01): terminating connection due to administrator command'
+    );
+    expect(stderr).not.toContain(password);
+  });
+
+  test('survives losing a checked-out client between statements', async () => {
+    const { result, stderr } = await dropConnections('checked-out');
+    expect(result).toEqual({ dropped: 1, heldQueryRejected: true });
+    // One line per lost connection, though pg reports this one twice.
+    expect(
+      stderr.match(/\[world-postgres\] Pooled PostgreSQL connection lost/g)
+    ).toEqual(['[world-postgres] Pooled PostgreSQL connection lost']);
+    expect(stderr).toContain(
+      '[world-postgres] Pooled PostgreSQL connection lost (57P01)'
+    );
+    expect(stderr).not.toContain(password);
+  });
+});

packages/world-postgres/test/fixtures/dropped-connections.mjs

@@ -0,0 +1,60 @@
+// Drops every PostgreSQL connection this process holds, then prints what
+// still works. A dropped connection with no `'error'` listener ends the
+// process instead, so the test reads a crash as a non-zero exit.
+// argv[2]: `world` (a World's own pool, never started) or `checked-out`.
+import assert from 'node:assert/strict';
+import { randomUUID } from 'node:crypto';
+import { setTimeout as sleep } from 'node:timers/promises';
+import pg from 'pg';
+import { handleConnectionErrors } from '../../dist/connection-errors.js';
+import { createWorld } from '../../dist/index.js';
+
+assert(process.env.WORKFLOW_POSTGRES_URL, 'WORKFLOW_POSTGRES_URL is required');
+const name = `dropped_${randomUUID().replaceAll('-', '')}`;
+const url = new URL(process.env.WORKFLOW_POSTGRES_URL);
+url.searchParams.set('application_name', name);
+
+const admin = new pg.Client({
+  connectionString: process.env.WORKFLOW_POSTGRES_URL,
+});
+admin.on('error', () => {});
+await admin.connect();
+// What a database restart does to every connection at once.
+const dropConnections = async () =>
+  (
+    await admin.query(
+      'SELECT count(pg_terminate_backend(pid))::int AS n FROM pg_stat_activity WHERE application_name = $1',
+      [name]
+    )
+  ).rows[0].n;
+
+if (process.argv[2] === 'world') {
+  // Reads only, like an observability UI: no start(), so no Graphile Worker
+  // listeners on the pool.
+  const world = createWorld({ connectionString: url.href });
+  const read = () => world.runs.list({ pagination: { limit: 1 } });
+  await Promise.all([read(), read(), read()]);
+  const dropped = await dropConnections();
+  await sleep(500);
+  await read();
+  await world.close();
+  console.log(JSON.stringify({ dropped, readAfterDrop: true }));
+} else {
+  // The way createWorld() sets up its own pool. A transaction holds its
+  // client across statements, and pg-pool listens only on idle clients.
+  const pool = new pg.Pool({ connectionString: url.href });
+  handleConnectionErrors(pool, () => false);
+  const held = await pool.connect();
+  await held.query('BEGIN');
+  const dropped = await dropConnections();
+  await sleep(500);
+  const heldQueryRejected = await held.query('SELECT 1').then(
+    () => false,
+    () => true
+  );
+  held.release();
+  await pool.query('SELECT 1');
+  await pool.end();
+  console.log(JSON.stringify({ dropped, heldQueryRejected }));
+}
+await admin.end();