fix(relay): stop a terminated checked-out PostgreSQL client from killing the cell (#21840)

pg-pool removes its own `error` listener when it hands a client out
(pg-pool@3.14.0 index.js:344) and only reattaches it in `_release`
(index.js:385). Between acquire and release the client therefore has no
`error` listener, so when Cloud SQL terminates that session mid-statement
the emit becomes an unhandled 'error' event and the process exits.
`absorbPostgresIdleClientErrors` cannot see it: pg-pool routes to
`pool.on('error')` only from the idle listener.

Attach a per-checkout `error` listener in the one seam every relay
checkout passes through, log a single warn line, and release the client
with the error so pg-pool destroys it instead of pooling a dead
connection. The listener is removed on release so it cannot accumulate.
The in-flight query still rejects, so existing failure reporting and the
transaction retry ladder are unchanged.

Claude-Session: https://claude.ai/session/ced32ebb-7155-4413-adad-1eccd14c2010

cloud/apps/relay/src/cell-inventory-lock-contention.test.ts

@@ -8,6 +8,14 @@ const fakes = vi.hoisted(() => ({
     return { rows: [], rowCount: 0 }
   }),
   release: vi.fn(),
+  // A real pooled client is an EventEmitter, and the acquire path attaches an
+  // `error` listener to it before handing it to the caller.
+  client: () => ({
+    query: fakes.query,
+    release: fakes.release,
+    on: vi.fn(),
+    removeListener: vi.fn()
+  }),
   end: vi.fn(async () => undefined)
 }))
 
@@ -19,7 +27,7 @@ vi.mock('pg', () => ({
       waitingCount = 0
       end = fakes.end
       on = vi.fn()
-      connect = vi.fn(async () => ({ query: fakes.query, release: fakes.release }))
+      connect = vi.fn(async () => fakes.client())
     }
   }
 }))

cloud/apps/relay/src/database-postgres-timeout.test.ts

@@ -8,6 +8,14 @@ const fakes = vi.hoisted(() => ({
   lifecycle: [] as string[],
   query: vi.fn(async (_sql: string) => ({ rows: [], rowCount: 0 })),
   release: vi.fn(),
+  // A real pooled client is an EventEmitter, and the acquire path attaches an
+  // `error` listener to it before handing it to the caller.
+  client: () => ({
+    query: fakes.query,
+    release: fakes.release,
+    on: vi.fn(),
+    removeListener: vi.fn()
+  }),
   end: vi.fn(async () => undefined)
 }))
 
