diff --git a/cloud/apps/relay/src/database-connect-stall-postgres.test.ts b/cloud/apps/relay/src/database-connect-stall-postgres.test.ts new file mode 100644 index 00000000000..1bc14cf132a --- /dev/null +++ b/cloud/apps/relay/src/database-connect-stall-postgres.test.ts @@ -0,0 +1,158 @@ +import { + connect, + createServer as createTcpServer, + type AddressInfo, + type Server, + type Socket +} from 'node:net' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { isRelayDatabaseTransientError, openRelayDatabase, type RelayDatabase } from './database.js' + +const databaseUrl = process.env.ORCA_RELAY_TEST_POSTGRES_URL +const describePostgres = databaseUrl ? describe : describe.skip + +// A Cloud SQL instance that is starting has an IP but never answers. The 2026-10-11 staging stall +// came from the transport blocking connect() itself, which no test can reproduce; these pin what +// the pool must do over a transport that does not block: fail fast and keep the event loop turning. +type StallingDatabase = { + port: number + // 'forward' needs a real PostgreSQL behind it. + mode: 'blackhole' | 'refuse' | 'forward' + sockets: Set + server: Server +} + +const cleanups: Array<() => Promise> = [] + +afterEach(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup() + vi.restoreAllMocks() +}) + +describe('PostgreSQL pool over a database that accepts and never answers', () => { + it('fails every caller fast and retryably without stalling the event loop', async () => { + vi.spyOn(console, 'warn').mockImplementation(() => {}) + const stalling = await stallingDatabase('blackhole') + const database = await openStallingPool(stalling, 'postgresql://relay:secret@127.0.0.1') + const loop = watchEventLoop() + + // Twice the pool: three dial, three queue behind them in the general lane. + const outcomes = await Promise.all( + Array.from({ length: 6 }, async () => { + const startedAt = performance.now() + const error = await database.query('SELECT 1').catch((failure: unknown) => failure) + return { error, elapsedMs: performance.now() - startedAt } + }) + ) + + const gapMs = loop.stop() + for (const { error, elapsedMs } of outcomes) { + expect(isRelayDatabaseTransientError(error)).toBe(true) + // A queued caller waits out one connect timeout for a slot, then one for its own dial. + expect(elapsedMs).toBeLessThan(4_500) + } + expect(stalling.sockets.size).toBeGreaterThan(0) + expect(gapMs).toBeLessThan(250) + }, 15_000) +}) + +describePostgres('PostgreSQL pool through a database stop and start', () => { + it('fails fast while the database is down or starting, then serves again', async () => { + vi.spyOn(console, 'warn').mockImplementation(() => {}) + const stalling = await stallingDatabase('forward') + const database = await openStallingPool(stalling, databaseUrl!) + const loop = watchEventLoop() + expect(await database.query('SELECT 1 AS up')).toEqual([{ up: 1 }]) + + // Stopped: the instance refuses connections, and every pooled client is gone with it. + stalling.mode = 'refuse' + dropConnections(stalling) + await expectFastTransientFailure(database, 1_000) + + // Starting: the address answers TCP but PostgreSQL never speaks. + stalling.mode = 'blackhole' + await expectFastTransientFailure(database, 2_500) + + stalling.mode = 'forward' + dropConnections(stalling) + // The first caller after the start may still draw a client from the starting phase. + await expect.poll(async () => await database.query('SELECT 1 AS up').catch(() => []), { + timeout: 10_000, + interval: 100 + }).toEqual([{ up: 1 }]) + expect(await database.query('SELECT 2 AS up')).toEqual([{ up: 2 }]) + expect(loop.stop()).toBeLessThan(250) + }, 30_000) +}) + +async function expectFastTransientFailure(database: RelayDatabase, boundMs: number): Promise { + const startedAt = performance.now() + const error = await database.query('SELECT 1').catch((failure: unknown) => failure) + expect(isRelayDatabaseTransientError(error)).toBe(true) + expect(performance.now() - startedAt).toBeLessThan(boundMs) +} + +async function openStallingPool(stalling: StallingDatabase, url: string): Promise { + const viaStall = new URL(url) + viaStall.hostname = '127.0.0.1' + viaStall.port = String(stalling.port) + if (!viaStall.pathname || viaStall.pathname === '/') viaStall.pathname = '/relay' + const database = await openRelayDatabase({ + databaseUrl: viaStall.toString(), + dataDir: '', + poolMax: 3, + appliesPostgresSchema: false + }) + cleanups.push(async () => await database.close().catch(() => undefined)) + return database +} + +async function stallingDatabase(mode: StallingDatabase['mode']): Promise { + const target = databaseUrl ? new URL(databaseUrl) : undefined + const sockets = new Set() + const stalling: StallingDatabase = { port: 0, mode, sockets, server: createTcpServer() } + stalling.server.on('connection', (socket) => { + if (stalling.mode === 'refuse') { + socket.destroy() + return + } + sockets.add(socket) + socket.on('close', () => sockets.delete(socket)) + socket.on('error', () => undefined) + if (stalling.mode === 'blackhole' || !target) return + const upstream = connect(Number(target.port || 5432), target.hostname) + socket.pipe(upstream).pipe(socket) + upstream.on('error', () => socket.destroy()) + socket.on('close', () => upstream.destroy()) + }) + await new Promise((resolve) => stalling.server.listen(0, '127.0.0.1', resolve)) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: a TCP server listening on a host and port reports an AddressInfo, never a pipe path. + stalling.port = (stalling.server.address() as AddressInfo).port + cleanups.push(async () => { + dropConnections(stalling) + await new Promise((resolve) => stalling.server.close(() => resolve())) + }) + return stalling +} + +function dropConnections(stalling: StallingDatabase): void { + for (const socket of stalling.sockets) socket.destroy() + stalling.sockets.clear() +} + +// The longest gap between 10 ms ticks: a blocked loop shows as one long gap. +function watchEventLoop(): { stop: () => number } { + let last = performance.now() + let longestGapMs = 0 + const timer = setInterval(() => { + const now = performance.now() + longestGapMs = Math.max(longestGapMs, now - last) + last = now + }, 10) + return { + stop: () => { + clearInterval(timer) + return longestGapMs + } + } +} diff --git a/cloud/apps/relay/src/relay-observability.test.ts b/cloud/apps/relay/src/relay-observability.test.ts index a993225acfe..2517282b947 100644 --- a/cloud/apps/relay/src/relay-observability.test.ts +++ b/cloud/apps/relay/src/relay-observability.test.ts @@ -72,6 +72,25 @@ describe('relay observability', () => { expect(RELAY_FIX_LEVEL).toBeGreaterThanOrEqual(1) }) + // Why: a director whose loop froze for 134 s still reported a 68 ms p99 (staging 2026-10-11). + it('reports one long event-loop freeze as the window max', async () => { + const entries: Array> = [] + const observability = new RelayObservability( + { role: 'director', cellId: 'director', region: 'us-central1' }, + (entry) => entries.push(entry) + ) + observability.start(() => counts, 3_600_000) + await new Promise((resolve) => setTimeout(resolve, 100)) + const frozenUntil = performance.now() + 300 + while (performance.now() < frozenUntil) { + // Holds the loop the way a blocking connect() does. + } + await new Promise((resolve) => setTimeout(resolve, 100)) + observability.flush(counts) + observability.stop() + expect(entries[0]!.eventLoopDelayMsMax).toBeGreaterThanOrEqual(250) + }) + it('emits safe readiness dependency outcomes', () => { const entries: Array> = [] const observability = new RelayObservability( diff --git a/cloud/apps/relay/src/relay-observability.ts b/cloud/apps/relay/src/relay-observability.ts index b24b0608ea6..17dd1ab3a44 100644 --- a/cloud/apps/relay/src/relay-observability.ts +++ b/cloud/apps/relay/src/relay-observability.ts @@ -485,6 +485,8 @@ export class RelayObservability implements RelayRuntimeObserver { const controlRenewalFlush = latencySummary(deltas.controlRenewalFlushLatenciesMs) const memory = process.memoryUsage() const p99 = this.eventLoop.count === 0 ? 0 : this.eventLoop.percentile(99) / 1_000_000 + // One sample covers a whole freeze, so p99 cannot see a single long one; max can. + const max = this.eventLoop.count === 0 ? 0 : this.eventLoop.max / 1_000_000 this.eventLoop.reset() this.write({ severity: 'INFO', @@ -595,7 +597,8 @@ export class RelayObservability implements RelayRuntimeObserver { httpLatencyMsMax: roundMs(deltas.httpLatencyMsMax), heapUsedBytes: memory.heapUsed, heapTotalBytes: memory.heapTotal, - eventLoopDelayMsP99: Number(p99.toFixed(3)) + eventLoopDelayMsP99: Number(p99.toFixed(3)), + eventLoopDelayMsMax: Number(max.toFixed(3)) }) } } diff --git a/cloud/apps/relay/src/shadow-seat-directory.test.ts b/cloud/apps/relay/src/shadow-seat-directory.test.ts index e57428a6b03..09cc2a352dd 100644 --- a/cloud/apps/relay/src/shadow-seat-directory.test.ts +++ b/cloud/apps/relay/src/shadow-seat-directory.test.ts @@ -427,6 +427,27 @@ describe('startShadowSeatPoller', () => { expect(listCells.mock.calls.length).toBeGreaterThanOrEqual(5) }) + it('keeps polling the last cell list while a cell-list read hangs', async () => { + let finishRead: (cells: SeatFeedCell[]) => void = () => undefined + const listCells = vi + .fn<() => Promise>() + .mockResolvedValueOnce(CELLS) + .mockImplementation(() => new Promise((resolve) => (finishRead = resolve))) + const cell = poller({ 'cell-a': () => json(feed({ seq: 1, full: [] })) }, ['cell-a'], listCells) + await cell.tick() + cell.advance(31_000) + const round = cell.tick() + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(listCells).toHaveBeenCalledTimes(2) + expect(cell.requests.map((url) => url.hostname)).toEqual([ + 'cell-a.example.test', + 'cell-a.example.test' + ]) + finishRead(CELLS) + await round + cell.stop() + }) + it('treats a malformed or mislabelled body as unverifiable', async () => { const cell = poller({ 'cell-a': () => json({ v: 2 }), diff --git a/cloud/apps/relay/src/shadow-seat-directory.ts b/cloud/apps/relay/src/shadow-seat-directory.ts index 6a491123e71..1e2064e650c 100644 --- a/cloud/apps/relay/src/shadow-seat-directory.ts +++ b/cloud/apps/relay/src/shadow-seat-directory.ts @@ -741,10 +741,13 @@ export function startShadowSeatPoller( const round = async (awaitPolls: boolean): Promise => { if (stopped) return - await refreshCells() + const refresh = refreshCells() + // Only the first list is waited for: later rounds poll the last list, so the round that starts + // a refresh no longer skips its polls (one round, ~1 s, per refresh) while the read is slow. + if (cellsReadAt === undefined) await refresh // A slow cell holds only its own slot; the timer never waits on it. const polls = cells.map((cell) => poll(cell)) - if (awaitPolls) await Promise.all(polls) + if (awaitPolls) await Promise.all([refresh, ...polls]) if (now() - lastSummaryAt >= SUMMARY_INTERVAL_MS) { lastSummaryAt = now() log(JSON.stringify({ event: 'orca_relay_shadow_seat_summary', ...directory.summary(now()) }))