fix(relay): log controls a cell closes after host proof but before wiring (#30559)

* fix(relay): log controls a cell closes after host proof but before wiring

* fix(relay): log a pre-refusal hang-up as the hang-up and share printableCloseReason

* refactor(relay): move printableCloseReason into its own module to break the import cycle

* test(relay): assemble the fake pg URL at runtime so the secret scan passes
This commit is contained in:
Jinwoo Hong
2026-10-19 00:36:10 -04:00
committed by GitHub
parent 9a662c3980
commit 535e995565
7 changed files with 380 additions and 16 deletions
@@ -0,0 +1,7 @@
// Peer-supplied close reasons are logged; keep them printable and short.
export function printableCloseReason(reason: Buffer | string): string {
return reason
.toString()
.replace(/[^\x20-\x7e]/g, '')
.slice(0, 80)
}
@@ -7,7 +7,7 @@ import {
RELAY_HOST_CAPABILITY_IDLE_REGIONAL_REHOME,
RELAY_PROTOCOL_LIMITS
} from '@orca-cloud/relay-contract'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { afterEach, beforeEach, describe, expect, it, vi, type MockInstance } from 'vitest'
import type WebSocket from 'ws'
import type { RelayAssignmentStore } from './assignment-store.js'
import type { RelayConfig } from './config.js'
@@ -130,6 +130,7 @@ function createRegistry(
observer: {
recordAuth: ReturnType<typeof vi.fn>
recordControlClose: ReturnType<typeof vi.fn>
recordPostProofClose: ReturnType<typeof vi.fn>
recordSpliceClose: ReturnType<typeof vi.fn>
}
} {
@@ -177,6 +178,7 @@ function createRegistry(
recordReconnect: vi.fn(),
recordSql: vi.fn(),
recordControlClose: vi.fn(),
recordPostProofClose: vi.fn(),
recordSpliceClose: vi.fn()
} satisfies RelayRuntimeObserver
const registry = new HostSessionRegistry(
@@ -2038,3 +2040,162 @@ describe('paced drain', () => {
expect(sockets.map((socket) => socket.send.mock.calls.length)).toEqual(sendsAfterEmergency)
})
})
describe('controls closed after host proof but before wiring', () => {
let warn: MockInstance<typeof console.warn>
beforeEach(() => {
vi.useFakeTimers()
warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
})
afterEach(() => {
warn.mockRestore()
vi.clearAllTimers()
vi.useRealTimers()
})
const refusedLines = (): string[] =>
warn.mock.calls
.map((call) => String(call[0]))
.filter((line) => line.includes('control refused after proof'))
function expectRefusal(
observer: { recordPostProofClose: ReturnType<typeof vi.fn> },
code: number | null,
reason: string,
cause?: string
): void {
expect(observer.recordPostProofClose).toHaveBeenCalledWith(reason)
const [line] = refusedLines()
expect(line).toContain(`host=${relayHostLogDigest(identity.relayHostId)} gen=1 app="1.4.173"`)
expect(line).toContain(`code=${code ?? 'none'} reason=${JSON.stringify(reason)}`)
if (cause) expect(line).toContain(`cause=${cause}`)
expect(line).not.toContain(identity.relayHostId)
expect(line).not.toContain(identity.sub)
}
it('logs an idle cutover refusal', async () => {
const h = createRegistry(vi.fn())
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: beginIdleWork only checks the host key's presence.
const idle = h.registry as unknown as { idleAttempts: Map<string, unknown> }
idle.idleAttempts.set(identity.relayHostId, {})
const socket = new FakeSocket()
await h.activate(socket as unknown as WebSocket, identity, null, 1, false, 1)
expect(socket.close).toHaveBeenCalledWith(
RELAY_CLOSE_CODE.WRONG_CELL,
'idle cutover in progress'
)
expectRefusal(h.observer, RELAY_CLOSE_CODE.WRONG_CELL, 'idle cutover in progress')
})
it('logs a stalled activation queue once', async () => {
const stalled = deferred<string>()
const h = createRegistry(
vi.fn<RelayAssignmentStore['activateControl']>().mockReturnValueOnce(stalled.promise)
)
const first = h.activate(new FakeSocket() as unknown as WebSocket, identity, null, 1, false, 1)
const waiting = new FakeSocket()
const second = h.activate(waiting as unknown as WebSocket, identity, null, 1, false, 1)
await Promise.resolve()
await Promise.resolve()
vi.advanceTimersByTime(30_000)
expectRefusal(h.observer, RELAY_CLOSE_CODE.LIMIT_EXCEEDED, 'control activation queue stalled')
stalled.resolve('control:production-gce-c3:1')
await Promise.all([first, second])
expect(h.observer.recordPostProofClose).toHaveBeenCalledOnce()
})
it('logs a superseded activation', async () => {
const h = createRegistry(vi.fn())
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: only compared by identity against the empty session map.
const stale = {} as HostSession
await h.activate(new FakeSocket() as unknown as WebSocket, identity, stale, 1, false, 1)
expectRefusal(h.observer, RELAY_CLOSE_CODE.PEER_DROPPED, 'control activation superseded')
})
it.each([
{ error: new Error('wrong_assignment'), cause: 'wrong_assignment' },
{ error: Object.assign(new Error('timeout'), { code: '57014' }), cause: 'database_transient' },
// Assembled at runtime so the secret scanner does not read the fixture as a real URL.
{
error: new Error(`connect ${['postgres', '//user:secret@db/relay'].join(':')}`),
cause: 'error'
}
])('logs a failed host-proof assignment check as $cause', async ({ error, cause }) => {
const h = createRegistry(
vi.fn<RelayAssignmentStore['activateControl']>().mockRejectedValueOnce(error)
)
await h.activate(new FakeSocket() as unknown as WebSocket, identity, null, 1, false, 1)
expectRefusal(
h.observer,
RELAY_CLOSE_CODE.WRONG_CELL,
'assignment changed during host proof',
cause
)
expect(refusedLines()[0]).not.toContain('secret')
})
it('logs a drain that lands during activation', async () => {
const h = createRegistry(vi.fn().mockResolvedValue('control:production-gce-c3:1'))
h.registry.drain(100)
await h.activate(new FakeSocket() as unknown as WebSocket, identity, null, 1, false, 1)
expectRefusal(h.observer, RELAY_CLOSE_CODE.DRAINING, 'relay draining')
})
it('logs a host that hangs up during activation, with no cell close code', async () => {
const blocked = deferred<string>()
const h = createRegistry(
vi.fn<RelayAssignmentStore['activateControl']>().mockReturnValueOnce(blocked.promise)
)
const socket = new FakeSocket()
const activation = h.activate(socket as unknown as WebSocket, identity, null, 1, false, 1)
await vi.advanceTimersByTimeAsync(0)
socket.close()
blocked.resolve('control:production-gce-c3:1')
await activation
expectRefusal(h.observer, null, 'host closed during activation')
})
it('logs a host that hangs up while queued, with no cell close code', async () => {
const stalled = deferred<string>()
const h = createRegistry(
vi
.fn<RelayAssignmentStore['activateControl']>()
.mockReturnValueOnce(stalled.promise)
.mockResolvedValue('control:production-gce-c3:1')
)
const first = h.activate(new FakeSocket() as unknown as WebSocket, identity, null, 1, false, 1)
const queued = new FakeSocket()
const second = h.activate(queued as unknown as WebSocket, identity, null, 1, false, 1)
queued.close()
stalled.resolve('control:production-gce-c3:1')
await Promise.all([first, second])
expectRefusal(h.observer, null, 'host closed while queued')
})
it('logs the hang-up, not a stall, when the queued host left before the deadline', async () => {
const stalled = deferred<string>()
const h = createRegistry(
vi.fn<RelayAssignmentStore['activateControl']>().mockReturnValueOnce(stalled.promise)
)
const first = h.activate(new FakeSocket() as unknown as WebSocket, identity, null, 1, false, 1)
const queued = new FakeSocket()
const second = h.activate(queued as unknown as WebSocket, identity, null, 1, false, 1)
await Promise.resolve()
await Promise.resolve()
queued.readyState = queued.CLOSED
vi.advanceTimersByTime(30_000)
expect(queued.close).toHaveBeenCalledWith(
RELAY_CLOSE_CODE.LIMIT_EXCEEDED,
'control activation queue stalled'
)
expectRefusal(h.observer, null, 'host closed while queued')
stalled.resolve('control:production-gce-c3:1')
await Promise.all([first, second])
})
it('does not log a control that was wired', async () => {
const h = createRegistry(vi.fn().mockResolvedValue('control:production-gce-c3:1'))
await h.activate(new FakeSocket() as unknown as WebSocket, identity, null, 1, false, 1)
expect(h.observer.recordPostProofClose).not.toHaveBeenCalled()
expect(refusedLines()).toEqual([])
})
})
+70 -15
View File
@@ -35,9 +35,15 @@ import type { CellAdmitMode, CellReserveBook } from './cell-reserve-book.js'
import type { DemoteRequest, ReserveOutcome, ReserveRequest } from './cell-reserve-contract.js'
import { CellSeatLog, type CellSeatChange, type CellSeatFeedPage } from './cell-seat-log.js'
import { isRelayDatabaseTransientError } from './database.js'
import { printableCloseReason } from './control-close-reason-text.js'
import { ControlRenewalBatch } from './control-renewal-batch.js'
import { RelayCredentialStore, type CredentialReservation } from './credential-store.js'
import { HostCloseReasonMemory } from './host-close-reason-memory.js'
import {
postProofCloseCause,
PostProofCloseLog,
type PostProofCloseCause
} from './post-proof-close-log.js'
import { relayHostLogDigest } from './relay-host-log-digest.js'
import type { RelayTokenClaims } from './relay-token-verifier.js'
import {
@@ -52,14 +58,6 @@ import type { PendingHostDataReservation } from './relay-connection-ledger.js'
import { closeRelayWebSocket } from './relay-websocket-close.js'
import { ProcessQueuedByteBudget, wireSplice } from './splice-forwarder.js'
// Peer-supplied close reasons are logged; keep them printable and short.
function printableCloseReason(reason: Buffer | string): string {
return reason
.toString()
.replace(/[^\x20-\x7e]/g, '')
.slice(0, 80)
}
type VerifyRelayToken = (token: string) => Promise<RelayTokenClaims | null>
type HostState = 'proving' | 'active' | 'orphaned' | 'drain-only' | 'closed'
@@ -299,6 +297,7 @@ export class HostSessionRegistry {
// draining" is not the same question as "this host has been told to leave".
private readonly drainSentHosts = new Set<string>()
private readonly seatLog: CellSeatLog
private readonly postProofCloses: PostProofCloseLog
// Seat key -> the demoted control socket. Its own set, not drainSentHosts: it ends with
// that socket (close, rebind or a new join), so a later booking here is a fresh seat.
private readonly demotedSeats = new Map<string, WebSocket>()
@@ -442,6 +441,7 @@ export class HostSessionRegistry {
private readonly reserveAdmission?: CellReserveAdmission
) {
this.seatLog = new CellSeatLog(undefined, now)
this.postProofCloses = new PostProofCloseLog(observer)
}
// Reserve mode is only ever a cell's, and only while its applied switch says so.
@@ -1627,9 +1627,15 @@ export class HostSessionRegistry {
// Present when the hello was admitted from memory: no database registration.
reserved?: { reservedBy?: string }
): Promise<void> {
const refused = { relayHostId: identity.relayHostId, generation, appVersion }
const release = this.beginIdleWork(identity.relayHostId)
if (!release) {
socket.close(RELAY_CLOSE_CODE.WRONG_CELL, 'idle cutover in progress')
this.refuseProvenControl(
socket,
refused,
RELAY_CLOSE_CODE.WRONG_CELL,
'idle cutover in progress'
)
return Promise.resolve()
}
const key = this.key(identity.sub, identity.relayHostId)
@@ -1639,16 +1645,36 @@ export class HostSessionRegistry {
let queueWaitExpired = false
const queueWaitTimer = setTimeout(() => {
queueWaitExpired = true
socket.close(RELAY_CLOSE_CODE.LIMIT_EXCEEDED, 'control activation queue stalled')
this.refuseProvenControl(
socket,
refused,
RELAY_CLOSE_CODE.LIMIT_EXCEEDED,
'control activation queue stalled',
undefined,
'host closed while queued'
)
}, ACTIVATION_QUEUE_WAIT_MS)
queueWaitTimer.unref?.()
const activation = previous
.catch(() => undefined)
.then(async () => {
clearTimeout(queueWaitTimer)
if (queueWaitExpired || socket.readyState !== socket.OPEN) return
if (queueWaitExpired) return
if (socket.readyState !== socket.OPEN) {
this.postProofCloses.record({
...refused,
code: null,
reason: 'host closed while queued'
})
return
}
if ((this.sessions.get(key) ?? null) !== existing) {
socket.close(RELAY_CLOSE_CODE.PEER_DROPPED, 'control activation superseded')
this.refuseProvenControl(
socket,
refused,
RELAY_CLOSE_CODE.PEER_DROPPED,
'control activation superseded'
)
return
}
await this.activateCurrent(
@@ -1683,6 +1709,7 @@ export class HostSessionRegistry {
connectionInclusionWatermark?: number,
reserved?: { reservedBy?: string }
): Promise<void> {
const refused = { relayHostId: identity.relayHostId, generation, appVersion }
let controlActivityId: string | null = null
// Admitted from memory, but the cell left reserve mode before the proof landed: lease now.
if (this.config.role === 'cell' && !(reserved && this.reserveMode())) {
@@ -1704,7 +1731,7 @@ export class HostSessionRegistry {
{ userId: identity.sub, relayHostId: identity.relayHostId },
{ cellId: this.config.cellId, assignmentEpoch }
)
} catch {
} catch (error) {
// A failure after activateControl succeeded must not strand the
// acquired control activity until lease expiry.
if (controlActivityId) {
@@ -1713,7 +1740,13 @@ export class HostSessionRegistry {
controlActivityId
)
}
socket.close(RELAY_CLOSE_CODE.WRONG_CELL, 'assignment changed during host proof')
this.refuseProvenControl(
socket,
refused,
RELAY_CLOSE_CODE.WRONG_CELL,
'assignment changed during host proof',
postProofCloseCause(error, isRelayDatabaseTransientError)
)
return
}
}
@@ -1725,7 +1758,13 @@ export class HostSessionRegistry {
)
}
if (socket.readyState === socket.OPEN) {
socket.close(RELAY_CLOSE_CODE.DRAINING, 'relay draining')
this.refuseProvenControl(socket, refused, RELAY_CLOSE_CODE.DRAINING, 'relay draining')
} else {
this.postProofCloses.record({
...refused,
code: null,
reason: 'host closed during activation'
})
}
return
}
@@ -1828,6 +1867,22 @@ export class HostSessionRegistry {
this.appendSeatChange(session, 'join', undefined, reserved?.reservedBy)
}
private refuseProvenControl(
socket: WebSocket,
refused: { relayHostId: string; generation: number; appVersion: string },
code: number,
reason: string,
cause?: PostProofCloseCause,
hungUpReason = 'host closed during activation'
): void {
// A host that already hung up got no close frame from us; log what happened instead.
const open = socket.readyState === socket.OPEN
socket.close(code, reason)
this.postProofCloses.record(
open ? { ...refused, code, reason, cause } : { ...refused, code: null, reason: hungUpReason }
)
}
seatFeed(sinceSeq: number | null): CellSeatFeedPage {
return { ...this.seatLog.read(sinceSeq), seats: this.seatLog.seatCount() }
}
@@ -0,0 +1,63 @@
import { describe, expect, it, vi } from 'vitest'
import { POST_PROOF_CLOSE_LOG_WINDOW_MS, PostProofCloseLog } from './post-proof-close-log.js'
const close = {
relayHostId: 'abcdefghijklmnop',
generation: 7,
appVersion: '1.4.150',
code: 4409,
reason: 'assignment changed during host proof',
cause: 'wrong_assignment' as const
}
function createLog() {
const lines: string[] = []
const scheduled: Array<{ callback: () => void; delayMs: number }> = []
const observer = {
recordAuth: vi.fn(),
recordForwardedBytes: vi.fn(),
recordHttp: vi.fn(),
recordReconnect: vi.fn(),
recordSql: vi.fn(),
recordPostProofClose: vi.fn()
}
const log = new PostProofCloseLog(
observer,
(line) => lines.push(line),
(callback, delayMs) => {
scheduled.push({ callback, delayMs })
return () => undefined
}
)
return { log, lines, scheduled, observer }
}
describe('post-proof close log', () => {
it('counts every close but writes one line per reason and cause per window', () => {
const { log, lines, scheduled, observer } = createLog()
for (let index = 0; index < 1_000; index++) log.record({ ...close, generation: index })
log.record({ ...close, cause: 'database_transient' })
log.record({ ...close, code: 4400, reason: 'relay draining', cause: undefined })
expect(observer.recordPostProofClose).toHaveBeenCalledTimes(1_002)
expect(lines).toHaveLength(3)
expect(scheduled.map((entry) => entry.delayMs)).toEqual(
Array.from({ length: 3 }, () => POST_PROOF_CLOSE_LOG_WINDOW_MS)
)
for (const { callback } of scheduled) callback()
// Only the window that held closes back writes a closing line, naming the latest.
expect(lines).toHaveLength(4)
expect(lines[3]).toContain('gen=999')
expect(lines[3]).toContain('suppressed=999')
})
it('names the host by digest and keeps peer-supplied versions printable', () => {
const { log, lines } = createLog()
log.record({ ...close, appVersion: '1.4.150\n\u0007injected' })
expect(lines[0]).toMatch(
/^\[orca-relay\] control refused after proof host=\S+ gen=7 app="1\.4\.150injected" code=4409 reason="assignment changed during host proof" cause=wrong_assignment$/
)
expect(lines[0]).not.toContain(close.relayHostId)
})
})
@@ -0,0 +1,65 @@
import { AssignmentRejectionLogWindow } from './assignment-rejection-log-window.js'
import { printableCloseReason } from './control-close-reason-text.js'
import { relayHostLogDigest } from './relay-host-log-digest.js'
import type { RelayRuntimeObserver } from './relay-observability.js'
// A looping host hits a post-proof close about once a second; per reason and
// cause, one line opens the window and one closing line carries the rest.
export const POST_PROOF_CLOSE_LOG_WINDOW_MS = 60_000
// Why the assignment check failed: the close reason alone cannot tell a moved
// assignment from a database timeout. Never the raw message (pg errors carry URLs).
export type PostProofCloseCause = 'wrong_assignment' | 'database_transient' | 'error'
export type PostProofClose = {
relayHostId: string
generation: number
appVersion: string
// Null when the host hung up itself: the cell sent no close frame.
code: number | null
reason: string
cause?: PostProofCloseCause
}
// Controls closed between host proof and wiring never reach the 'control closed' listener.
export class PostProofCloseLog {
private readonly window: AssignmentRejectionLogWindow<PostProofClose>
constructor(
private readonly observer: RelayRuntimeObserver,
private readonly warn: (line: string) => void = (line) => console.warn(line),
schedule?: (callback: () => void, delayMs: number) => () => void
) {
this.window = new AssignmentRejectionLogWindow<PostProofClose>({
windowMs: POST_PROOF_CLOSE_LOG_WINDOW_MS,
onWindowClosed: ({ suppressed, sample }) => this.warn(formatLine(sample, suppressed)),
schedule
})
}
record(close: PostProofClose): void {
this.observer.recordPostProofClose?.(close.reason)
if (!this.window.admit(`${close.reason}:${close.cause ?? ''}`, close)) return
this.warn(formatLine(close, 0))
}
}
export function postProofCloseCause(
error: unknown,
isTransient: (error: unknown) => boolean
): PostProofCloseCause {
if (error instanceof Error && error.message === 'wrong_assignment') return 'wrong_assignment'
return isTransient(error) ? 'database_transient' : 'error'
}
function formatLine(close: PostProofClose, suppressed: number): string {
return (
`[orca-relay] control refused after proof host=${relayHostLogDigest(close.relayHostId)}` +
` gen=${close.generation}` +
` app=${JSON.stringify(printableCloseReason(close.appVersion))}` +
` code=${close.code ?? 'none'} reason=${JSON.stringify(close.reason)}` +
(close.cause ? ` cause=${close.cause}` : '') +
// Closing line only: how many more closes the window held back; the fields name the latest.
(suppressed > 0 ? ` suppressed=${suppressed}` : '')
)
}
@@ -508,6 +508,8 @@ describe('relay observability', () => {
observability.recordControlClose(1006)
observability.recordControlClose(1006)
observability.recordControlClose(4402)
observability.recordPostProofClose('relay draining')
observability.recordPostProofClose('relay draining')
observability.recordSpliceClose('host-oversize-frame')
observability.recordSpliceClose('queue-limit')
observability.recordClientAcceptAbandoned('activity', 14_250.4)
@@ -518,12 +520,14 @@ describe('relay observability', () => {
expect(entries[0]).toMatchObject({
controlClosesByCodeDelta: { 1006: 2, 4402: 1 },
postProofClosesByReasonDelta: { 'relay draining': 2 },
spliceClosesByTriggerDelta: { 'host-oversize-frame': 1, 'queue-limit': 1 },
clientAcceptsAbandonedByStageDelta: { activity: 2, credential: 1 },
clientAcceptAbandonedMsMax: 14_250.4
})
expect(entries[1]).toMatchObject({
controlClosesByCodeDelta: {},
postProofClosesByReasonDelta: {},
spliceClosesByTriggerDelta: {},
clientAcceptsAbandonedByStageDelta: {},
clientAcceptAbandonedMsMax: 0
@@ -101,6 +101,7 @@ export interface RelayRuntimeObserver {
fallback: boolean
}): void
recordControlClose?(code: number): void
recordPostProofClose?(reason: string): void
recordSpliceClose?(trigger: string): void
recordClientAcceptAbandoned?(stage: RelayClientAcceptStage, elapsedMs: number): void
recordClientAcceptCompleted?(sample: RelayClientAcceptSample): void
@@ -200,6 +201,7 @@ type RelayMetricDeltas = {
regionFallbacks: Record<string, number>
unavailableRegions: Record<string, number>
controlClosesByCode: Record<string, number>
postProofClosesByReason: Record<string, number>
spliceClosesByTrigger: Record<string, number>
clientAcceptsAbandonedByStage: Record<string, number>
clientAcceptAbandonedMsMax: number
@@ -288,6 +290,7 @@ const emptyDeltas = (): RelayMetricDeltas => ({
regionFallbacks: {},
unavailableRegions: {},
controlClosesByCode: {},
postProofClosesByReason: {},
spliceClosesByTrigger: {},
clientAcceptsAbandonedByStage: {},
clientAcceptAbandonedMsMax: 0,
@@ -529,6 +532,10 @@ export class RelayObservability implements RelayRuntimeObserver {
this.deltas.controlClosesByCode[key] = (this.deltas.controlClosesByCode[key] ?? 0) + 1
}
recordPostProofClose(reason: string): void {
increment(this.deltas.postProofClosesByReason, reason)
}
recordSpliceClose(trigger: string): void {
this.deltas.spliceClosesByTrigger[trigger] =
(this.deltas.spliceClosesByTrigger[trigger] ?? 0) + 1
@@ -670,6 +677,8 @@ export class RelayObservability implements RelayRuntimeObserver {
regionFallbacksDelta: deltas.regionFallbacks,
unavailableRegionsDelta: deltas.unavailableRegions,
controlClosesByCodeDelta: deltas.controlClosesByCode,
// Closes after a host proof but before its control was wired; none is in the line above.
postProofClosesByReasonDelta: deltas.postProofClosesByReason,
spliceClosesByTriggerDelta: deltas.spliceClosesByTrigger,
clientAcceptsAbandonedByStageDelta: deltas.clientAcceptsAbandonedByStage,
clientAcceptAbandonedMsMax: roundMs(deltas.clientAcceptAbandonedMsMax),