@@ -18,7 +26,7 @@ vi.mock('pg', () => ({
       idleCount = 1
       waitingCount = 0
       on = vi.fn()
-      connect = vi.fn(async () => ({ query: fakes.query, release: fakes.release }))
+      connect = vi.fn(async () => fakes.client())
       private readonly label: string
 
       constructor(config: Record<string, unknown>) {

cloud/apps/relay/src/database.ts

@@ -992,7 +992,7 @@ async function waitForPostgresRetry(random: () => number = Math.random): Promise
   await new Promise((resolve) => setTimeout(resolve, delayMs))
 }
 
-class PostgresDatabase implements RelayDatabase {
+export class PostgresDatabase implements RelayDatabase {
   readonly dialect = 'postgres' as const
   private readonly pressure: PostgresPoolPressure
   private readonly holds = new CellInventoryHoldSamples()

cloud/apps/relay/src/postgres-checked-out-client-error.test.ts

@@ -0,0 +1,80 @@
+import { EventEmitter } from 'node:events'
+import { describe, expect, it, vi } from 'vitest'
+import { PostgresDatabase } from './database.js'
+
+// Stands in for a pg client between acquire and release. pg-pool assigns
+// `release` per checkout, which is the property the guard wraps.
+class FakePoolClient extends EventEmitter {
+  readonly released: Array<Error | boolean | undefined> = []
+  readonly statements: string[] = []
+
+  constructor(private readonly respond: (sql: string) => { rows: unknown[]; rowCount: number }) {
+    super()
+  }
+
+  query = vi.fn((sql: string) => {
+    this.statements.push(sql)
+    return Promise.resolve(this.respond(sql))
+  })
+
+  release = (error?: Error | boolean): void => {
+    this.released.push(error)
+  }
+}
+
+function poolOf(client: FakePoolClient) {
+  return { totalCount: 1, idleCount: 0, waitingCount: 0, connect: async () => client }
+}
+
+describe('checked-out PostgreSQL client failure handling', () => {
+  it('crashes the process when nothing listens, which is the bug being fixed', () => {
+    // Node's own contract: this is what killed cell c28 on 2026-09-20 20:18Z.
+    const unguarded = new EventEmitter()
+    expect(() => unguarded.emit('error', new Error('Connection terminated unexpectedly'))).toThrow(
+      'Connection terminated unexpectedly'
+    )
+  })
+
+  it('absorbs the error, rejects the transaction, and releases the client as failed', async () => {
+    const terminated = Object.assign(new Error('Connection terminated unexpectedly'), {
+      code: '57P01'
+    })
+    let listenersWhileCheckedOut = 0
+    const client: FakePoolClient = new FakePoolClient((sql) => {
+      if (sql !== 'SELECT 1') return { rows: [], rowCount: 0 }
+      listenersWhileCheckedOut = client.listenerCount('error')
+      // Cloud SQL terminating the session: the client emits `error` and the
+      // in-flight statement rejects with the same failure.
+      expect(() => client.emit('error', terminated)).not.toThrow()
+      throw terminated
+    })
+    const warning = vi.spyOn(console, 'warn').mockImplementation(() => {})
+    const database = new PostgresDatabase(poolOf(client) as never)
+
+    await expect(
+      database.transaction(async (transaction) => await transaction.query('SELECT 1'))
+    ).rejects.toBe(terminated)
+
+    expect(listenersWhileCheckedOut).toBe(1)
+    expect(client.listenerCount('error')).toBe(0)
+    expect(client.released).toEqual([terminated])
+    expect(client.statements).toEqual(['BEGIN', 'SELECT 1', 'ROLLBACK'])
+    expect(warning).toHaveBeenCalledWith(
+      '[orca-relay] checked-out PostgreSQL client failed: 57P01 Connection terminated unexpectedly'
+    )
+
+    warning.mockRestore()
+  })
+
+  it('releases a healthy client back to the pool with no error', async () => {
+    const client = new FakePoolClient(() => ({ rows: [{ one: 1 }], rowCount: 1 }))
+    const database = new PostgresDatabase(poolOf(client) as never)
+
+    await expect(
+      database.transaction(async (transaction) => await transaction.query('SELECT 1'))
+    ).resolves.toEqual([{ one: 1 }])
+
+    expect(client.released).toEqual([undefined])
+    expect(client.listenerCount('error')).toBe(0)
+  })
+})

cloud/apps/relay/src/postgres-pool-pressure.test.ts

@@ -1,3 +1,4 @@
+import { EventEmitter } from 'node:events'
 import { describe, expect, it, vi } from 'vitest'
 import { PostgresPoolPressure } from './postgres-pool-pressure.js'
 
@@ -32,7 +33,8 @@ describe('PostgreSQL pool pressure', () => {
 
     now = 2_250
     pool.waitingCount--
-    resolveConnection({ query: vi.fn(), release: vi.fn() })
+    // An EventEmitter because the acquire path now attaches an `error` listener.
+    resolveConnection(Object.assign(new EventEmitter(), { query: vi.fn(), release: vi.fn() }))
     await pending
     expect(pressure.consumeCounts()).toMatchObject({
       databasePoolWaiting: 0,

cloud/apps/relay/src/postgres-pool-pressure.ts

@@ -30,6 +30,10 @@ function errorMessage(error: unknown): string {
   return String((error as { message?: unknown } | null)?.message)
 }
 
+function errorCode(error: unknown): string {
+  return String((error as { code?: unknown } | null)?.code)
+}
+
 function isPostgresPoolAcquireFailure(error: unknown): boolean {
   return typeof error === 'object' && error !== null && poolAcquireFailures.has(error)
 }
@@ -131,12 +135,45 @@ export class PostgresPoolPressure {
 }
 
 async function markedAcquire(connection: Promise<pg.PoolClient>): Promise<pg.PoolClient> {
+  let client: pg.PoolClient
   try {
-    return await connection
+    client = await connection
   } catch (error) {
     if (typeof error === 'object' && error !== null) poolAcquireFailures.add(error)
     throw error
   }
+  return guardCheckedOutClient(client)
+}
+
+// pg-pool strips its own `error` listener when it hands a client out
+// (pg-pool@3.14.0 index.js:344) and only reattaches it in `_release`
+// (index.js:385), so a checked-out client has no `error` listener at all. A
+// backend that terminates that session mid-statement therefore emits `error`
+// with nothing listening, which is an unhandled 'error' event and kills the
+// process. `pool.on('error')` cannot cover this: pg-pool routes there only from
+// the idle listener. Every relay checkout awaits this function, so it is the
+// one seam that sees them all.
+function guardCheckedOutClient(client: pg.PoolClient): pg.PoolClient {
+  let failure: Error | undefined
+  const onError = (error: Error) => {
+    failure ??= error
+    // Printable unlike the idle path: a checked-out client is past the
+    // handshake, so its error carries no connection string.
+    console.warn(
+      `[orca-relay] checked-out PostgreSQL client failed: ${errorCode(error)} ${errorMessage(error)}`
+    )
+  }
+  client.on('error', onError)
+
+  // pg-pool assigns a fresh `release` on every acquire, so this never stacks.
+  const release = client.release.bind(client)
+  client.release = (releaseError?: Error | boolean) => {
+    client.removeListener('error', onError)
+    // Passing the error makes pg-pool destroy the client instead of returning a
+    // dead connection to the pool for the next caller to trip over.
+    release(releaseError ?? failure)
+  }
+  return client
 }
 
 export function emptyPostgresPoolPressureCounts(): PostgresPoolPressureCounts {

cloud/apps/relay/src/postgres-query-failure.test.ts

@@ -3,7 +3,15 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 const fakes = vi.hoisted(() => ({
   connectError: undefined as unknown,
   query: vi.fn(async (_sql: string, _params?: unknown[]) => ({ rows: [], rowCount: 0 })),
-  release: vi.fn()
+  release: vi.fn(),
+  // A real pooled client is an EventEmitter, and the acquire path attaches an
+  // `error` listener to it before handing it to the caller.
+  client: () => ({
+    query: fakes.query,
+    release: fakes.release,
+    on: vi.fn(),
+    removeListener: vi.fn()
+  })
 }))
 
 vi.mock('pg', () => ({
@@ -15,7 +23,7 @@ vi.mock('pg', () => ({
       on = vi.fn()
       async connect() {
         if (fakes.connectError) throw fakes.connectError
-        return { query: fakes.query, release: fakes.release }
+        return fakes.client()
       }
       async end() {}
     }