diff --git a/cloud/apps/relay/src/cell-inventory-lock-contention.test.ts b/cloud/apps/relay/src/cell-inventory-lock-contention.test.ts index dcc63e8c02d..4c89315ec3d 100644 --- a/cloud/apps/relay/src/cell-inventory-lock-contention.test.ts +++ b/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()) } } })) diff --git a/cloud/apps/relay/src/database-postgres-timeout.test.ts b/cloud/apps/relay/src/database-postgres-timeout.test.ts index adc27cc451c..2922ea3a4d6 100644 --- a/cloud/apps/relay/src/database-postgres-timeout.test.ts +++ b/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) { diff --git a/cloud/apps/relay/src/database.ts b/cloud/apps/relay/src/database.ts index 2b3cebbd89c..005716c0657 100644 --- a/cloud/apps/relay/src/database.ts +++ b/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() diff --git a/cloud/apps/relay/src/postgres-checked-out-client-error.test.ts b/cloud/apps/relay/src/postgres-checked-out-client-error.test.ts new file mode 100644 index 00000000000..871b013cf96 --- /dev/null +++ b/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 = [] + 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) + }) +}) diff --git a/cloud/apps/relay/src/postgres-pool-pressure.test.ts b/cloud/apps/relay/src/postgres-pool-pressure.test.ts index 2e020bf43fb..4a59dbfef97 100644 --- a/cloud/apps/relay/src/postgres-pool-pressure.test.ts +++ b/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, diff --git a/cloud/apps/relay/src/postgres-pool-pressure.ts b/cloud/apps/relay/src/postgres-pool-pressure.ts index ab0f188067f..afeab79f5a1 100644 --- a/cloud/apps/relay/src/postgres-pool-pressure.ts +++ b/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): Promise { + 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 { diff --git a/cloud/apps/relay/src/postgres-query-failure.test.ts b/cloud/apps/relay/src/postgres-query-failure.test.ts index c0dca423bab..25a3d7f7900 100644 --- a/cloud/apps/relay/src/postgres-query-failure.test.ts +++ b/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() {} }