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

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<NonNullable<RelayRuntimeObserver['recordHostPreAuth']>>()
+  const observer = {
+    recordAuth: vi.fn(),
+    recordForwardedBytes: vi.fn(),
+    recordHttp: vi.fn(),
+    recordReconnect: vi.fn(),
+    recordSql: vi.fn(),
+    recordHostPreAuth
+  } satisfies RelayRuntimeObserver
+  const take = vi.fn<CellReserveBook['take']>((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<boolean>((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'
+  }
+}

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<string>,
     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<void> {
+    // 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(

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<Record<string, unknown>> = []
+    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<string, number>>): Record<string, number> => ({
+      '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<Record<string, unknown>> = []
+    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 = {

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<HostPreAuthPath, Record<HostPreAuthOutcome, PreAuthLatencyGroup>>
+
+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<HostPreAuthOutcome, PreAuthLatencyGroup> => ({
+    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<string, unknown>) => 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<HostPreAuthPath, Record<HostPreAuthOutcome, Record<string, unknown>>> {
+  const summarize = (group: PreAuthLatencyGroup): Record<string, unknown> => {
+    if (group.observed === 0) return { count: 0 }
+    const { p50, p95, p99 } = latencySummary(group.samplesMs)
+    const bucketsLeMs: Record<string, number> = {}
+    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<HostPreAuthOutcome, PreAuthLatencyGroup>
+  ): Record<HostPreAuthOutcome, Record<string, unknown>> => ({
+    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<string, number>, waiterRole: string): number {
   return Object.entries(byKey)
     .filter(([key]) => key.startsWith(`${waiterRole}:relay_cells:`))