mirror of
https://github.com/stablyai/orca.git
synced 2026-10-31 00:57:45 +00:00
fix(relay): report event-loop freezes and keep seat polls off the cell-list read (#27405)
* fix(relay): keep seat polls off the cell-list read and report event-loop freezes The seat poller no longer waits on its database cell-list read before polling cells once it has a list, so a stalled read cannot starve the polls that feed a reserve cell's dead-man. Runtime metrics add eventLoopDelayMsMax: a single long freeze is one sample, so p99 read 68 ms across a 134 s staging freeze. Adds a pool test against a database that accepts and never answers, and a real-Postgres stop/start recovery test. * fix(relay): state the seat-poller gain precisely, justify the AddressInfo cast
This commit is contained in:
@@ -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<Socket>
|
||||
server: Server
|
||||
}
|
||||
|
||||
const cleanups: Array<() => Promise<void>> = []
|
||||
|
||||
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<void> {
|
||||
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<RelayDatabase> {
|
||||
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<StallingDatabase> {
|
||||
const target = databaseUrl ? new URL(databaseUrl) : undefined
|
||||
const sockets = new Set<Socket>()
|
||||
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<void>((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<void>((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
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<Record<string, unknown>> = []
|
||||
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<Record<string, unknown>> = []
|
||||
const observability = new RelayObservability(
|
||||
|
||||
@@ -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))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<SeatFeedCell[]>>()
|
||||
.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 }),
|
||||
|
||||
@@ -741,10 +741,13 @@ export function startShadowSeatPoller(
|
||||
|
||||
const round = async (awaitPolls: boolean): Promise<void> => {
|
||||
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()) }))
|
||||
|
||||
Reference in New Issue
Block a user