mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
feat(remote): trace why each remote-runtime connection closed (#33094)
Client control-socket closes and server heartbeat reaps only reached the console, which a packaged app never shows. Each control-socket close now writes a remote-runtime.control-socket-close span (cause, close code, error code, last keepalive RTT, socket and ready uptime). Each server reap writes remote-runtime.server-socket-reaped. Both land in the main trace file that Send app diagnostics collects. Keepalive pongs now measure RTT, exposed as lastRttMs and lastCloseCause on the shared-control diagnostics.
This commit is contained in:
+36
@@ -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
|
||||
})
|
||||
})
|
||||
])
|
||||
})
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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<WebSocket>()
|
||||
// Why: consecutive probes a socket has left unanswered. Absent = zero, so live sockets cost nothing.
|
||||
private readonly missedProbes = new WeakMap<WebSocket, number>()
|
||||
private readonly trackedAt = new WeakMap<WebSocket, number>()
|
||||
private readonly unansweredPingAt = new WeakMap<WebSocket, number>()
|
||||
private readonly lastRttMs = new WeakMap<WebSocket, number>()
|
||||
|
||||
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<WebSocket>): 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()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<typeof setTimeout> | 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
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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()
|
||||
})
|
||||
@@ -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<unknown>
|
||||
@@ -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
|
||||
|
||||
@@ -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<SharedControlDiagnosticsTracker['get']>[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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
+7
-2
@@ -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<string, SharedControlLogicalSubscription>
|
||||
reconnect: { attempt: number }
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user