From 17959a90f77f2af4bd5af4e2b938e6a2fd5a41e1 Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Wed, 14 Oct 2026 15:35:27 -0400 Subject: [PATCH] feat(relay): measure each host hello's pre-auth time on the cell (#28695) * feat(relay): measure each host hello's pre-auth time on the cell Adds preAuthLatencyMs to the per-cell orca_relay_runtime_metrics line: per admit path (db, reserve) and outcome (admitted, refused), a count plus p50/p95/p99/max for the window. Each sample runs from the cell accepting the host control WebSocket to the hello being admitted (proof verified) or refused. Samples are kept in a bounded reservoir, shared with the control round-trip metric. No behaviour, wire, or flag change. Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f * fix(relay): count abandoned hellos and publish mergeable pre-auth buckets - Record a third outcome, abandoned, for a host that hangs up before a decision or before the cell acts on one, so the slowest hellos no longer drop out of the window. - Add non-cumulative bucket counts per group (bounds 50..5000 ms and +Inf, 450 exact) so the share at or under 450 ms sums exactly across windows and cells. - Split refusals that end before an admission check answers into a beforeCheck path. - One refuse() closure per phase records the metric, and the path is passed to settle. - Drop the half-used LatencyReservoir type; the shared reservoir function stays. - Replace test casts with a typed accessor or SAFETY comments. Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f * docs(relay): note pre-auth path labels for late throws and the combined role Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f --- .../src/host-session-preauth-latency.test.ts | 321 ++++++++++++++++++ cloud/apps/relay/src/host-session-registry.ts | 202 +++++++---- .../relay/src/relay-observability.test.ts | 104 ++++++ cloud/apps/relay/src/relay-observability.ts | 121 ++++++- 4 files changed, 668 insertions(+), 80 deletions(-) create mode 100644 cloud/apps/relay/src/host-session-preauth-latency.test.ts diff --git a/cloud/apps/relay/src/host-session-preauth-latency.test.ts b/cloud/apps/relay/src/host-session-preauth-latency.test.ts new file mode 100644 index 00000000000..ae7b9eec5d3 --- /dev/null +++ b/cloud/apps/relay/src/host-session-preauth-latency.test.ts @@ -0,0 +1,321 @@ +import { createHash, createHmac } from 'node:crypto' +import { EventEmitter } from 'node:events' +import { + buildHostProofMacInput, + HOST_CHALLENGE_PLAINTEXT_DOMAIN, + RELAY_CLOSE_CODE +} from '@orca-cloud/relay-contract' +import nacl from 'tweetnacl' +import { afterEach, describe, expect, it, vi } from 'vitest' +import type WebSocket from 'ws' +import type { RelayAssignmentStore } from './assignment-store.js' +import type { CellReserveBook } from './cell-reserve-book.js' +import type { RelayConfig } from './config.js' +import type { RelayCredentialStore } from './credential-store.js' +import { HostSessionRegistry, type CellReserveAdmission } from './host-session-registry.js' +import type { RelayRuntimeObserver } from './relay-observability.js' +import type { RelayTokenClaims } from './relay-token-verifier.js' +import { ProcessQueuedByteBudget } from './splice-forwarder.js' + +const CELL_ID = 'staging-gce-c1' + +class FakeSocket extends EventEmitter { + readonly CLOSING = 2 + readonly OPEN = 1 + readonly CLOSED = 3 + readyState = this.OPEN + readonly send = vi.fn() + readonly close = vi.fn((code?: number) => { + this.readyState = this.CLOSED + this.emit('close', code, Buffer.from('')) + }) + + asWebSocket(): WebSocket { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: FakeSocket implements the registry's WebSocket event and lifecycle surface. + return this as unknown as WebSocket + } + + // The peer's close frame: ws leaves OPEN at once and emits 'close' later. + hangUp(): void { + this.readyState = this.CLOSING + } +} + +function hostKey(): { keyPair: nacl.BoxKeyPair; identity: RelayTokenClaims } { + const keyPair = nacl.box.keyPair() + const relayHostId = createHash('sha256') + .update(keyPair.publicKey) + .digest('base64url') + .slice(0, 16) + return { + keyPair, + identity: { + sub: 'user-1', + prof: 'profile-1', + relayHostId, + purpose: 'host-control', + exp: 4_102_444_800 + } + } +} + +function setup(input: { + admitMode: 'db' | 'reserve' + databaseSays: boolean + databaseShedding?: boolean +}) { + let clock = 1_000_000 + const recordHostPreAuth = vi.fn>() + const observer = { + recordAuth: vi.fn(), + recordForwardedBytes: vi.fn(), + recordHttp: vi.fn(), + recordReconnect: vi.fn(), + recordSql: vi.fn(), + recordHostPreAuth + } satisfies RelayRuntimeObserver + const take = vi.fn((userId, relayHostId, epoch) => ({ + userId, + relayHostId, + epoch, + directorId: 'director-1', + expiresAt: clock + 60_000 + })) + const reserveAdmission: CellReserveAdmission = { + mode: () => input.admitMode, + ticketEnforce: () => false, + dryRunEnabled: () => false, + databaseShedding: () => input.databaseShedding ?? false, + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: beginProof calls only take() on the reserve book. + book: { take } as unknown as CellReserveBook, + verifyLease: vi.fn(async () => 'absent' as const) + } + let answerDatabase: (valid: boolean) => void = () => {} + const databaseAnswered = new Promise((resolve) => { + answerDatabase = resolve + }) + // Answers at once unless the test holds it with holdDatabase(). + let held = false + const verifyCellAssignment = vi.fn(async () => (held ? databaseAnswered : input.databaseSays)) + const registry = new HostSessionRegistry( + config(), + vi.fn(), + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the host hello path never reads credentials. + {} as RelayCredentialStore, + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the host hello path calls only verifyCellAssignment(). + { verifyCellAssignment } as unknown as RelayAssignmentStore, + new ProcessQueuedByteBudget(), + observer, + () => clock, + Math.random, + 'incarnation-1', + undefined, + reserveAdmission + ) + return { + registry, + recordHostPreAuth, + verifyCellAssignment, + take, + holdDatabase: () => { + held = true + }, + answerDatabase: (valid: boolean) => answerDatabase(valid), + advance: (ms: number) => { + clock += ms + } + } +} + +function sendHello(socket: FakeSocket, keyPair: nacl.BoxKeyPair, relayHostId: string): void { + socket.emit( + 'message', + Buffer.from( + JSON.stringify({ + type: 'host-hello', + v: 1, + relayHostId, + assignmentEpoch: 4, + hostPublicKeyB64: Buffer.from(keyPair.publicKey).toString('base64'), + appVersion: '1.4.200' + }) + ), + false + ) +} + +function answerChallenge(socket: FakeSocket, keyPair: nacl.BoxKeyPair): void { + const challenge: unknown = JSON.parse(String(socket.send.mock.calls[0]?.[0])) + const field = (name: string): string => { + const value: unknown = + typeof challenge === 'object' && challenge !== null ? Reflect.get(challenge, name) : null + if (typeof value !== 'string') throw new Error(`host-challenge has no ${name}`) + return value + } + const plaintext = nacl.box.open( + Buffer.from(field('ciphertextB64'), 'base64'), + Buffer.from(field('nonceB64'), 'base64'), + Buffer.from(field('relayEphemeralPublicKeyB64'), 'base64'), + keyPair.secretKey + ) + if (!plaintext) throw new Error('host-challenge did not open') + const domainLength = new TextEncoder().encode(`${HOST_CHALLENGE_PLAINTEXT_DOMAIN}\0`).length + const transcriptLength = new DataView( + plaintext.buffer, + plaintext.byteOffset + domainLength, + 4 + ).getUint32(0, false) + const transcriptStart = domainLength + 4 + const transcript = plaintext.slice(transcriptStart, transcriptStart + transcriptLength) + const secret = plaintext.slice(transcriptStart + transcriptLength) + socket.emit( + 'message', + Buffer.from( + JSON.stringify({ + type: 'host-challenge-ack', + challengeId: field('challengeId'), + proofB64: createHmac('sha256', secret) + .update(buildHostProofMacInput(transcript)) + .digest('base64') + }) + ), + false + ) +} + +describe('host pre-auth latency', () => { + afterEach(() => vi.useRealTimers()) + + it('records a booked hello on the reserve path, from accept to a verified proof', async () => { + const h = setup({ admitMode: 'reserve', databaseSays: false }) + const { keyPair, identity } = hostKey() + const socket = new FakeSocket() + h.registry.acceptControl(socket.asWebSocket(), identity) + h.advance(30) + sendHello(socket, keyPair, identity.relayHostId) + await vi.waitFor(() => expect(socket.send).toHaveBeenCalledTimes(1)) + expect(h.recordHostPreAuth).not.toHaveBeenCalled() + h.advance(45) + answerChallenge(socket, keyPair) + + expect(h.take).toHaveBeenCalledOnce() + expect(h.verifyCellAssignment).not.toHaveBeenCalled() + expect(h.recordHostPreAuth).toHaveBeenCalledExactlyOnceWith('reserve', 'admitted', 75) + }) + + it('records a hello the database refuses on the db path', async () => { + const h = setup({ admitMode: 'db', databaseSays: false }) + const { keyPair, identity } = hostKey() + const socket = new FakeSocket() + h.registry.acceptControl(socket.asWebSocket(), identity) + h.advance(120) + sendHello(socket, keyPair, identity.relayHostId) + await vi.waitFor(() => expect(socket.close).toHaveBeenCalled()) + + expect(socket.close).toHaveBeenCalledWith(RELAY_CLOSE_CODE.WRONG_CELL, 'wrong assignment epoch') + expect(h.take).not.toHaveBeenCalled() + expect(h.recordHostPreAuth).toHaveBeenCalledExactlyOnceWith('db', 'refused', 120) + }) + + it('records a hello timeout as a refusal before any check, once', () => { + vi.useFakeTimers() + const h = setup({ admitMode: 'db', databaseSays: true }) + const { identity } = hostKey() + const socket = new FakeSocket() + h.registry.acceptControl(socket.asWebSocket(), identity) + h.advance(2_000) + vi.advanceTimersByTime(2_000) + + expect(socket.close).toHaveBeenCalledWith( + RELAY_CLOSE_CODE.BAD_OUTER_CREDENTIAL, + 'host hello timeout' + ) + expect(h.recordHostPreAuth).toHaveBeenCalledExactlyOnceWith('beforeCheck', 'refused', 2_000) + }) + + it('records a host that hangs up before its hello as abandoned', () => { + const h = setup({ admitMode: 'db', databaseSays: true }) + const { identity } = hostKey() + const socket = new FakeSocket() + h.registry.acceptControl(socket.asWebSocket(), identity) + h.advance(300) + socket.close() + + expect(h.recordHostPreAuth).toHaveBeenCalledExactlyOnceWith('beforeCheck', 'abandoned', 300) + }) + + it.each([true, false])( + 'records a database answer (%s) that arrives after the host hung up as abandoned', + async (valid) => { + const h = setup({ admitMode: 'db', databaseSays: valid }) + h.holdDatabase() + const { keyPair, identity } = hostKey() + const socket = new FakeSocket() + h.registry.acceptControl(socket.asWebSocket(), identity) + sendHello(socket, keyPair, identity.relayHostId) + await vi.waitFor(() => expect(h.verifyCellAssignment).toHaveBeenCalled()) + socket.hangUp() + h.advance(900) + h.answerDatabase(valid) + await vi.waitFor(() => expect(h.recordHostPreAuth).toHaveBeenCalled()) + + expect(socket.send).not.toHaveBeenCalled() + expect(h.recordHostPreAuth).toHaveBeenCalledExactlyOnceWith('db', 'abandoned', 900) + } + ) + + it('records a host that hangs up during its proof as abandoned on its path', async () => { + const h = setup({ admitMode: 'reserve', databaseSays: false }) + const { keyPair, identity } = hostKey() + const socket = new FakeSocket() + h.registry.acceptControl(socket.asWebSocket(), identity) + sendHello(socket, keyPair, identity.relayHostId) + await vi.waitFor(() => expect(socket.send).toHaveBeenCalledTimes(1)) + h.advance(700) + socket.close() + + expect(h.recordHostPreAuth).toHaveBeenCalledExactlyOnceWith('reserve', 'abandoned', 700) + }) + + it('records the pool-pressure shed as a refusal before any check', async () => { + const h = setup({ admitMode: 'reserve', databaseSays: true, databaseShedding: true }) + h.take.mockReturnValue(null) + const { keyPair, identity } = hostKey() + const socket = new FakeSocket() + h.registry.acceptControl(socket.asWebSocket(), identity) + h.advance(40) + sendHello(socket, keyPair, identity.relayHostId) + await vi.waitFor(() => expect(socket.close).toHaveBeenCalled()) + + expect(h.verifyCellAssignment).not.toHaveBeenCalled() + expect(h.recordHostPreAuth).toHaveBeenCalledExactlyOnceWith('beforeCheck', 'refused', 40) + }) +}) + +function config(): RelayConfig { + return { + port: 8080, + publicUrl: 'https://c1.relay.example.test', + cellUrl: 'https://c1.relay.example.test', + authIssuer: 'https://auth.example.test', + authAudience: 'orca-relay', + jwksUrl: 'https://auth.example.test/jwks', + assignmentSigningKey: new TextEncoder().encode('0123456789abcdef0123456789abcdef'), + role: 'cell', + cellId: CELL_ID, + cells: [], + adminAudience: 'https://relay.example.test/v1/admin/drain', + deployServiceAccount: 'deploy@example.test', + runtimeServiceAccount: 'relay-cell@example.test', + adminJwksUrl: 'https://auth.example.test/jwks', + databasePoolMax: 10, + publicAssignmentsEnabled: true, + publicAssignmentConcurrency: 2, + publicAssignmentQueueMax: 128, + publicAssignmentWaitMs: 4_000, + publicResolveConcurrency: 1, + publicResolveWaitMs: 5_000, + publicAssignmentRetryAfterSeconds: 5, + dataDir: './data' + } +} diff --git a/cloud/apps/relay/src/host-session-registry.ts b/cloud/apps/relay/src/host-session-registry.ts index 1e3210a3fda..1df66ae5edb 100644 --- a/cloud/apps/relay/src/host-session-registry.ts +++ b/cloud/apps/relay/src/host-session-registry.ts @@ -42,6 +42,8 @@ import { relayHostLogDigest } from './relay-host-log-digest.js' import type { RelayTokenClaims } from './relay-token-verifier.js' import { percentile, + type HostPreAuthOutcome, + type HostPreAuthPath, type RelayClientAcceptStage, type RelayClientAcceptTimedStage, type RelayRuntimeObserver @@ -186,26 +188,44 @@ function readControlFrame( socket: WebSocket, timeoutMs: number, timeoutReason: string, - receive: (raw: RawData, isBinary: boolean) => void + receive: (raw: RawData, isBinary: boolean) => void, + ended?: (how: 'timeout' | 'closed') => void ): void { - if (socket.readyState !== socket.OPEN) return + if (socket.readyState !== socket.OPEN) { + ended?.('closed') + return + } const timer = setTimeout(() => { finish() + ended?.('timeout') socket.close(RELAY_CLOSE_CODE.BAD_OUTER_CREDENTIAL, timeoutReason) }, timeoutMs) function finish(): void { clearTimeout(timer) socket.off('message', onMessage) - socket.off('close', finish) + socket.off('close', onClose) + } + function onClose(): void { + finish() + ended?.('closed') } function onMessage(raw: RawData, isBinary: boolean): void { finish() receive(raw, isBinary) } socket.once('message', onMessage) - socket.once('close', finish) + socket.once('close', onClose) } +// Records one host control's pre-auth time, from the cell accepting the WebSocket to the +// hello's admission and proof being decided, exactly once. +type SettlePreAuth = (outcome: HostPreAuthOutcome, path: HostPreAuthPath) => void + +const preAuthEnded = + (settle: SettlePreAuth, path: HostPreAuthPath) => + (how: 'timeout' | 'closed'): void => + settle(how === 'timeout' ? 'refused' : 'abandoned', path) + // Hosts abandon connects after 15s; waiting much longer than that behind a // stalled predecessor only accumulates doomed sockets. const ACTIVATION_QUEUE_WAIT_MS = 30_000 @@ -1212,35 +1232,64 @@ export class HostSessionRegistry { hostCapabilities?: ReadonlySet, assignmentLease?: string ): void { + const settlePreAuth = this.startPreAuth(socket) + const refuse = (code: number, reason: string): void => { + settlePreAuth('refused', 'beforeCheck') + socket.close(code, reason) + } if (this.idleAttempts.has(identity.relayHostId)) { - socket.close(RELAY_CLOSE_CODE.WRONG_CELL, 'idle cutover in progress') + refuse(RELAY_CLOSE_CODE.WRONG_CELL, 'idle cutover in progress') return } // Keyed by socket, not session: a rebind swaps the session's socket, and the // successor's own advertisement is the only one that describes its decoder. if (hostCapabilities?.size) this.hostCapabilities.set(socket, hostCapabilities) if (this.draining) { - socket.close(RELAY_CLOSE_CODE.DRAINING, 'relay draining') + refuse(RELAY_CLOSE_CODE.DRAINING, 'relay draining') return } - readControlFrame(socket, 2_000, 'host hello timeout', (raw, isBinary) => { - if (isBinary) { - socket.close(RELAY_CLOSE_CODE.BAD_OUTER_CREDENTIAL, 'host hello must be text') - return - } - this.guardSessionTask( - () => - this.beginProof( - socket, - identity, - payload(raw, 'host-hello'), - connectionInclusionWatermark, - assignmentLease - ), - socket, - 'host hello proof' - ) - }) + readControlFrame( + socket, + 2_000, + 'host hello timeout', + (raw, isBinary) => { + if (isBinary) { + refuse(RELAY_CLOSE_CODE.BAD_OUTER_CREDENTIAL, 'host hello must be text') + return + } + this.guardSessionTask( + () => + this.beginProof( + socket, + identity, + payload(raw, 'host-hello'), + settlePreAuth, + connectionInclusionWatermark, + assignmentLease + ).catch((error: unknown) => { + // guardSessionTask closes the socket, so any throw refuses the hello as beforeCheck, + // including one after the database answered (e.g. in send or the session lookup). + settlePreAuth('refused', 'beforeCheck') + throw error + }), + socket, + 'host hello proof' + ) + }, + preAuthEnded(settlePreAuth, 'beforeCheck') + ) + } + + private startPreAuth(socket: WebSocket): SettlePreAuth { + const startedAt = this.now() + let settled = false + return (outcome, path) => { + if (settled) return + settled = true + // A decision made after the host hung up never reached it, so it counts as abandoned. + const recorded = socket.readyState === socket.OPEN ? outcome : 'abandoned' + this.observer.recordHostPreAuth?.(path, recorded, nonNegativeMs(this.now() - startedAt)) + } } // A dependency failure (e.g. a database connect timeout) must cost one @@ -1379,29 +1428,34 @@ export class HostSessionRegistry { socket: WebSocket, identity: RelayTokenClaims, candidate: unknown, + settlePreAuth: SettlePreAuth, connectionInclusionWatermark?: number, assignmentLease?: string ): Promise { + // Which check decided the hello; refusals before one answers stay 'beforeCheck'. + let path: HostPreAuthPath = 'beforeCheck' + const refuse = (code: number, reason: string): void => { + settlePreAuth('refused', path) + this.observer.recordAuth(false) + socket.close(code, reason) + } const hello = HostHelloSchema.safeParse(candidate) const hostPublicKey = hello.success ? decodeCanonicalBase64(hello.data.hostPublicKeyB64, 32) : null if (!hello.success) { - this.observer.recordAuth(false) - socket.close(RELAY_CLOSE_CODE.BAD_OUTER_CREDENTIAL, 'invalid host hello') + refuse(RELAY_CLOSE_CODE.BAD_OUTER_CREDENTIAL, 'invalid host hello') return } if (!hostPublicKey) { - this.observer.recordAuth(false) - socket.close(RELAY_CLOSE_CODE.BAD_OUTER_CREDENTIAL, 'invalid host public key') + refuse(RELAY_CLOSE_CODE.BAD_OUTER_CREDENTIAL, 'invalid host public key') return } if ( hello.data.relayHostId !== identity.relayHostId || relayHostId(hostPublicKey) !== identity.relayHostId ) { - this.observer.recordAuth(false) - socket.close(RELAY_CLOSE_CODE.BAD_OUTER_CREDENTIAL, 'host key binding mismatch') + refuse(RELAY_CLOSE_CODE.BAD_OUTER_CREDENTIAL, 'host key binding mismatch') return } const leaseShadow = this.config.role === 'cell' ? this.assignmentLeaseShadow : undefined @@ -1434,12 +1488,12 @@ export class HostSessionRegistry { relayHostId: identity.relayHostId, helloEpoch: hello.data.assignmentEpoch })) === 'valid') + if (admittedFromMemory) path = 'reserve' } if (reserveMode && !admittedFromMemory && this.reserveAdmission!.databaseShedding()) { // Memory cannot admit it (e.g. seated here, but at another epoch), and the wedged pool // would only time out its database check: shed it now, as the upgrade shed does. - this.observer.recordAuth(false) - socket.close(RELAY_CLOSE_CODE.LIMIT_EXCEEDED, 'relay temporarily unavailable') + refuse(RELAY_CLOSE_CODE.LIMIT_EXCEEDED, 'relay temporarily unavailable') return } const assignmentReadStartedAt = performance.now() @@ -1462,12 +1516,16 @@ export class HostSessionRegistry { performance.now() - assignmentReadStartedAt ) } + // The `combined` role compares the epoch without querying but is still labelled db. + if (!admittedFromMemory) path = 'db' if (!assignmentValid) { - this.observer.recordAuth(false) - socket.close(RELAY_CLOSE_CODE.WRONG_CELL, 'wrong assignment epoch') + refuse(RELAY_CLOSE_CODE.WRONG_CELL, 'wrong assignment epoch') + return + } + if (socket.readyState !== socket.OPEN) { + settlePreAuth('abandoned', path) return } - if (socket.readyState !== socket.OPEN) return const key = this.key(identity.sub, identity.relayHostId) const existing = this.sessions.get(key) @@ -1513,40 +1571,46 @@ export class HostSessionRegistry { ciphertextB64: Buffer.from(ciphertext).toString('base64'), expiresAt }) - readControlFrame(socket, 10_000, 'host proof timeout', (raw, isBinary) => { - const ack = isBinary - ? null - : HostChallengeAckSchema.safeParse(payload(raw, 'host-challenge-ack')) - const proof = ack?.success ? decodeCanonicalBase64(ack.data.proofB64, 32) : null - if ( - !ack?.success || - ack.data.challengeId !== challengeId || - !proof || - this.now() > expiresAt || - !timingSafeEqual(proof, expectedProof) - ) { - this.observer.recordAuth(false) - socket.close(RELAY_CLOSE_CODE.BAD_OUTER_CREDENTIAL, 'invalid host proof') - return - } - this.observer.recordAuth(true) - this.guardSessionTask( - () => - this.activate( - socket, - identity, - existing ?? null, - generation, - rebind, - hello.data.assignmentEpoch, - hello.data.appVersion, - connectionInclusionWatermark, - admittedFromMemory ? { reservedBy } : undefined - ), - socket, - 'host activation' - ) - }) + readControlFrame( + socket, + 10_000, + 'host proof timeout', + (raw, isBinary) => { + const ack = isBinary + ? null + : HostChallengeAckSchema.safeParse(payload(raw, 'host-challenge-ack')) + const proof = ack?.success ? decodeCanonicalBase64(ack.data.proofB64, 32) : null + if ( + !ack?.success || + ack.data.challengeId !== challengeId || + !proof || + this.now() > expiresAt || + !timingSafeEqual(proof, expectedProof) + ) { + refuse(RELAY_CLOSE_CODE.BAD_OUTER_CREDENTIAL, 'invalid host proof') + return + } + settlePreAuth('admitted', path) + this.observer.recordAuth(true) + this.guardSessionTask( + () => + this.activate( + socket, + identity, + existing ?? null, + generation, + rebind, + hello.data.assignmentEpoch, + hello.data.appVersion, + connectionInclusionWatermark, + admittedFromMemory ? { reservedBy } : undefined + ), + socket, + 'host activation' + ) + }, + preAuthEnded(settlePreAuth, path) + ) } private activate( diff --git a/cloud/apps/relay/src/relay-observability.test.ts b/cloud/apps/relay/src/relay-observability.test.ts index 2517282b947..a1ec6b5030f 100644 --- a/cloud/apps/relay/src/relay-observability.test.ts +++ b/cloud/apps/relay/src/relay-observability.test.ts @@ -5,6 +5,7 @@ import { observeRelayDatabase } from './observed-relay-database.js' import { CONTROL_RTT_RESERVOIR_LIMIT, observedRelayRequests, + PRE_AUTH_RESERVOIR_LIMIT, percentile, RelayObservability, type RelayProcessCounts @@ -570,6 +571,109 @@ describe('relay observability', () => { expect(entries[0]!.controlRttMsMax).toBe(900) }) + it('summarises host pre-auth latency per path and outcome', () => { + const entries: Array> = [] + const observability = new RelayObservability( + { role: 'cell', cellId: 'staging-gce-c1', region: 'us-central1' }, + (entry) => entries.push(entry) + ) + for (let ms = 1; ms <= 100; ms++) observability.recordHostPreAuth('db', 'admitted', ms) + observability.recordHostPreAuth('db', 'abandoned', 6_000) + observability.recordHostPreAuth('beforeCheck', 'refused', 2_000) + observability.recordHostPreAuth('reserve', 'admitted', 12.3456) + observability.flush(counts) + observability.flush(counts) + + const none = { admitted: { count: 0 }, refused: { count: 0 }, abandoned: { count: 0 } } + const buckets = (filled: Partial>): Record => ({ + '50': 0, + '100': 0, + '200': 0, + '300': 0, + '450': 0, + '700': 0, + '1000': 0, + '2000': 0, + '5000': 0, + '+Inf': 0, + ...filled + }) + expect(entries[0]!.preAuthLatencyMs).toEqual({ + beforeCheck: { + ...none, + refused: { + count: 1, + p50: 2_000, + p95: 2_000, + p99: 2_000, + max: 2_000, + bucketsLeMs: buckets({ '2000': 1 }) + } + }, + db: { + ...none, + admitted: { + count: 100, + p50: 50, + p95: 95, + p99: 99, + max: 100, + bucketsLeMs: buckets({ '50': 50, '100': 50 }) + }, + abandoned: { + count: 1, + p50: 6_000, + p95: 6_000, + p99: 6_000, + max: 6_000, + bucketsLeMs: buckets({ '+Inf': 1 }) + } + }, + reserve: { + ...none, + admitted: { + count: 1, + p50: 12.346, + p95: 12.346, + p99: 12.346, + max: 12.346, + bucketsLeMs: buckets({ '50': 1 }) + } + } + }) + // The window resets, and an empty window publishes counts only. + expect(entries[1]!.preAuthLatencyMs).toEqual({ beforeCheck: none, db: none, reserve: none }) + }) + + it('counts every hello in the pre-auth buckets, inclusive of each bound', () => { + const entries: Array> = [] + const observability = new RelayObservability( + { role: 'cell', cellId: 'staging-gce-c1', region: 'us-central1' }, + (entry) => entries.push(entry) + ) + const flooded = PRE_AUTH_RESERVOIR_LIMIT * 5 + for (let sample = 0; sample < flooded; sample++) { + observability.recordHostPreAuth('reserve', 'admitted', 450) + } + observability.recordHostPreAuth('reserve', 'admitted', 450.001) + observability.recordHostPreAuth('reserve', 'admitted', 9_000) + observability.flush(counts) + + expect(entries[0]).toMatchObject({ + preAuthLatencyMs: { + reserve: { + admitted: { + // Exact past the reservoir cap, even when it declined the slowest hello. + count: flooded + 2, + max: 9_000, + p50: 450, + bucketsLeMs: { '300': 0, '450': flooded, '700': 1, '5000': 0, '+Inf': 1 } + } + } + } + }) + }) + it('observes successful and failed database calls including transactions', async () => { const recordSql = vi.fn() const underlying: RelayDatabase = { diff --git a/cloud/apps/relay/src/relay-observability.ts b/cloud/apps/relay/src/relay-observability.ts index 17dd1ab3a44..d56b6ae84a1 100644 --- a/cloud/apps/relay/src/relay-observability.ts +++ b/cloud/apps/relay/src/relay-observability.ts @@ -103,6 +103,42 @@ export interface RelayRuntimeObserver { recordClientAcceptAbandoned?(stage: RelayClientAcceptStage, elapsedMs: number): void recordClientAcceptCompleted?(sample: RelayClientAcceptSample): void recordControlRtt?(rttMs: number): void + recordHostPreAuth?(path: HostPreAuthPath, outcome: HostPreAuthOutcome, durationMs: number): void +} + +// Which check decided a host hello: the cell's memory in reserve mode (a booking, a seat it +// remembers, or a lease), the database assignment check, or neither (beforeCheck: draining, +// idle cutover, a bad or late hello, the pool shed, or a check that threw). +export type HostPreAuthPath = 'beforeCheck' | 'db' | 'reserve' +// Abandoned: the host hung up before a decision, or before the cell could act on one. +export type HostPreAuthOutcome = 'admitted' | 'refused' | 'abandoned' + +// Upper bounds in ms; the last bucket is unbounded. 450 is the step-5 t_preauth gate. +const PRE_AUTH_BUCKET_BOUNDS_MS = [50, 100, 200, 300, 450, 700, 1_000, 2_000, 5_000] as const + +type PreAuthLatencyGroup = { + samplesMs: number[] + observed: number + maxMs: number + // Non-cumulative; one more slot than there are bounds, for the unbounded bucket. + bucketCounts: number[] +} +type PreAuthLatency = Record> + +const emptyPreAuthLatencyGroup = (): PreAuthLatencyGroup => ({ + samplesMs: [], + observed: 0, + maxMs: 0, + bucketCounts: Array.from({ length: PRE_AUTH_BUCKET_BOUNDS_MS.length + 1 }, () => 0) +}) + +const emptyPreAuthLatency = (): PreAuthLatency => { + const byOutcome = (): Record => ({ + admitted: emptyPreAuthLatencyGroup(), + refused: emptyPreAuthLatencyGroup(), + abandoned: emptyPreAuthLatencyGroup() + }) + return { beforeCheck: byOutcome(), db: byOutcome(), reserve: byOutcome() } } // Which serialized accept step the phone had already hung up behind. @@ -170,11 +206,29 @@ type RelayMetricDeltas = { controlActivityRecoveries: number controlActivityRecoveryFailures: number hostHellosShed: number + preAuthLatency: PreAuthLatency } // A host chooses how often it answers a ping, so the process-wide window is a // reservoir: the heap cost of a flood is capped and the percentiles stay unbiased. export const CONTROL_RTT_RESERVOIR_LIMIT = 1024 +// Per path and outcome; a reconnect burst can put thousands of hellos in one window. +export const PRE_AUTH_RESERVOIR_LIMIT = 1024 + +// Algorithm R: every sample in the window keeps an equal chance of being kept. +function keepInReservoir( + samples: number[], + observedBefore: number, + value: number, + limit: number +): void { + if (samples.length < limit) { + samples.push(value) + return + } + const slot = Math.floor(Math.random() * (observedBefore + 1)) + if (slot < limit) samples[slot] = value +} type MetricWriter = (entry: Record) => void @@ -227,7 +281,8 @@ const emptyDeltas = (): RelayMetricDeltas => ({ controlRenewalFlushRowsMax: 0, controlActivityRecoveries: 0, controlActivityRecoveryFailures: 0, - hostHellosShed: 0 + hostHellosShed: 0, + preAuthLatency: emptyPreAuthLatency() }) function ascending(values: number[]): number[] { @@ -252,12 +307,18 @@ function roundMs(value: number): number { // enough of it, so the maximum is folded instead. The fold is also not // interchangeable with the sorted last element: it is seeded with zero, so an // all-negative or NaN window reads differently. -function latencySummary(samples: number[]): { p50: number; p95: number; max: number } { - // One sorted copy serves both ranks. +function latencySummary(samples: number[]): { + p50: number + p95: number + p99: number + max: number +} { + // One sorted copy serves every rank. const sorted = samples.length === 0 ? samples : ascending(samples) return { p50: roundMs(nearestRank(sorted, 0.5)), p95: roundMs(nearestRank(sorted, 0.95)), + p99: roundMs(nearestRank(sorted, 0.99)), max: roundMs(samples.reduce((highest, sample) => Math.max(highest, sample), 0)) } } @@ -433,15 +494,23 @@ export class RelayObservability implements RelayRuntimeObserver { } recordControlRtt(rttMs: number): void { - const samples = this.deltas.controlRttSamplesMs const observedBefore = this.deltas.controlRttObserved++ - if (samples.length < CONTROL_RTT_RESERVOIR_LIMIT) { - samples.push(rttMs) - return - } - // Algorithm R: every round trip in the window keeps an equal chance of being kept. - const slot = Math.floor(Math.random() * (observedBefore + 1)) - if (slot < CONTROL_RTT_RESERVOIR_LIMIT) samples[slot] = rttMs + keepInReservoir( + this.deltas.controlRttSamplesMs, + observedBefore, + rttMs, + CONTROL_RTT_RESERVOIR_LIMIT + ) + } + + recordHostPreAuth(path: HostPreAuthPath, outcome: HostPreAuthOutcome, durationMs: number): void { + const group = this.deltas.preAuthLatency[path][outcome] + const observedBefore = group.observed++ + // Folded outside the reservoir so a dropped sample cannot hide the worst hello. + group.maxMs = Math.max(group.maxMs, durationMs) + const bucket = PRE_AUTH_BUCKET_BOUNDS_MS.findIndex((bound) => durationMs <= bound) + group.bucketCounts[bucket === -1 ? PRE_AUTH_BUCKET_BOUNDS_MS.length : bucket]! += 1 + keepInReservoir(group.samplesMs, observedBefore, durationMs, PRE_AUTH_RESERVOIR_LIMIT) } start(readCounts: () => RelayProcessCounts, intervalMs = 30_000): void { @@ -584,6 +653,7 @@ export class RelayObservability implements RelayRuntimeObserver { controlActivityRecoveriesDelta: deltas.controlActivityRecoveries, controlActivityRecoveryFailuresDelta: deltas.controlActivityRecoveryFailures, hostHellosShedDelta: deltas.hostHellosShed, + preAuthLatencyMs: preAuthLatencyFields(deltas.preAuthLatency), // Meaning changed when renewals began batching: for a batched row this is // the flush's duration, not that row's own statement latency. The // per-flush fields below are the ones to read for statement cost. @@ -619,6 +689,35 @@ function regionCounterFields( ) } +// Count and buckets cover every hello in the window, so buckets sum exactly across windows and +// cells; the percentiles summarise the reservoir and are omitted when empty. +function preAuthLatencyFields( + latency: PreAuthLatency +): Record>> { + const summarize = (group: PreAuthLatencyGroup): Record => { + if (group.observed === 0) return { count: 0 } + const { p50, p95, p99 } = latencySummary(group.samplesMs) + const bucketsLeMs: Record = {} + PRE_AUTH_BUCKET_BOUNDS_MS.forEach((bound, index) => { + bucketsLeMs[String(bound)] = group.bucketCounts[index]! + }) + bucketsLeMs['+Inf'] = group.bucketCounts[PRE_AUTH_BUCKET_BOUNDS_MS.length]! + return { count: group.observed, p50, p95, p99, max: roundMs(group.maxMs), bucketsLeMs } + } + const byOutcome = ( + groups: Record + ): Record> => ({ + admitted: summarize(groups.admitted), + refused: summarize(groups.refused), + abandoned: summarize(groups.abandoned) + }) + return { + beforeCheck: byOutcome(latency.beforeCheck), + db: byOutcome(latency.db), + reserve: byOutcome(latency.reserve) + } +} + function lockWaiters(byKey: Record, waiterRole: string): number { return Object.entries(byKey) .filter(([key]) => key.startsWith(`${waiterRole}:relay_cells:`))