diff --git a/cloud/apps/relay/src/assignment-store.ts b/cloud/apps/relay/src/assignment-store.ts index 8a59931ee08..f953343f3df 100644 --- a/cloud/apps/relay/src/assignment-store.ts +++ b/cloud/apps/relay/src/assignment-store.ts @@ -73,6 +73,7 @@ import { import type { RelayCellConfig } from './config.js' import type { CellLockHoldSite } from './cell-inventory-hold-samples.js' import { isPostgresPoolConnectFailure } from './postgres-pool-pressure.js' +import { isPostgresReadTimeout } from './relay-database-rejection-fence.js' import type { RelayDatabase, RelayLockOptions, @@ -4096,9 +4097,11 @@ export class RelayAssignmentStore { message: String((error as { message?: unknown }).message) }) ) - // No connection at all: per-host statements would only queue up behind the - // same saturated pool, multiplying the load that starved this flush. - if (isPostgresPoolConnectFailure(error)) return outcomes.fill('database_error') + // No connection at all, or a lost reply: per-host statements would only queue up + // behind the same saturated pool or stalled server, multiplying the load. + if (isPostgresPoolConnectFailure(error) || isPostgresReadTimeout(error)) { + return outcomes.fill('database_error') + } await Promise.all( ordered.map(async (row) => { try { diff --git a/cloud/apps/relay/src/cell-admission-startup.ts b/cloud/apps/relay/src/cell-admission-startup.ts index f10f756ecd6..677980f27b5 100644 --- a/cloud/apps/relay/src/cell-admission-startup.ts +++ b/cloud/apps/relay/src/cell-admission-startup.ts @@ -28,6 +28,8 @@ export async function reconcileCellAdmissionAtStartup( // Admission is operator/director state. A new worker must not enable itself // before its distinct candidate has passed production preflight. if (config.role === 'cell') return + // Retries every transient failure, a lost reply included: the reconcile is upserts and + // limit-change deletes that converge on the same rows if a lost COMMIT had landed. await retryTransientDatabaseStartup( async () => await assignments.reconcileCellsAtStartup(config.cells), STARTUP_RECONCILE_RETRY, diff --git a/cloud/apps/relay/src/database-postgres-timeout.test.ts b/cloud/apps/relay/src/database-postgres-timeout.test.ts index 2922ea3a4d6..61042aeb425 100644 --- a/cloud/apps/relay/src/database-postgres-timeout.test.ts +++ b/cloud/apps/relay/src/database-postgres-timeout.test.ts @@ -46,6 +46,7 @@ vi.mock('pg', () => ({ import { openRelayDatabase, POSTGRES_SCHEMA_MIGRATIONS, + relayPostgresReadTimeoutMarginMs, relayPostgresStatementTimeoutMs } from './database.js' import { applyPostgresSchema } from './postgres-schema-startup.js' @@ -89,10 +90,15 @@ describe('PostgreSQL relay deadlines', () => { application_name: 'orca-relay/director/director', connectionTimeoutMillis: 2_000, statement_timeout: 5_000, + // A lost reply ends at the statement deadline plus 10 s, on this pool only. + query_timeout: 15_000, lock_timeout: 1_000, - idle_in_transaction_session_timeout: 5_000 + idle_in_transaction_session_timeout: 5_000, + options: '-c idle_session_timeout=30000' }) ]) + expect(fakes.configs[0]).not.toHaveProperty('query_timeout') + expect(fakes.configs[0]).not.toHaveProperty('options') await database.close() }) @@ -160,7 +166,7 @@ describe('PostgreSQL relay deadlines', () => { expect(fakes.configs).toEqual([ expect.objectContaining({ statement_timeout: 0 }), - expect.objectContaining({ statement_timeout: 2_500 }) + expect.objectContaining({ statement_timeout: 2_500, query_timeout: 12_500 }) ]) await database.close() }) @@ -174,6 +180,18 @@ describe('PostgreSQL relay deadlines', () => { } ) + it('takes the read timeout margin from the environment, defaulting to 10 s', () => { + expect(relayPostgresReadTimeoutMarginMs({})).toBe(10_000) + expect( + relayPostgresReadTimeoutMarginMs({ ORCA_RELAY_POSTGRES_READ_TIMEOUT_MARGIN_MS: '4000' }) + ).toBe(4_000) + for (const value of ['0', '-1', '2.5', 'soon']) { + expect(() => + relayPostgresReadTimeoutMarginMs({ ORCA_RELAY_POSTGRES_READ_TIMEOUT_MARGIN_MS: value }) + ).toThrow('invalid_read_timeout_margin') + } + }) + it.each([undefined, ''])('defaults to 5s when the environment says %s', (value) => { expect( relayPostgresStatementTimeoutMs( diff --git a/cloud/apps/relay/src/database-transient-error.test.ts b/cloud/apps/relay/src/database-transient-error.test.ts index 3f9b5689c9e..5f2b895e01d 100644 --- a/cloud/apps/relay/src/database-transient-error.test.ts +++ b/cloud/apps/relay/src/database-transient-error.test.ts @@ -1,23 +1,56 @@ import { describe, expect, it, vi } from 'vitest' -import { isRelayDatabaseTransientError } from './database.js' +import { isRelayDatabaseTransientError, PostgresDatabase } from './database.js' import { PostgresPoolPressure } from './postgres-pool-pressure.js' // The only way to mark an error as an acquire failure is to fail a real // acquire, so the gated cases go through the pressure wrapper the pool uses. -async function failedAcquire(message: string): Promise { +async function failedAcquire(failure: string | Error): Promise { const pool = { totalCount: 0, idleCount: 0, waitingCount: 0, connect: vi.fn(async () => { - throw new Error(message) + throw typeof failure === 'string' ? new Error(failure) : failure }) } return await new PostgresPoolPressure(pool as never).connect().catch((error: unknown) => error) } +async function failedQuery(failure: Error): Promise { + const client = { + query: vi.fn(async () => { + throw failure + }), + release: vi.fn(), + on: vi.fn(), + removeListener: vi.fn() + } + const database = new PostgresDatabase({ + totalCount: 1, + idleCount: 0, + waitingCount: 0, + connect: async () => client + } as never) + vi.spyOn(console, 'warn').mockImplementation(() => {}) + return await database.query('SELECT 1').catch((error: unknown) => error) +} + describe('relay database transient errors', () => { - it.each(['40P01', '40001', '55P03', '57014', '53300', '57P03', '08001', '08006', '25P03'])( + it.each([ + '40P01', + '40001', + '55P03', + '57014', + '53300', + '57P01', + '57P02', + '57P03', + '08000', + '08001', + '08003', + '08006', + '25P03' + ])( 'classifies PostgreSQL code %s as retryable overload', (code) => { expect(isRelayDatabaseTransientError({ code })).toBe(true) @@ -40,15 +73,50 @@ describe('relay database transient errors', () => { ).toBe(true) }) - it('classifies an early-ended socket only when it ended during the acquire', async () => { + it('classifies an early-ended socket during the acquire', async () => { expect( isRelayDatabaseTransientError(await failedAcquire('Connection terminated unexpectedly')) ).toBe(true) - // The same message mid-statement leaves the commit outcome unknown, so it - // must stay a hard failure rather than invite a retry. - expect( - isRelayDatabaseTransientError(new Error('Connection terminated unexpectedly')) - ).toBe(false) + }) + + // Staging B1 re-test: a Cloud SQL restart failed the transaction's acquire with `write EPIPE`. + it.each(['EPIPE', 'ECONNRESET', 'ECONNREFUSED', 'ETIMEDOUT', 'EHOSTUNREACH'])( + 'classifies an acquire that failed with %s, but not the bare errno', + async (code) => { + const failure = (): Error => Object.assign(new Error(`connect ${code}`), { code }) + expect(isRelayDatabaseTransientError(await failedAcquire(failure()))).toBe(true) + expect(isRelayDatabaseTransientError(failure())).toBe(false) + } + ) + + it('classifies an EPIPE from the transaction acquire', async () => { + const database = new PostgresDatabase({ + totalCount: 0, + idleCount: 0, + waitingCount: 0, + connect: async () => { + throw Object.assign(new Error('write EPIPE'), { code: 'EPIPE' }) + } + } as never) + const error = await database.transaction(async () => 'unreached').catch((e: unknown) => e) + expect(error).toMatchObject({ code: 'EPIPE' }) + expect(isRelayDatabaseTransientError(error)).toBe(true) + }) + + // Staging B1: pg_terminate_backend mid-request answered /v1/assign with 500. A dropped + // connection is 503 + Retry-After, as 08006 always was; transaction() still never retries it, + // since the commit outcome is unknown. + it('classifies a mid-statement drop only from the database layer', async () => { + const failures = [ + (): Error => new Error('Connection terminated unexpectedly'), + (): Error => Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' }), + (): Error => Object.assign(new Error('write EPIPE'), { code: 'EPIPE' }) + ] + for (const failure of failures) { + expect(isRelayDatabaseTransientError(await failedQuery(failure()))).toBe(true) + // The same errno or message from any other socket is not a database answer. + expect(isRelayDatabaseTransientError(failure())).toBe(false) + } }) it.each([null, undefined, 'a thrown string'])( diff --git a/cloud/apps/relay/src/database.ts b/cloud/apps/relay/src/database.ts index 72830c1eaca..0f219b964e3 100644 --- a/cloud/apps/relay/src/database.ts +++ b/cloud/apps/relay/src/database.ts @@ -14,7 +14,11 @@ import { import { applyPostgresSchema } from './postgres-schema-startup.js' import { POSTGRES_STATEMENT_STATS_MIGRATION } from './postgres-statement-stats.js' import { reportPostgresQueryFailure } from './postgres-query-failure.js' -import { markRelayDatabaseError } from './relay-database-rejection-fence.js' +import { + isPostgresReadTimeout, + isRelayDatabaseLayerError, + markRelayDatabaseError +} from './relay-database-rejection-fence.js' import { CellInventoryHoldSamples, emptyCellInventoryHoldCounts, @@ -890,15 +894,33 @@ class PostgresTransaction implements RelayDatabase { private lockUnavailable = 0 private lockTimeouts = 0 private state: 'open' | 'committed' | 'rolled-back' = 'open' + // A read timeout leaves the statement running with its reply lost: pg keeps it as the + // client's active query, so anything sent next queues behind it, and if it then lands, a + // later COMMIT would commit a statement its caller saw fail. The connection is finished. + private lostReply: Error | undefined constructor(protected readonly client: pg.PoolClient) {} + get poisonedBy(): Error | undefined { + return this.lostReply + } + + private failIfPoisoned(): void { + if (this.lostReply) throw this.lostReply + } + + private noteFailure(error: unknown): void { + markRelayDatabaseError(error) + if (isPostgresReadTimeout(error)) this.lostReply ??= error + } + get open(): boolean { return this.state === 'open' } async commitWithFinal(sql: string, params: unknown[] = []): Promise { this.assertNotCommitted() + this.failIfPoisoned() // Zero rows divides by zero, so the message stops before COMMIT exactly when the write missed. const message = `WITH final_write AS (${inlinePostgresParameters(sql, params, this.client)}) ` + @@ -909,13 +931,13 @@ class PostgresTransaction implements RelayDatabase { // Simple query stops at the first error, so any server error means COMMIT never ran: // retryable codes take the caller's normal rollback-and-retry path. A lost connection // leaves the outcome unknown, and nothing retries it. - markRelayDatabaseError(error) + this.noteFailure(error) if (String((error as { code?: unknown }).code) !== '22012') { rememberPostgresTransactionPhase(error, sql) throw error } await this.client.query('ROLLBACK').catch((rollbackError: unknown) => { - markRelayDatabaseError(rollbackError) + this.noteFailure(rollbackError) throw rollbackError }) this.state = 'rolled-back' @@ -953,11 +975,12 @@ class PostgresTransaction implements RelayDatabase { async query(sql: string, params: unknown[] = []): Promise { this.assertNotCommitted() + this.failIfPoisoned() try { const result = await this.client.query(postgresSql(sql), params) return returnsRows(sql) ? (result.rows as SqlRow[]) : [{ changes: result.rowCount ?? 0 }] } catch (error) { - markRelayDatabaseError(error) + this.noteFailure(error) rememberPostgresTransactionPhase(error, sql) throw error } @@ -1021,6 +1044,11 @@ const POSTGRES_CONNECTION_TIMEOUT_MS = 2_000 // leaves room for the connect timeout above. export const POSTGRES_STATEMENT_TIMEOUT_MS = 5_000 const POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS = 5_000 +// Past statement_timeout because a whole-VM database stall delays replies the server has +// already timed: 5-7 s stalls are routine (connect-burst RCA), and a read timeout inside one +// would destroy healthy clients and fail requests that succeed today. +export const POSTGRES_READ_TIMEOUT_MARGIN_MS = 10_000 +const POSTGRES_IDLE_SESSION_TIMEOUT_MS = 30_000 export function relayPostgresStatementTimeoutMs( env: NodeJS.ProcessEnv = process.env @@ -1036,6 +1064,18 @@ export function relayPostgresStatementTimeoutMs( return milliseconds } +export function relayPostgresReadTimeoutMarginMs( + env: NodeJS.ProcessEnv = process.env +): number { + const configured = env.ORCA_RELAY_POSTGRES_READ_TIMEOUT_MARGIN_MS + if (configured === undefined || configured === '') return POSTGRES_READ_TIMEOUT_MARGIN_MS + const milliseconds = Number(configured) + if (!Number.isInteger(milliseconds) || milliseconds < 1) { + throw new Error('invalid_read_timeout_margin') + } + return milliseconds +} + function retryablePostgresTransactionError(error: unknown): boolean { const code = String((error as { code?: unknown }).code) // 57014 is the pool statement_timeout firing. It aborts the transaction the @@ -1056,14 +1096,33 @@ export function isRelayDatabaseTransientError(error: unknown): boolean { '57014', '53300', '57P03', + '08000', '08001', + '08003', '08006', + // The server ended the session: pg_terminate_backend or a fast/immediate shutdown. + '57P01', + '57P02', // The server ended a session idle in a transaction past its limit (a database stall). '25P03' ].includes(code) ) { return true } + // The connection dropped mid-statement: pg's own message, or the socket's reset. Only from + // the database layer, since any other socket can raise the same errno. + if ( + isRelayDatabaseLayerError(error) && + (code === 'ECONNRESET' || + code === 'EPIPE' || + String((error as { message?: unknown }).message).startsWith('Connection terminated')) + ) { + return true + } + // A lost reply, answered 503 like 08006. Its write may have landed, so transaction() never + // retries it. The one startup retry that can see it, the director's cell reconcile + // (cell-admission-startup.ts), re-runs only upserts that converge on the same rows. + if (isPostgresReadTimeout(error)) return true // A pool that cannot hand out a client reports no SQLSTATE at all, so the // acquire boundary owns that vocabulary. return isPostgresPoolConnectFailure(error) @@ -1103,6 +1162,7 @@ export class PostgresDatabase implements RelayDatabase { const startedAt = performance.now() let phase: 'acquire' | 'execute' = 'acquire' let client: pg.PoolClient | undefined + let lostReply: Error | undefined try { client = await this.pressure.connect(lane) phase = 'execute' @@ -1110,6 +1170,7 @@ export class PostgresDatabase implements RelayDatabase { return returnsRows(sql) ? (result.rows as SqlRow[]) : [{ changes: result.rowCount ?? 0 }] } catch (error) { markRelayDatabaseError(error) + if (isPostgresReadTimeout(error)) lostReply = error reportPostgresQueryFailure({ error, phase, @@ -1122,7 +1183,8 @@ export class PostgresDatabase implements RelayDatabase { }) throw error } finally { - client?.release() + // An error destroys the client rather than handing the next caller a dead connection. + client?.release(lostReply) } } @@ -1154,9 +1216,12 @@ export class PostgresDatabase implements RelayDatabase { for (let attempt = 1; attempt <= POSTGRES_TRANSACTION_ATTEMPTS; attempt++) { const client = await this.pressure.connect() const transaction = new PostgresTransaction(client) + let lostReply: Error | undefined try { await client.query('BEGIN') const result = await operation(transaction) + // Even when the operation caught it: COMMIT must not run after a lost reply. + if (transaction.poisonedBy) throw transaction.poisonedBy if (transaction.open) await client.query('COMMIT') recordMeasuredHold(this.holds, transaction) this.holds.recordUnavailable(transaction.consumeLockUnavailable()) @@ -1164,13 +1229,17 @@ export class PostgresDatabase implements RelayDatabase { return result } catch (error) { markRelayDatabaseError(error) + // No ROLLBACK behind a lost reply: it would queue behind the dead statement. Destroying + // the client ends the session, and the server drops its locks. + lostReply = transaction.poisonedBy ?? (isPostgresReadTimeout(error) ? error : undefined) const open = transaction.open - if (open) await client.query('ROLLBACK').catch(() => undefined) + if (open && !lostReply) await client.query('ROLLBACK').catch(() => undefined) this.holds.recordUnavailable(transaction.consumeLockUnavailable()) this.holds.recordLockTimeout(transaction.consumeLockTimeouts()) // Once the fused commit has ended the transaction, a retry would apply the work twice. if ( !open || + lostReply || !retryablePostgresTransactionError(error) || attempt === POSTGRES_TRANSACTION_ATTEMPTS ) { @@ -1197,7 +1266,7 @@ export class PostgresDatabase implements RelayDatabase { ) } } finally { - client.release() + client.release(lostReply) } // A PostgreSQL transaction is unusable after an abort, so retry all work // on a fresh pooled client with a small full-jitter delay. @@ -1260,6 +1329,27 @@ export function keepPostgresClientErrorsHandled(pool: Pick): void }) } +// An `options=` in the database URL silently replaces the pool's, so the session reports what +// it actually got. The URL itself is never read or logged. +function reportPostgresIdleSessionTimeoutOnce(pool: Pick): void { + let reported = false + pool.on('connect', (client) => { + if (reported) return + reported = true + client + .query("SELECT current_setting('idle_session_timeout') AS value") + .then((result: pg.QueryResult) => { + console.log( + JSON.stringify({ + event: 'orca_relay_postgres_session_settings', + idleSessionTimeout: String(result.rows[0]?.value) + }) + ) + }) + .catch(() => undefined) + }) +} + export function absorbPostgresIdleClientErrors(pool: Pick): void { pool.on('error', () => { // Why: node-postgres removes failed idle clients itself; leaving `error` @@ -1336,6 +1426,7 @@ export type RelayDatabaseOpenInput = { poolMax?: number applicationName?: string statementTimeoutMs?: number + readTimeoutMarginMs?: number // Directors own the PostgreSQL schema. A cell skips it and never touches the database // at boot, so it starts listening while the database is down and stays unready until // its first successful query. @@ -1349,17 +1440,26 @@ export async function openRelayDatabase(input: RelayDatabaseOpenInput): Promise< if (appliesPostgresSchema) { await applySchemaOnUntimedPool(input.databaseUrl, input.applicationName) } + const statementTimeoutMs = input.statementTimeoutMs ?? relayPostgresStatementTimeoutMs() const pool = new pg.Pool({ connectionString: input.databaseUrl, max: input.poolMax ?? 10, application_name: input.applicationName, connectionTimeoutMillis: POSTGRES_CONNECTION_TIMEOUT_MS, - statement_timeout: input.statementTimeoutMs ?? relayPostgresStatementTimeoutMs(), + statement_timeout: statementTimeoutMs, + // The server answers every statement within statement_timeout, as 57014 at worst, so + // this fires only on a lost reply, or a database or event-loop stall past the margin. + query_timeout: + statementTimeoutMs + (input.readTimeoutMarginMs ?? relayPostgresReadTimeoutMarginMs()), lock_timeout: POSTGRES_LOCK_TIMEOUT_MS, - idle_in_transaction_session_timeout: POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS + idle_in_transaction_session_timeout: POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS, + // Bounds a backend whose client was destroyed outside a transaction (after a lost + // reply on an autocommit write or COMMIT). pg-pool closes healthy idle clients at 10 s. + options: `-c idle_session_timeout=${POSTGRES_IDLE_SESSION_TIMEOUT_MS}` }) absorbPostgresIdleClientErrors(pool) keepPostgresClientErrorsHandled(pool) + reportPostgresIdleSessionTimeoutOnce(pool) database = new PostgresDatabase(pool) } else { mkdirSync(input.dataDir, { recursive: true }) diff --git a/cloud/apps/relay/src/postgres-lost-reply-postgres.test.ts b/cloud/apps/relay/src/postgres-lost-reply-postgres.test.ts new file mode 100644 index 00000000000..e7c5fd1b6fa --- /dev/null +++ b/cloud/apps/relay/src/postgres-lost-reply-postgres.test.ts @@ -0,0 +1,156 @@ +import { createServer, connect, type Server, type Socket } from 'node:net' +import { afterEach, describe, expect, it } from 'vitest' +import { + isRelayDatabaseTransientError, + openRelayDatabase, + POSTGRES_READ_TIMEOUT_MARGIN_MS, + type RelayDatabase +} from './database.js' + +const databaseUrl = process.env.ORCA_RELAY_TEST_POSTGRES_URL +const describePostgres = databaseUrl ? describe : describe.skip + +const STATEMENT_TIMEOUT_MS = 500 +// A short margin keeps the blackhole case fast; production's is 10 s. +const READ_TIMEOUT_MARGIN_MS = 2_000 +const READ_TIMEOUT_MS = STATEMENT_TIMEOUT_MS + READ_TIMEOUT_MARGIN_MS +const PROBE_TABLE = 'relay_lost_reply_probe' + +// Forwards each connection to PostgreSQL. Once a connection sends a statement naming the +// trigger, the server's replies on that connection are dropped (a reply lost in the network), +// or, with `delayMs`, held that long and then delivered (a whole-VM database stall). +async function blackholeProxy(target: URL, trigger: string, delayMs?: number) { + const sockets = new Set() + const afterTrigger: Buffer[] = [] + const server: Server = createServer((client) => { + const upstream = connect(Number(target.port || 5432), target.hostname) + sockets.add(client) + sockets.add(upstream) + let swallowing = false + client.on('data', (chunk: Buffer) => { + if (swallowing) afterTrigger.push(chunk) + else if (chunk.includes(trigger)) swallowing = true + upstream.write(chunk) + }) + upstream.on('data', (chunk: Buffer) => { + if (!swallowing) client.write(chunk) + else if (delayMs !== undefined) setTimeout(() => client.write(chunk), delayMs) + }) + for (const socket of [client, upstream]) { + socket.on('error', () => undefined) + socket.on('close', () => { + client.destroy() + upstream.destroy() + }) + } + }) + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)) + const address = server.address() + if (typeof address !== 'object' || address === null) throw new Error('proxy not listening') + const url = new URL(target) + url.hostname = '127.0.0.1' + url.port = String(address.port) + return { + url: url.toString(), + sentAfterTrigger: () => Buffer.concat(afterTrigger).toString('latin1'), + close: async () => { + for (const socket of sockets) socket.destroy() + await new Promise((resolve) => server.close(() => resolve())) + } + } +} + +describePostgres('a lost PostgreSQL reply against a real server', () => { + const cleanups: Array<() => Promise> = [] + afterEach(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup() + }) + + async function open( + url: string, + options: { readTimeoutMarginMs?: number; poolMax?: number } = {} + ): Promise { + const database = await openRelayDatabase({ + databaseUrl: url, + dataDir: '', + appliesPostgresSchema: false, + statementTimeoutMs: STATEMENT_TIMEOUT_MS, + readTimeoutMarginMs: options.readTimeoutMarginMs ?? READ_TIMEOUT_MARGIN_MS, + poolMax: options.poolMax + }) + cleanups.push(async () => await database.close()) + return database + } + + it('fails within the read timeout, sends nothing after the lost statement, and recovers', async () => { + const direct = await open(databaseUrl!) + await direct.query(`CREATE TABLE IF NOT EXISTS ${PROBE_TABLE} (id TEXT PRIMARY KEY)`) + await direct.query(`INSERT INTO ${PROBE_TABLE} (id) VALUES ('c1') ON CONFLICT DO NOTHING`) + cleanups.push(async () => { + await direct.query(`DROP TABLE IF EXISTS ${PROBE_TABLE}`) + }) + const proxy = await blackholeProxy(new URL(databaseUrl!), `FROM ${PROBE_TABLE}`) + cleanups.push(proxy.close) + const database = await open(proxy.url) + + const startedAt = performance.now() + const failure = await database + .transaction( + async (transaction) => + await transaction.queryLocked(`SELECT id FROM ${PROBE_TABLE} WHERE id = ?`, ['c1'], { + lockTimeoutMs: 200 + }) + ) + .catch((error: unknown) => error) + const elapsedMs = performance.now() - startedAt + + expect(failure).toBeInstanceOf(Error) + expect(isRelayDatabaseTransientError(failure)).toBe(true) + expect(elapsedMs).toBeLessThan(READ_TIMEOUT_MS + 1_000) + // pg's Terminate ('X') may follow as the client is destroyed; no statement may. + const after = proxy.sentAfterTrigger() + for (const statement of ['lock_timeout', 'ROLLBACK', 'COMMIT', 'SELECT']) { + expect(after).not.toContain(statement) + } + // The next checkout is a fresh connection, so it is not stuck behind the lost reply. + await expect(database.query('SELECT 1 AS one')).resolves.toEqual([{ one: 1 }]) + // The row lock went with the destroyed session. + await expect( + direct.transaction( + async (transaction) => + await transaction.queryLocked(`SELECT id FROM ${PROBE_TABLE} WHERE id = ?`, ['c1'], { + failIfUnavailable: true + }) + ) + ).resolves.toEqual([{ id: 'c1' }]) + }, 20_000) + + // A reply the server sent after statement_timeout, held up by a whole-VM stall, is still a + // reply: the default margin must outlast the routine 5-7 s stalls. + it('keeps a reply delayed 6.5 s by a stall, on the same connection', async () => { + const proxy = await blackholeProxy(new URL(databaseUrl!), 'stalled_reply_probe', 6_500) + cleanups.push(proxy.close) + const database = await open(proxy.url, { + readTimeoutMarginMs: POSTGRES_READ_TIMEOUT_MARGIN_MS, + poolMax: 1 + }) + const pid = async () => (await database.query('SELECT pg_backend_pid() AS pid'))[0]?.pid + const before = await pid() + const startedAt = performance.now() + // Autocommit: inside a transaction the proxy's hold would trip the server's own 5 s + // idle-in-transaction limit, which a real VM stall pauses along with everything else. + await expect(database.query("SELECT 'stalled_reply_probe' AS probe")).resolves.toEqual([ + { probe: 'stalled_reply_probe' } + ]) + expect(performance.now() - startedAt).toBeGreaterThan(6_000) + // Not destroyed: the one pooled connection is still the same backend. + expect(await pid()).toBe(before) + }, 30_000) + + it('runs request sessions with the server bounding idle backends at 30 s', async () => { + const database = await open(databaseUrl!) + await expect( + database.query("SELECT current_setting('idle_session_timeout') AS value") + ).resolves.toEqual([{ value: '30s' }]) + }) +}) diff --git a/cloud/apps/relay/src/postgres-lost-reply.test.ts b/cloud/apps/relay/src/postgres-lost-reply.test.ts new file mode 100644 index 00000000000..34b6ecca478 --- /dev/null +++ b/cloud/apps/relay/src/postgres-lost-reply.test.ts @@ -0,0 +1,124 @@ +import { EventEmitter } from 'node:events' +import { describe, expect, it, vi } from 'vitest' +import { isRelayDatabaseTransientError, PostgresDatabase } from './database.js' +import { + isFencedRelayDatabaseRejection, + POSTGRES_READ_TIMEOUT_MESSAGE +} from './relay-database-rejection-fence.js' + +// pg's query_timeout firing: the statement's reply never arrived. +const readTimeout = () => new Error(POSTGRES_READ_TIMEOUT_MESSAGE) + +class FakePoolClient extends EventEmitter { + readonly statements: string[] = [] + readonly released: Array = [] + + constructor(private readonly timesOut: (sql: string) => boolean) { + super() + } + + query = vi.fn(async (sql: string) => { + this.statements.push(sql) + if (this.timesOut(sql)) throw readTimeout() + return { rows: [{ one: 1 }], rowCount: 1 } + }) + + release = (error?: Error | boolean): void => { + this.released.push(error) + } +} + +function databaseOver(timesOut: (sql: string) => boolean) { + const clients: FakePoolClient[] = [] + const database = new PostgresDatabase({ + totalCount: 1, + idleCount: 0, + waitingCount: 0, + connect: async () => { + const client = new FakePoolClient(timesOut) + clients.push(client) + return client + } + } as never) + return { database, clients } +} + +describe('a lost PostgreSQL reply', () => { + it('ends a bounded locked statement with nothing else sent on its connection', async () => { + const { database, clients } = databaseOver((sql) => sql.includes('FOR UPDATE')) + const failure = await database + .transaction( + async (transaction) => + await transaction.queryLocked('SELECT id FROM cells WHERE id = ?', ['c1'], { + lockTimeoutMs: 200 + }) + ) + .catch((error: unknown) => error) + expect(failure).toEqual(readTimeout()) + // Not retried, no lock_timeout restore, no ROLLBACK, no COMMIT behind the dead statement. + expect(clients).toHaveLength(1) + expect(clients[0]?.statements).toEqual([ + 'BEGIN', + "SET LOCAL lock_timeout = '200ms'", + 'SELECT id FROM cells WHERE id = $1 FOR UPDATE' + ]) + // Released with the error, so pg-pool destroys it instead of reusing it. + expect(clients[0]?.released).toEqual([failure]) + // Answered 503 like a dropped connection, and survivable if nothing awaited it. + expect(isRelayDatabaseTransientError(failure)).toBe(true) + expect(isFencedRelayDatabaseRejection(failure)).toBe(true) + }) + + it('never commits after an operation that caught the timeout and carried on', async () => { + const { database, clients } = databaseOver((sql) => sql === 'UPDATE cells SET n = n + 1') + const failure = await database + .transaction(async (transaction) => { + await transaction.query('UPDATE cells SET n = n + 1').catch(() => undefined) + // Every later statement on the poisoned transaction fails at once, unsent. + await expect(transaction.query('SELECT 1')).rejects.toEqual(readTimeout()) + return 'carried on' + }) + .catch((error: unknown) => error) + expect(failure).toEqual(readTimeout()) + expect(clients[0]?.statements).toEqual(['BEGIN', 'UPDATE cells SET n = n + 1']) + expect(clients[0]?.released).toEqual([failure]) + }) + + it('treats a lost COMMIT reply as an unknown outcome: no ROLLBACK, no retry', async () => { + const { database, clients } = databaseOver((sql) => sql === 'COMMIT') + await expect( + database.transaction(async (transaction) => await transaction.query('SELECT 1')) + ).rejects.toEqual(readTimeout()) + expect(clients).toHaveLength(1) + expect(clients[0]?.statements).toEqual(['BEGIN', 'SELECT 1', 'COMMIT']) + expect(clients[0]?.released[0]).toEqual(readTimeout()) + }) + + it('destroys the client of a timed-out autocommit statement and serves the next one', async () => { + let first = true + const { database, clients } = databaseOver(() => { + const timesOut = first + first = false + return timesOut + }) + vi.spyOn(console, 'warn').mockImplementation(() => {}) + await expect(database.query('SELECT 1')).rejects.toEqual(readTimeout()) + await expect(database.query('SELECT 1')).resolves.toEqual([{ one: 1 }]) + expect(clients.map((client) => client.released)).toEqual([[readTimeout()], [undefined]]) + }) + + it('still rolls back and retries an ordinary server error', async () => { + let failures = 1 + const { database, clients } = databaseOver(() => false) + const result = await database.transaction(async (transaction) => { + if (failures-- > 0) throw Object.assign(new Error('deadlock detected'), { code: '40P01' }) + return await transaction.query('SELECT 1') + }) + expect(result).toEqual([{ one: 1 }]) + expect(clients.map((client) => client.statements)).toEqual([ + ['BEGIN', 'ROLLBACK'], + ['BEGIN', 'SELECT 1', 'COMMIT'] + ]) + expect(clients.map((client) => client.released)).toEqual([[undefined], [undefined]]) + }) +}) diff --git a/cloud/apps/relay/src/postgres-pool-pressure.ts b/cloud/apps/relay/src/postgres-pool-pressure.ts index 721748b9b91..0851402cdbe 100644 --- a/cloud/apps/relay/src/postgres-pool-pressure.ts +++ b/cloud/apps/relay/src/postgres-pool-pressure.ts @@ -21,6 +21,14 @@ const POOL_CONNECT_TIMEOUT_MESSAGES = [ // pg raises this whenever a socket ends early, during the handshake and mid // statement alike, so only the acquire boundary can tell the two apart. const CONNECTION_TERMINATED_MESSAGE = 'Connection terminated unexpectedly' +// The socket to Postgres (or its proxy) failed before a client existed, e.g. a Cloud SQL restart. +const ACQUIRE_SOCKET_ERRNOS = new Set([ + 'EPIPE', + 'ECONNRESET', + 'ECONNREFUSED', + 'ETIMEDOUT', + 'EHOSTUNREACH' +]) // Membership is tracked beside the error rather than on it: an error object may // be frozen, and a mutated one would leak the marker into logs. @@ -49,9 +57,10 @@ export function isPostgresPoolConnectTimeout(error: unknown): boolean { // outcome is unknown is not safe. export function isPostgresPoolConnectFailure(error: unknown): boolean { if (isPostgresPoolConnectTimeout(error)) return true + if (!isPostgresPoolAcquireFailure(error)) return false return ( - errorMessage(error).includes(CONNECTION_TERMINATED_MESSAGE) && - isPostgresPoolAcquireFailure(error) + errorMessage(error).includes(CONNECTION_TERMINATED_MESSAGE) || + ACQUIRE_SOCKET_ERRNOS.has(errorCode(error)) ) } diff --git a/cloud/apps/relay/src/postgres-query-failure.test.ts b/cloud/apps/relay/src/postgres-query-failure.test.ts index 25a3d7f7900..c35a9029255 100644 --- a/cloud/apps/relay/src/postgres-query-failure.test.ts +++ b/cloud/apps/relay/src/postgres-query-failure.test.ts @@ -72,12 +72,12 @@ describe('PostgreSQL query failure diagnostics', () => { expect(JSON.stringify(vi.mocked(console.warn).mock.calls)).not.toContain('private') }) - // ECONNRESET carries no SQLSTATE the routes retry on, and it arrives after the - // statement went out, so it stays a hard failure. The pair pins that boundary. + // A reset from the database layer answers 503 like 08006 (staging B1); transaction() still + // never retries it, since the statement went out. it.each([ ['57014', true], ['55P03', true], - ['ECONNRESET', false] + ['ECONNRESET', true] ] as const)( 'identifies execute failure %s and releases its client', async (code, transient) => { diff --git a/cloud/apps/relay/src/relay-database-rejection-fence.ts b/cloud/apps/relay/src/relay-database-rejection-fence.ts index 6a66ee53af0..f8eaf84e184 100644 --- a/cloud/apps/relay/src/relay-database-rejection-fence.ts +++ b/cloud/apps/relay/src/relay-database-rejection-fence.ts @@ -25,6 +25,16 @@ const FENCED_SQLSTATE_CLASSES = new Set(['08', '23', '40', '53', '55', '57', '58 // does to any backend caught idle mid-transaction at its onset. const FENCED_SQLSTATES = new Set(['25P03']) +export function isPostgresReadTimeout(error: unknown): error is Error { + return error instanceof Error && error.message === POSTGRES_READ_TIMEOUT_MESSAGE +} + +// The database layer threw or rethrew this; an errno or message alone could be any socket's. +export function isRelayDatabaseLayerError(error: unknown): boolean { + if (typeof error !== 'object' || error === null) return false + return databaseLayerErrors.has(error) || isPostgresPoolAcquireFailure(error) +} + export function markRelayDatabaseError(error: unknown): void { if (typeof error === 'object' && error !== null) databaseLayerErrors.add(error) } diff --git a/cloud/apps/relay/src/relay-fix-level.ts b/cloud/apps/relay/src/relay-fix-level.ts index 027494e42b4..d29b70dd8a7 100644 --- a/cloud/apps/relay/src/relay-fix-level.ts +++ b/cloud/apps/relay/src/relay-fix-level.ts @@ -3,4 +3,4 @@ // the fixed floor `relay_cell_min_fix_level` (production.tfvars) for 6 hours; raise the floor // once every serving cell runs the new level. Images from before this constant report nothing, // which the alert reads as below. -export const RELAY_FIX_LEVEL = 1 +export const RELAY_FIX_LEVEL = 2