diff --git a/src/main/ipc/runtime-environments/runtime-environment-control-socket-close-trace.test.ts b/src/main/ipc/runtime-environments/runtime-environment-control-socket-close-trace.test.ts new file mode 100644 index 00000000000..5a600c07b6c --- /dev/null +++ b/src/main/ipc/runtime-environments/runtime-environment-control-socket-close-trace.test.ts @@ -0,0 +1,36 @@ +import { afterEach, expect, it } from 'vitest' +import { setActiveSink } from '../../observability/tracer' +import { sharedControlEnvironmentPolicy } from './runtime-environment-shared-control-policy' + +afterEach(() => setActiveSink(null)) + +it('traces each control socket close for the support bundle', () => { + const records: unknown[] = [] + setActiveSink({ push: (record) => records.push(record), flush() {}, close() {} }) + + sharedControlEnvironmentPolicy('env-1').onSocketClosed?.({ + cause: 'keepalive_timeout', + errorCode: 'remote_runtime_unavailable', + closeCode: null, + state: 'ready', + lastRttMs: 320, + socketUptimeMs: 60_000, + readyUptimeMs: 59_000, + reconnectAttempt: 0, + pendingRequestCount: 1, + subscriptionCount: 4 + }) + + expect(records).toEqual([ + expect.objectContaining({ + type: 'effect-span', + name: 'remote-runtime.control-socket-close', + attributes: expect.objectContaining({ + environmentId: 'env-1', + cause: 'keepalive_timeout', + lastRttMs: 320, + socketUptimeMs: 60_000 + }) + }) + ]) +}) diff --git a/src/main/ipc/runtime-environments/runtime-environment-control-socket-close-trace.ts b/src/main/ipc/runtime-environments/runtime-environment-control-socket-close-trace.ts new file mode 100644 index 00000000000..3a03bfea175 --- /dev/null +++ b/src/main/ipc/runtime-environments/runtime-environment-control-socket-close-trace.ts @@ -0,0 +1,13 @@ +import { startSpan } from '../../observability/tracer' +import type { SharedControlSocketCloseRecord } from '../../../shared/remote-runtime/control/remote-runtime-shared-control-types' + +/** One trace span per closed control socket, so a support bundle shows why a host reconnected. */ +export function traceSharedControlSocketClose( + environmentId: string, + record: SharedControlSocketCloseRecord +): void { + startSpan('remote-runtime.control-socket-close', { + kind: 'client', + attributes: { environmentId, ...record } + }).end() +} diff --git a/src/main/ipc/runtime-environments/runtime-environment-shared-control-policy.ts b/src/main/ipc/runtime-environments/runtime-environment-shared-control-policy.ts index 02205df5099..4b7b5d5e9a0 100644 --- a/src/main/ipc/runtime-environments/runtime-environment-shared-control-policy.ts +++ b/src/main/ipc/runtime-environments/runtime-environment-shared-control-policy.ts @@ -2,17 +2,19 @@ import type { RemoteRuntimeSharedControlConnectionOptions } from '../../../share import { isRuntimeEnvironmentCapabilityPaused } from './runtime-environment-capability-evidence' import { isRuntimeEnvironmentManuallyDisconnected } from './runtime-environment-manual-disconnect' import { handleSharedControlRefusal } from './runtime-environment-pairing-refusal-registry' +import { traceSharedControlSocketClose } from './runtime-environment-control-socket-close-trace' -/** When an environment's shared control socket may reconnect, and who answers a refused token. */ +/** When a shared control socket may reconnect, who answers a refused token, and where closes go. */ export function sharedControlEnvironmentPolicy( environmentId: string ): Pick< RemoteRuntimeSharedControlConnectionOptions, - 'isManuallyDisconnected' | 'isCapabilityPaused' | 'onAuthenticationRejected' + 'isManuallyDisconnected' | 'isCapabilityPaused' | 'onAuthenticationRejected' | 'onSocketClosed' > { return { isManuallyDisconnected: () => isRuntimeEnvironmentManuallyDisconnected(environmentId), isCapabilityPaused: () => isRuntimeEnvironmentCapabilityPaused(environmentId), - onAuthenticationRejected: () => handleSharedControlRefusal(environmentId) + onAuthenticationRejected: () => handleSharedControlRefusal(environmentId), + onSocketClosed: (record) => traceSharedControlSocketClose(environmentId, record) } } diff --git a/src/main/runtime/rpc/node-websocket-lifecycle.ts b/src/main/runtime/rpc/node-websocket-lifecycle.ts index e37e38375da..8e235cc6ae2 100644 --- a/src/main/runtime/rpc/node-websocket-lifecycle.ts +++ b/src/main/runtime/rpc/node-websocket-lifecycle.ts @@ -32,7 +32,7 @@ export function attachNodeWebSocketLifecycle(args: { }): void { const { ws } = args let finalized = false - const onPong = (): void => args.heartbeat.noteAlive(ws) + const onPong = (): void => args.heartbeat.notePong(ws) const onMessage = (data: WebSocket.RawData, isBinary: boolean): void => { // Why: any inbound frame counts as proof of life, so an actively-talking client isn't reaped mid-request. args.heartbeat.noteAlive(ws) @@ -89,6 +89,7 @@ export function attachNodeWebSocketLifecycle(args: { ws.on('error', onError) // Why: install lifecycle ownership before periodic heartbeat ticks can observe this socket. args.heartbeatConnections.add(ws) + args.heartbeat.track(ws) args.heartbeat.noteAlive(ws) if (args.heartbeatConnections.size === 1) { // Unauthenticated sockets are protected by the pre-auth timeout; heartbeat probes begin only diff --git a/src/main/runtime/rpc/remote-runtime-server-heartbeat.test.ts b/src/main/runtime/rpc/remote-runtime-server-heartbeat.test.ts index 5965da30e6c..99f5efe445b 100644 --- a/src/main/runtime/rpc/remote-runtime-server-heartbeat.test.ts +++ b/src/main/runtime/rpc/remote-runtime-server-heartbeat.test.ts @@ -1,8 +1,10 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import type { WebSocket } from 'ws' import { RemoteRuntimeServerHeartbeat } from './remote-runtime-server-heartbeat' +import { setActiveSink } from '../../observability/tracer' afterEach(() => { + setActiveSink(null) vi.useRealTimers() }) @@ -83,4 +85,40 @@ describe('RemoteRuntimeServerHeartbeat', () => { expect(socket.terminate).toHaveBeenCalledTimes(1) heartbeat.stop() }) + + it('writes a trace span for each reap with the socket uptime and last round trip', async () => { + vi.useFakeTimers() + const records: unknown[] = [] + setActiveSink({ push: (record) => records.push(record), flush() {}, close() {} }) + let now = 1_000 + // SAFETY: the heartbeat only calls ping() and terminate() on its sockets. + const socket = { ping: vi.fn(), terminate: vi.fn() } as unknown as WebSocket + const heartbeat = new RemoteRuntimeServerHeartbeat(100, () => now, 128, 1) + heartbeat.track(socket) + heartbeat.noteAlive(socket) + heartbeat.start(() => [socket]) + now += 100 + await vi.advanceTimersByTimeAsync(100) + now += 40 + heartbeat.notePong(socket) + + now += 60 + await vi.advanceTimersByTimeAsync(100) + now += 100 + await vi.advanceTimersByTimeAsync(100) + + expect(socket.terminate).toHaveBeenCalledTimes(1) + expect(records).toEqual([ + expect.objectContaining({ + name: 'remote-runtime.server-socket-reaped', + attributes: expect.objectContaining({ + cause: 'heartbeat_missed_probes', + lastRttMs: 40, + socketUptimeMs: 300, + trackedSockets: 1 + }) + }) + ]) + heartbeat.stop() + }) }) diff --git a/src/main/runtime/rpc/remote-runtime-server-heartbeat.ts b/src/main/runtime/rpc/remote-runtime-server-heartbeat.ts index 4bce033a67a..bb21a3ab768 100644 --- a/src/main/runtime/rpc/remote-runtime-server-heartbeat.ts +++ b/src/main/runtime/rpc/remote-runtime-server-heartbeat.ts @@ -1,4 +1,5 @@ import type { WebSocket } from 'ws' +import { startSpan } from '../../observability/tracer' // Why: one unanswered probe is UNKNOWN, not death — a cellular/Tailscale blackhole or a stalled TCP // retransmit routinely swallows a single pong from a peer that is still there (STA-3320). Only a run of @@ -12,6 +13,9 @@ export class RemoteRuntimeServerHeartbeat { private readonly alive = new WeakSet() // Why: consecutive probes a socket has left unanswered. Absent = zero, so live sockets cost nothing. private readonly missedProbes = new WeakMap() + private readonly trackedAt = new WeakMap() + private readonly unansweredPingAt = new WeakMap() + private readonly lastRttMs = new WeakMap() constructor( private readonly intervalMs: number, @@ -26,6 +30,23 @@ export class RemoteRuntimeServerHeartbeat { this.alive.add(socket) } + /** Starts the socket's uptime clock, reported if it is ever reaped. */ + track(socket: WebSocket): void { + this.trackedAt.set(socket, this.now()) + } + + notePong(socket: WebSocket): void { + this.noteAlive(socket) + const sentAt = this.unansweredPingAt.get(socket) + if (sentAt !== undefined) { + this.unansweredPingAt.delete(socket) + const rttMs = this.now() - sentAt + if (rttMs >= 0) { + this.lastRttMs.set(socket, rttMs) + } + } + } + start(getClients: () => Iterable): void { if (this.timer) { return @@ -54,6 +75,7 @@ export class RemoteRuntimeServerHeartbeat { // answering the probe still sent below. const stalledTick = elapsedMs < 0 || elapsedMs > this.intervalMs * 1.5 let reaped = 0 + const reapedSockets: WebSocket[] = [] let clientCount = 0 for (const socket of clients) { clientCount += 1 @@ -66,6 +88,7 @@ export class RemoteRuntimeServerHeartbeat { this.missedProbes.delete(socket) socket.terminate() reaped += 1 + reapedSockets.push(socket) continue } this.missedProbes.set(socket, missed) @@ -74,12 +97,34 @@ export class RemoteRuntimeServerHeartbeat { // Why: re-probe every sweep, including missed ones, so a path that just recovered can prove // itself on the next tick instead of waiting out the rest of the budget. socket.ping() + if (!this.unansweredPingAt.has(socket) || stalledTick) { + this.unansweredPingAt.set(socket, tickAt) + } } catch { // Why: a mid-teardown socket is finalized by its close/error listener. } } + for (const socket of reapedSockets) { + this.traceReap(socket, tickAt, clientCount) + } if (reaped > 0 || clientCount >= this.warningClientCount) { console.warn(`[ws-transport] heartbeat reaped ${reaped}; ${clientCount} tracked sockets`) } } + + // Why: the console line reaches nothing in a packaged host; a span lands in the support bundle. + private traceReap(socket: WebSocket, reapedAt: number, trackedSockets: number): void { + const trackedAt = this.trackedAt.get(socket) + startSpan('remote-runtime.server-socket-reaped', { + kind: 'server', + attributes: { + cause: 'heartbeat_missed_probes', + missedProbes: this.missedProbeLimit, + intervalMs: this.intervalMs, + lastRttMs: this.lastRttMs.get(socket) ?? null, + socketUptimeMs: trackedAt === undefined ? null : reapedAt - trackedAt, + trackedSockets + } + }).end() + } } diff --git a/src/shared/remote-runtime-socket-liveness.ts b/src/shared/remote-runtime-socket-liveness.ts index 63a439304a4..2e996106809 100644 --- a/src/shared/remote-runtime-socket-liveness.ts +++ b/src/shared/remote-runtime-socket-liveness.ts @@ -25,8 +25,13 @@ export const REMOTE_RUNTIME_SOCKET_RESUME_PROBE_DEADLINE_MS = 8_000 // Why: inbound data this recent suggests the path survived; the regular keepalive still covers a miss. export const REMOTE_RUNTIME_SOCKET_RESUME_PROBE_RECENT_ACTIVITY_MS = 3_000 +/** Why the monitor gave up: the regular keepalive window, or one a wake/online probe started. */ +export type RemoteRuntimeSocketDeathCause = 'keepalive_timeout' | 'wake_probe_timeout' + export type RemoteRuntimeSocketLivenessMonitor = { noteActivity: () => void + /** A pong arrived; also measures round-trip time against the oldest unanswered ping. */ + notePong: () => void /** Ping now; declare the socket dead unless anything arrives within the liveness window. */ probeNow: (deadlineMs?: number) => void stop: () => void @@ -48,7 +53,8 @@ export function probeAllRemoteRuntimeSocketsNow( export function startRemoteRuntimeSocketLiveness(args: { /** Returns false when nothing was sent, e.g. the socket is still connecting. */ ping: () => boolean - onDead: () => void + onDead: (cause: RemoteRuntimeSocketDeathCause) => void + onRoundTrip?: (rttMs: number) => void options?: RemoteRuntimeSocketLivenessOptions now?: () => number }): RemoteRuntimeSocketLivenessMonitor { @@ -60,6 +66,7 @@ export function startRemoteRuntimeSocketLiveness(args: { let probeSentAt: number | null = null let activitySinceResumeProbe = false let lastActivityAt: number | null = null + let unansweredPingAt: number | null = null let resumeProbeTimer: ReturnType | null = null let stopped = false @@ -73,12 +80,14 @@ export function startRemoteRuntimeSocketLiveness(args: { // Why: sleep and background throttling age sockets without giving them a chance to answer. if (tickElapsedMs < 0 || tickElapsedMs > pingIntervalMs * 1.5) { probeSentAt = tickAt + // Why: a pong for a ping sent before the stall would report the stall as round-trip time. + unansweredPingAt = null tryPing() return } if (probeSentAt !== null && tickAt - probeSentAt > livenessTimeoutMs) { stop() - args.onDead() + args.onDead('keepalive_timeout') return } if (probeSentAt === null) { @@ -119,6 +128,8 @@ export function startRemoteRuntimeSocketLiveness(args: { return } activitySinceResumeProbe = false + // Why: a ping left from before the wake or network change would report the outage as RTT. + unansweredPingAt = null // Why: a socket still connecting has no path to probe yet; a deadline with no ping could kill it mid-handshake. if (!tryPing()) { return @@ -131,7 +142,7 @@ export function startRemoteRuntimeSocketLiveness(args: { tryPing() armResumeProbeTimer(now(), deathDelayMs, deadlineMs, () => { stop() - args.onDead() + args.onDead('wake_probe_timeout') }) }) } @@ -169,18 +180,32 @@ export function startRemoteRuntimeSocketLiveness(args: { function tryPing(): boolean { try { - return args.ping() + const sent = args.ping() + if (sent && unansweredPingAt === null) { + unansweredPingAt = now() + } + return sent } catch { // Why: ping() can throw while a socket is mid-teardown; the probe deadline still settles it. return true } } + function noteActivity(): void { + probeSentAt = null + activitySinceResumeProbe = true + lastActivityAt = now() + } + const monitor: RemoteRuntimeSocketLivenessMonitor = { - noteActivity: () => { - probeSentAt = null - activitySinceResumeProbe = true - lastActivityAt = now() + noteActivity, + notePong: () => { + noteActivity() + const rttMs = unansweredPingAt === null ? null : now() - unansweredPingAt + unansweredPingAt = null + if (rttMs !== null && rttMs >= 0) { + args.onRoundTrip?.(rttMs) + } }, probeNow, stop diff --git a/src/shared/remote-runtime-socket-resume-probe.test.ts b/src/shared/remote-runtime-socket-resume-probe.test.ts index 0aa2c43b8f2..46ca93ff7cb 100644 --- a/src/shared/remote-runtime-socket-resume-probe.test.ts +++ b/src/shared/remote-runtime-socket-resume-probe.test.ts @@ -46,7 +46,31 @@ describe('remote runtime resume probe', () => { await vi.advanceTimersByTimeAsync(24_999) expect(onDead).not.toHaveBeenCalled() await vi.advanceTimersByTimeAsync(1) - expect(onDead).toHaveBeenCalledTimes(1) + expect(onDead).toHaveBeenCalledExactlyOnceWith('wake_probe_timeout') + }) + + it('measures round trips from the oldest unanswered ping', async () => { + const onRoundTrip = vi.fn() + const monitor = startRemoteRuntimeSocketLiveness({ + ping: () => true, + onDead: vi.fn(), + onRoundTrip + }) + monitors.push(monitor) + + await vi.advanceTimersByTimeAsync(10_000) + await vi.advanceTimersByTimeAsync(350) + monitor.notePong() + monitor.notePong() + expect(onRoundTrip).toHaveBeenCalledExactlyOnceWith(350) + + // A ping left unanswered through a network change is not timed against the probe's pong. + await vi.advanceTimersByTimeAsync(10_000) + await vi.advanceTimersByTimeAsync(5_000) + probeAllRemoteRuntimeSocketsNow() + await vi.advanceTimersByTimeAsync(200) + monitor.notePong() + expect(onRoundTrip).toHaveBeenLastCalledWith(200) }) it('gives a network-change probe a full window even with an older keepalive unanswered', async () => { diff --git a/src/shared/remote-runtime/control/remote-runtime-shared-control-close-cause.test.ts b/src/shared/remote-runtime/control/remote-runtime-shared-control-close-cause.test.ts new file mode 100644 index 00000000000..280cd5167bb --- /dev/null +++ b/src/shared/remote-runtime/control/remote-runtime-shared-control-close-cause.test.ts @@ -0,0 +1,72 @@ +import { afterEach, expect, it, vi } from 'vitest' +import { RemoteRuntimeSharedControlConnection } from './remote-runtime-shared-control-connection' +import type { SharedControlSocketCloseRecord } from './remote-runtime-shared-control-types' +import { + closeSharedControlTestServers, + createSharedControlTestServer as createServer +} from './remote-runtime-shared-control-test-server' + +afterEach(closeSharedControlTestServers) + +it('reports a silent socket as a keepalive timeout, with its uptime', async () => { + const server = await createServer({ disableAutoPong: true }) + const closes: SharedControlSocketCloseRecord[] = [] + const connection = new RemoteRuntimeSharedControlConnection(server.pairing, { + liveness: { pingIntervalMs: 50, livenessTimeoutMs: 200 }, + onSocketClosed: (record) => closes.push(record) + }) + + await connection.request('worktree.ps', undefined, 1000) + await vi.waitFor(() => expect(closes.length).toBeGreaterThan(0), { timeout: 5000 }) + + expect(closes[0]).toMatchObject({ + cause: 'keepalive_timeout', + closeCode: null, + state: 'ready', + lastRttMs: null + }) + expect(closes[0]?.socketUptimeMs).toBeGreaterThanOrEqual(200) + expect(closes[0]?.readyUptimeMs).toBeGreaterThanOrEqual(0) + connection.close() +}) + +it('measures keepalive round trips and reports a server close with its code', async () => { + const server = await createServer() + const closes: SharedControlSocketCloseRecord[] = [] + const publishedCauses: unknown[] = [] + const connection: RemoteRuntimeSharedControlConnection = new RemoteRuntimeSharedControlConnection( + server.pairing, + { + liveness: { pingIntervalMs: 50, livenessTimeoutMs: 1000 }, + onSocketClosed: (record) => closes.push(record), + onDiagnosticsChanged: (diagnostics) => publishedCauses.push(diagnostics.lastCloseCause) + } + ) + + await connection.request('worktree.ps', undefined, 1000) + await vi.waitFor(() => expect(connection.getDiagnostics().lastRttMs).toBeGreaterThanOrEqual(0)) + server.closeClients() + await vi.waitFor(() => expect(closes.length).toBeGreaterThan(0)) + + expect(closes[0]).toMatchObject({ cause: 'socket_close', closeCode: 4001 }) + expect(closes[0]?.lastRttMs).toBeGreaterThanOrEqual(0) + expect(publishedCauses).toContain('socket_close') + connection.close() +}) + +it('reports an unexpected binary frame as a protocol error, not a transport failure', async () => { + const server = await createServer({ sendBinaryAfterAuth: true }) + const closes: SharedControlSocketCloseRecord[] = [] + const connection = new RemoteRuntimeSharedControlConnection(server.pairing, { + onSocketClosed: (record) => closes.push(record) + }) + + await connection.request('worktree.ps', undefined, 1000).catch(() => undefined) + await vi.waitFor(() => expect(closes.length).toBeGreaterThan(0)) + + expect(closes[0]).toMatchObject({ + cause: 'protocol_error', + errorCode: 'invalid_runtime_response' + }) + connection.close() +}) diff --git a/src/shared/remote-runtime/control/remote-runtime-shared-control-connection.ts b/src/shared/remote-runtime/control/remote-runtime-shared-control-connection.ts index 6efdac0de9c..f48d0098a4d 100644 --- a/src/shared/remote-runtime/control/remote-runtime-shared-control-connection.ts +++ b/src/shared/remote-runtime/control/remote-runtime-shared-control-connection.ts @@ -1,7 +1,6 @@ import type { PairedE2EEClientHandshake } from '../../paired-e2ee-client-handshake' import type WebSocket from 'ws' import type { PairedRuntimeConnection } from '../../paired-runtime-connection' -import type { RemoteRuntimeClientError } from '../../remote-runtime-client-error' import { remoteRuntimeClientCapabilities } from '../../remote-runtime-client-capabilities' import { remoteRuntimeUnavailableError } from '../requests/remote-runtime-request-frames' import { openSharedControlSocket } from './remote-runtime-shared-control-open' @@ -13,7 +12,7 @@ import { requestSharedControl } from '../requests/remote-runtime-shared-control- import { SharedControlRetiredRequestIds } from '../requests/remote-runtime-shared-control-retired-request-ids' import { SharedControlReadyStableResetTimer } from './remote-runtime-shared-control-stability' import * as sharedControlState from './remote-runtime-shared-control-state' -import { closeSharedControlSocket } from './remote-runtime-shared-control-socket-close' +import { closeSharedControlSocket, LOCAL_CLOSE } from './remote-runtime-shared-control-socket-close' import { startSharedControlSubscription } from '../requests/remote-runtime-shared-control-subscription-start' import { SharedControlSocketGeneration } from './remote-runtime-shared-control-socket-generation' import { refreshRemoteRuntimeSharedControl } from './remote-runtime-shared-control-refresh' @@ -22,6 +21,7 @@ import { ensureSharedControlReady } from './remote-runtime-shared-control-ready- import { handleRuntimeControlTextFrame } from './remote-runtime-shared-control-connection-frame' import * as sharedControlActions from './remote-runtime-shared-control-connection-actions' import type * as SharedControlTypes from './remote-runtime-shared-control-types' +import type { SharedControlSocketClosed } from './remote-runtime-shared-control-types' import { sendSharedControlSubscription } from './remote-runtime-shared-control-send' import { closeSharedControlConnectionSubscription } from '../requests/remote-runtime-shared-control-subscription-close' type PendingRequest = SharedControlTypes.SharedControlPendingRequest @@ -48,7 +48,7 @@ export class RemoteRuntimeSharedControlConnection { private readonly connection: PairedRuntimeConnection, private readonly options: SharedControlTypes.RemoteRuntimeSharedControlConnectionOptions = {} ) { - this.diagnostics = new SharedControlDiagnosticsTracker(options) + this.diagnostics = new SharedControlDiagnosticsTracker(options, () => this.diagnosticsCounts()) this.readyStableReset = new SharedControlReadyStableResetTimer( options.reconnectStableResetMs ?? 30_000 ) @@ -126,14 +126,14 @@ export class RemoteRuntimeSharedControlConnection { } private publishDiagnostics(): void { - this.diagnostics.publish(this.diagnosticsCounts()) + this.diagnostics.publish() } isReady(): boolean { const { state, ws, handshake } = this return sharedControlReady.isSharedControlReady({ state, ws, handshake }) } getDiagnostics(): SharedControlTypes.RemoteRuntimeSharedConnectionDiagnostics { - return this.diagnostics.get(this.diagnosticsCounts()) + return this.diagnostics.get() } private diagnosticsCounts() { return { @@ -148,7 +148,7 @@ export class RemoteRuntimeSharedControlConnection { refreshRemoteRuntimeSharedControl({ intentionallyClosed: this.intentionallyClosed, ready: this.isReady(), - close: (error) => this.closeSocket(error, true), + close: (error) => this.closeSocket(error, { preserveReadyWaitersAndPendingRequests: true }), open: () => this.open() }) } @@ -176,24 +176,24 @@ export class RemoteRuntimeSharedControlConnection { const socketGeneration = this.socketGeneration.begin() const opened = openSharedControlSocket(this.connection, { getCurrentSocket: () => this.ws, - onClose: (close, error) => { - if (this.socketGeneration.isCurrent(socketGeneration)) { + onSocketClosed: (error, cause, close) => { + if (close && this.socketGeneration.isCurrent(socketGeneration)) { this.diagnostics.markClose(close) } - this.handleSocketClosed(error, socketGeneration) + this.handleSocketClosed(error, socketGeneration, cause) }, - onError: (error) => this.handleSocketClosed(error, socketGeneration), onTextFrame: (frame) => this.handleTextFrame(frame, socketGeneration), onActivity: this.options.onAnswered, liveness: { options: this.options.liveness, - onDead: (error) => this.handleSocketClosed(error, socketGeneration) + onRoundTrip: (rttMs) => this.diagnostics.markRoundTrip(rttMs) } }) if (!opened.ok) { - this.handleSocketClosed(opened.error, socketGeneration) + this.handleSocketClosed(opened.error, socketGeneration, 'socket_error') return } + this.diagnostics.markSocketOpened() this.ws = opened.socket.ws this.handshake = opened.socket.handshake this.socketCleanup = opened.socket.cleanup @@ -217,7 +217,7 @@ export class RemoteRuntimeSharedControlConnection { setState: (state) => { this.state = state }, - handleSocketClosed: (error) => this.handleSocketClosed(error, socketGeneration), + handleSocketClosed: (e) => this.handleSocketClosed(e, socketGeneration, 'protocol_error'), sendEncrypted: (payload) => this.sendEncrypted(payload), clientCapabilities: remoteRuntimeClientCapabilities(this.options.clientCapabilities), getSocket: () => this.ws, @@ -265,15 +265,16 @@ export class RemoteRuntimeSharedControlConnection { }) } - private handleSocketClosed(refused: RemoteRuntimeClientError, socketGeneration: number): void { + private readonly handleSocketClosed: SharedControlSocketClosed = (refused, generation, cause) => { closeSharedControlAfterError({ refused, - socketGeneration, + socketGeneration: generation, generation: this.socketGeneration, onAuthenticationRejected: this.options.onAuthenticationRejected, everReady: this.everReady, subscriptions: this.subscriptions, - closeSocket: (error) => this.closeSocket(error), + closeSocket: (error) => + this.closeSocket(error, { close: { cause, errorCode: refused.code } }), markError: (message) => this.diagnostics.markError(message), isClosed: () => this.state === 'closed', retry: () => @@ -288,8 +289,9 @@ export class RemoteRuntimeSharedControlConnection { }) } - private closeSocket(error?: Error, preserveReadyWaitersAndPendingRequests = false): void { + private closeSocket(error?: Error, options: SharedControlTypes.CloseSocketOptions = {}): void { closeSharedControlSocket({ + reportClose: () => this.diagnostics.markSocketClosed(options.close ?? LOCAL_CLOSE), environmentId: this.options.environmentId, state: this.state, pendingRequests: this.pendingRequests, @@ -299,7 +301,7 @@ export class RemoteRuntimeSharedControlConnection { socketCleanup: this.socketCleanup, ws: this.ws, error, - preserveReadyWaitersAndPendingRequests, + preserveReadyWaitersAndPendingRequests: options.preserveReadyWaitersAndPendingRequests, clearReadyStableTimer: () => this.readyStableReset.clear() }) this.ws = this.handshake = null diff --git a/src/shared/remote-runtime/control/remote-runtime-shared-control-diagnostics.ts b/src/shared/remote-runtime/control/remote-runtime-shared-control-diagnostics.ts index 124a99260ea..9cbde9a367a 100644 --- a/src/shared/remote-runtime/control/remote-runtime-shared-control-diagnostics.ts +++ b/src/shared/remote-runtime/control/remote-runtime-shared-control-diagnostics.ts @@ -1,18 +1,35 @@ import type { RemoteRuntimeSharedConnectionDiagnostics, RemoteRuntimeSharedControlConnectionOptions, - SharedControlConnectionState + SharedControlConnectionState, + SharedControlSocketClose, + SharedControlSocketCloseCause, + SharedControlSocketCloseRecord } from './remote-runtime-shared-control-types' type DiagnosticClose = { code: number; reason: string } | null +type SharedControlDiagnosticsCounts = { + state: SharedControlConnectionState + reconnecting: boolean + pendingRequestCount: number + subscriptionCount: number + reconnectAttempt: number +} + export class SharedControlDiagnosticsTracker { private lastConnectedAt: number | null = null private lastClose: DiagnosticClose = null private lastError: string | null = null + private lastCloseCause: SharedControlSocketCloseCause | null = null + private lastRttMs: number | null = null + private socketOpenedAt: number | null = null private lastPublished: RemoteRuntimeSharedConnectionDiagnostics | null = null - constructor(private readonly options: RemoteRuntimeSharedControlConnectionOptions) {} + constructor( + private readonly options: RemoteRuntimeSharedControlConnectionOptions, + private readonly counts: () => SharedControlDiagnosticsCounts + ) {} markClose(close: DiagnosticClose): void { this.lastClose = close @@ -22,19 +39,52 @@ export class SharedControlDiagnosticsTracker { this.lastConnectedAt = Date.now() this.lastError = null this.lastClose = null + this.lastCloseCause = null + } + + markSocketOpened(): void { + this.socketOpenedAt = Date.now() + // Why: an RTT belongs to the socket it was measured on, not its replacement. + this.lastRttMs = null + } + + markRoundTrip(rttMs: number): void { + this.lastRttMs = rttMs + } + + /** Records why the current socket closed and reports it; never throws into the close path. */ + markSocketClosed(close: SharedControlSocketClose): void { + const counts = this.counts() + const closedAt = Date.now() + const record: SharedControlSocketCloseRecord = { + ...close, + closeCode: close.cause === 'socket_close' ? (this.lastClose?.code ?? null) : null, + state: counts.state, + lastRttMs: this.lastRttMs, + socketUptimeMs: this.socketOpenedAt === null ? null : closedAt - this.socketOpenedAt, + readyUptimeMs: + counts.state === 'ready' && this.lastConnectedAt !== null + ? closedAt - this.lastConnectedAt + : null, + reconnectAttempt: counts.reconnectAttempt, + pendingRequestCount: counts.pendingRequestCount, + subscriptionCount: counts.subscriptionCount + } + this.lastCloseCause = close.cause + this.socketOpenedAt = null + try { + this.options.onSocketClosed?.(record) + } catch (error) { + console.warn('[remote-runtime.shared-control] socket close callback failed:', error) + } } markError(error: string): void { this.lastError = error } - get(args: { - state: SharedControlConnectionState - reconnecting: boolean - pendingRequestCount: number - subscriptionCount: number - reconnectAttempt: number - }): RemoteRuntimeSharedConnectionDiagnostics { + get(): RemoteRuntimeSharedConnectionDiagnostics { + const args = this.counts() return { state: args.reconnecting ? 'reconnecting' : args.state, pendingRequestCount: args.pendingRequestCount, @@ -42,12 +92,14 @@ export class SharedControlDiagnosticsTracker { reconnectAttempt: args.reconnectAttempt, lastConnectedAt: this.lastConnectedAt, lastClose: this.lastClose, - lastError: this.lastError + lastError: this.lastError, + ...(this.lastCloseCause ? { lastCloseCause: this.lastCloseCause } : {}), + ...(this.lastRttMs === null ? {} : { lastRttMs: this.lastRttMs }) } } - publish(args: Parameters[0]): void { - const diagnostics = this.get(args) + publish(): void { + const diagnostics = this.get() const previous = this.lastPublished const closeUnchanged = previous?.lastClose?.code === diagnostics.lastClose?.code && @@ -60,6 +112,7 @@ export class SharedControlDiagnosticsTracker { previous.reconnectAttempt === diagnostics.reconnectAttempt && previous.lastConnectedAt === diagnostics.lastConnectedAt && closeUnchanged && + previous.lastCloseCause === diagnostics.lastCloseCause && previous.lastError === diagnostics.lastError ) { return diff --git a/src/shared/remote-runtime/control/remote-runtime-shared-control-open.ts b/src/shared/remote-runtime/control/remote-runtime-shared-control-open.ts index c6cf7bbdd89..ed9a703c346 100644 --- a/src/shared/remote-runtime/control/remote-runtime-shared-control-open.ts +++ b/src/shared/remote-runtime/control/remote-runtime-shared-control-open.ts @@ -11,13 +11,18 @@ import { startRemoteRuntimeSocketLiveness, type RemoteRuntimeSocketLivenessOptions } from '../../remote-runtime-socket-liveness' +import type { SharedControlSocketCloseCause } from './remote-runtime-shared-control-types' export function openSharedControlSocket( connection: PairedRuntimeConnection, callbacks: { getCurrentSocket: () => WebSocket | null - onClose: (close: { code: number; reason: string }, error: RemoteRuntimeClientError) => void - onError: (error: RemoteRuntimeClientError) => void + /** The current socket closed, errored or went silent; `close` is set for a close frame only. */ + onSocketClosed: ( + error: RemoteRuntimeClientError, + cause: SharedControlSocketCloseCause, + close?: { code: number; reason: string } + ) => void onTextFrame: (frame: string) => void /** Any inbound frame, ping or pong on the current socket. */ onActivity?: () => void @@ -26,11 +31,12 @@ export function openSharedControlSocket( // the server goes silent, so the close/reconnect path can run (#7718). liveness?: { options?: RemoteRuntimeSocketLivenessOptions - onDead: (error: RemoteRuntimeClientError) => void + onRoundTrip?: (rttMs: number) => void } } ): { ok: true; socket: RemoteRuntimeWebSocket } | { ok: false; error: RemoteRuntimeClientError } { let noteLiveness: () => void = () => {} + let notePong: () => void = () => {} const noteActivity = (): void => { noteLiveness() callbacks.onActivity?.() @@ -38,15 +44,17 @@ export function openSharedControlSocket( const opened = openRemoteRuntimeWebSocket(connection, { onClose: (ws, code, reason) => { if (callbacks.getCurrentSocket() === ws) { - callbacks.onClose( - { code, reason: reason.toString() }, - remoteRuntimeUnavailableError(formatSharedControlCloseMessage(code, reason)) + callbacks.onSocketClosed( + remoteRuntimeUnavailableError(formatSharedControlCloseMessage(code, reason)), + 'socket_close', + { code, reason: reason.toString() } ) } }, onError: (ws, error) => { if (callbacks.getCurrentSocket() === ws) { - callbacks.onError(error) + const cause = error.code === 'invalid_runtime_response' ? 'protocol_error' : 'socket_error' + callbacks.onSocketClosed(error, cause) } }, onTextFrame: (ws, frame) => { @@ -57,7 +65,8 @@ export function openSharedControlSocket( }, onPong: (ws) => { if (callbacks.getCurrentSocket() === ws) { - noteActivity() + notePong() + callbacks.onActivity?.() } }, onPing: (ws) => { @@ -80,7 +89,7 @@ export function openSharedControlSocket( ws.ping() return true }, - onDead: () => { + onDead: (cause) => { if (callbacks.getCurrentSocket() !== ws) { return } @@ -90,15 +99,18 @@ export function openSharedControlSocket( } catch { // Best-effort terminate; the dead callback resets connection state. } - liveness.onDead( + callbacks.onSocketClosed( remoteRuntimeUnavailableError( 'Remote Orca runtime stopped responding; resetting the control connection.' - ) + ), + cause ) }, + onRoundTrip: liveness.onRoundTrip, options: liveness.options }) noteLiveness = monitor.noteActivity + notePong = monitor.notePong return { ok: true, diff --git a/src/shared/remote-runtime/control/remote-runtime-shared-control-socket-close.ts b/src/shared/remote-runtime/control/remote-runtime-shared-control-socket-close.ts index aa1df795cb1..dca0904eb07 100644 --- a/src/shared/remote-runtime/control/remote-runtime-shared-control-socket-close.ts +++ b/src/shared/remote-runtime/control/remote-runtime-shared-control-socket-close.ts @@ -6,10 +6,15 @@ import type { SharedControlConnectionState, SharedControlLogicalSubscription, SharedControlPendingRequest, - SharedControlReadyWaiter + SharedControlReadyWaiter, + SharedControlSocketClose } from './remote-runtime-shared-control-types' +export const LOCAL_CLOSE: SharedControlSocketClose = { cause: 'local', errorCode: null } + export function closeSharedControlSocket(args: { + /** Records the close cause; runs only when a socket existed, like the log below. */ + reportClose: () => void environmentId?: string state: SharedControlConnectionState ws: WebSocket | null @@ -23,6 +28,7 @@ export function closeSharedControlSocket(args: { clearReadyStableTimer: () => void }): void { if (args.ws || args.socketCleanup) { + args.reportClose() logSharedControlSocketClose({ environmentId: args.environmentId ?? 'unknown', state: args.state, diff --git a/src/shared/remote-runtime/control/remote-runtime-shared-control-standing-intent.test.ts b/src/shared/remote-runtime/control/remote-runtime-shared-control-standing-intent.test.ts index 1eab8b7371b..fdbbebb81c0 100644 --- a/src/shared/remote-runtime/control/remote-runtime-shared-control-standing-intent.test.ts +++ b/src/shared/remote-runtime/control/remote-runtime-shared-control-standing-intent.test.ts @@ -200,7 +200,8 @@ function invokeSocketClose( ): void { unsafeConnection(connection).handleSocketClosed( new RemoteRuntimeClientError('runtime_unavailable', 'test close'), - generation + generation, + 'socket_close' ) } @@ -219,7 +220,11 @@ function replaceOpen(connection: RemoteRuntimeSharedControlConnection): ReturnTy function unsafeConnection(connection: RemoteRuntimeSharedControlConnection): { open: () => void - handleSocketClosed: (error: RemoteRuntimeClientError, generation: number) => void + handleSocketClosed: ( + error: RemoteRuntimeClientError, + generation: number, + cause: 'socket_close' + ) => void socketGeneration: { begin: () => number } subscriptions: Map reconnect: { attempt: number } diff --git a/src/shared/remote-runtime/control/remote-runtime-shared-control-types.ts b/src/shared/remote-runtime/control/remote-runtime-shared-control-types.ts index f842bc3cf52..487af8f6754 100644 --- a/src/shared/remote-runtime/control/remote-runtime-shared-control-types.ts +++ b/src/shared/remote-runtime/control/remote-runtime-shared-control-types.ts @@ -2,7 +2,10 @@ import type { RuntimeRpcResponse } from '../../runtime-rpc-envelope' import type { RemoteRuntimeClientError } from '../../remote-runtime-client-error' import type { RemoteRuntimePreparedRequest } from '../requests/remote-runtime-prepared-request-admission' import type { RuntimeCapability } from '../../protocol-version' -import type { RemoteRuntimeSocketLivenessOptions } from '../../remote-runtime-socket-liveness' +import type { + RemoteRuntimeSocketDeathCause, + RemoteRuntimeSocketLivenessOptions +} from '../../remote-runtime-socket-liveness' export type SharedControlConnectionState = | 'closed' @@ -71,6 +74,45 @@ export type RemoteRuntimeSharedConnectionDiagnostics = { lastConnectedAt: number | null lastClose: { code: number; reason: string } | null lastError: string | null + /** Why the last socket closed; cleared when a socket becomes ready, like lastClose. */ + lastCloseCause?: SharedControlSocketCloseCause + /** Last keepalive ping round trip; absent until one is measured. Not a publish trigger. */ + lastRttMs?: number +} + +export type SharedControlSocketCloseCause = + | RemoteRuntimeSocketDeathCause + | 'socket_close' + | 'socket_error' + | 'protocol_error' + | 'local' + +export type SharedControlSocketClosed = ( + error: RemoteRuntimeClientError, + socketGeneration: number, + cause: SharedControlSocketCloseCause +) => void + +export type SharedControlSocketClose = { + cause: SharedControlSocketCloseCause + errorCode: string | null +} + +export type CloseSocketOptions = { + preserveReadyWaitersAndPendingRequests?: boolean + close?: SharedControlSocketClose +} + +/** One closed control socket, for diagnostics; carries no hostnames, tokens or payloads. */ +export type SharedControlSocketCloseRecord = SharedControlSocketClose & { + closeCode: number | null + state: SharedControlConnectionState + lastRttMs: number | null + socketUptimeMs: number | null + readyUptimeMs: number | null + reconnectAttempt: number + pendingRequestCount: number + subscriptionCount: number } export type RemoteRuntimeSharedControlConnectionOptions = { environmentId?: string @@ -82,6 +124,8 @@ export type RemoteRuntimeSharedControlConnectionOptions = { /** Any inbound frame, ping or pong on the current socket. */ onAnswered?: () => void reconnectStableResetMs?: number + /** Every socket close with its cause, e.g. for the diagnostics trace. */ + onSocketClosed?: (record: SharedControlSocketCloseRecord) => void liveness?: RemoteRuntimeSocketLivenessOptions /** * The server refused this socket's token. Null leaves the usual reconnect; a promise means the