mirror of
https://github.com/stablyai/orca.git
synced 2026-11-13 17:37:40 +00:00
fix(remote-runtime): retire a shared-control socket that opens but never becomes ready (#38690)
The shared-control connection now bounds OPEN to ready with a 30s deadline owned by its readiness timer. When it elapses, the generation closes with a ready_timeout cause and reconnects through the existing backoff, which only resets after 30s of stable readiness, so a slow host cannot cause a storm. Fixes #37330
This commit is contained in:
@@ -70,3 +70,20 @@ it('reports an unexpected binary frame as a protocol error, not a transport fail
|
||||
})
|
||||
connection.close()
|
||||
})
|
||||
|
||||
it('retires a socket that opens but never becomes ready and reconnects with backoff', async () => {
|
||||
const server = await createServer({ suppressReadyFrameCount: 1 })
|
||||
const closes: SharedControlSocketCloseRecord[] = []
|
||||
const connection = new RemoteRuntimeSharedControlConnection(server.pairing, {
|
||||
readyDeadlineMs: 100,
|
||||
onSocketClosed: (record) => closes.push(record)
|
||||
})
|
||||
|
||||
await expect(connection.request('worktree.ps', undefined, 2000)).rejects.toThrow(
|
||||
'did not finish connecting'
|
||||
)
|
||||
expect(closes).toMatchObject([{ cause: 'ready_timeout', state: 'awaiting_ready' }])
|
||||
await vi.waitFor(() => expect(connection.isReady()).toBe(true))
|
||||
expect(server.connectionCount()).toBe(2)
|
||||
connection.close()
|
||||
})
|
||||
|
||||
@@ -10,7 +10,7 @@ import * as sharedControlProtocol from './remote-runtime-shared-control-protocol
|
||||
import { SharedControlReconnectScheduler } from './remote-runtime-shared-control-reconnect'
|
||||
import { requestSharedControl } from '../requests/remote-runtime-shared-control-requests'
|
||||
import { SharedControlRetiredRequestIds } from '../requests/remote-runtime-shared-control-retired-request-ids'
|
||||
import { SharedControlReadyStableResetTimer } from './remote-runtime-shared-control-stability'
|
||||
import { SharedControlReadinessTimer } from './remote-runtime-shared-control-stability'
|
||||
import * as sharedControlState from './remote-runtime-shared-control-state'
|
||||
import { closeSharedControlSocket, LOCAL_CLOSE } from './remote-runtime-shared-control-socket-close'
|
||||
import { startSharedControlSubscription } from '../requests/remote-runtime-shared-control-subscription-start'
|
||||
@@ -33,7 +33,7 @@ export class RemoteRuntimeSharedControlConnection {
|
||||
private handshake: PairedE2EEClientHandshake | null = null
|
||||
private socketCleanup: (() => void) | null = null
|
||||
private readonly reconnect = new SharedControlReconnectScheduler()
|
||||
private readonly readyStableReset: SharedControlReadyStableResetTimer
|
||||
private readonly readiness: SharedControlReadinessTimer
|
||||
private intentionallyClosed = false
|
||||
private readonly diagnostics: SharedControlDiagnosticsTracker
|
||||
private readonly pendingRequests = new Map<string, PendingRequest>()
|
||||
@@ -49,9 +49,12 @@ export class RemoteRuntimeSharedControlConnection {
|
||||
private readonly options: SharedControlTypes.RemoteRuntimeSharedControlConnectionOptions = {}
|
||||
) {
|
||||
this.diagnostics = new SharedControlDiagnosticsTracker(options, () => this.diagnosticsCounts())
|
||||
this.readyStableReset = new SharedControlReadyStableResetTimer(
|
||||
options.reconnectStableResetMs ?? 30_000
|
||||
)
|
||||
this.readiness = new SharedControlReadinessTimer(options, {
|
||||
getState: () => this.state,
|
||||
getSocket: () => this.ws,
|
||||
resetAttempt: () => this.reconnect.resetAttempt(),
|
||||
retire: this.handleSocketClosed
|
||||
})
|
||||
}
|
||||
request<TResult>(
|
||||
method: string,
|
||||
@@ -182,6 +185,7 @@ export class RemoteRuntimeSharedControlConnection {
|
||||
}
|
||||
this.handleSocketClosed(error, socketGeneration, cause)
|
||||
},
|
||||
onOpened: () => this.readiness.armReadyDeadline(socketGeneration),
|
||||
onTextFrame: (frame) => this.handleTextFrame(frame, socketGeneration),
|
||||
onActivity: this.options.onAnswered,
|
||||
liveness: {
|
||||
@@ -226,11 +230,7 @@ export class RemoteRuntimeSharedControlConnection {
|
||||
this.options.onReady?.()
|
||||
// Clear predecessor-attempt diagnostics on recovery so Connected cannot retain stale failure details.
|
||||
this.publishDiagnostics()
|
||||
this.readyStableReset.schedule({
|
||||
getState: () => this.state,
|
||||
getSocket: () => this.ws,
|
||||
reset: () => this.reconnect.resetAttempt()
|
||||
})
|
||||
this.readiness.scheduleStableReset()
|
||||
},
|
||||
replaySubscriptions: () => this.replaySubscriptions(),
|
||||
publishDiagnostics: () => this.publishDiagnostics()
|
||||
@@ -302,7 +302,7 @@ export class RemoteRuntimeSharedControlConnection {
|
||||
ws: this.ws,
|
||||
error,
|
||||
preserveReadyWaitersAndPendingRequests: options.preserveReadyWaitersAndPendingRequests,
|
||||
clearReadyStableTimer: () => this.readyStableReset.clear()
|
||||
clearReadinessTimer: () => this.readiness.clear()
|
||||
})
|
||||
this.ws = this.handshake = null
|
||||
this.socketCleanup = null
|
||||
|
||||
@@ -23,6 +23,8 @@ export function openSharedControlSocket(
|
||||
cause: SharedControlSocketCloseCause,
|
||||
close?: { code: number; reason: string }
|
||||
) => void
|
||||
/** The current socket finished its opening handshake. */
|
||||
onOpened: () => void
|
||||
onTextFrame: (frame: string) => void
|
||||
/** Any inbound frame, ping or pong on the current socket. */
|
||||
onActivity?: () => void
|
||||
@@ -40,6 +42,11 @@ export function openSharedControlSocket(
|
||||
callbacks.onActivity?.()
|
||||
}
|
||||
const opened = openRemoteRuntimeWebSocket(connection, {
|
||||
onOpen: (ws) => {
|
||||
if (callbacks.getCurrentSocket() === ws) {
|
||||
callbacks.onOpened()
|
||||
}
|
||||
},
|
||||
onClose: (ws, code, reason) => {
|
||||
if (callbacks.getCurrentSocket() === ws) {
|
||||
callbacks.onSocketClosed(
|
||||
|
||||
@@ -25,7 +25,7 @@ export function closeSharedControlSocket(args: {
|
||||
lastClose: { code: number; reason: string } | null
|
||||
error?: Error
|
||||
preserveReadyWaitersAndPendingRequests?: boolean
|
||||
clearReadyStableTimer: () => void
|
||||
clearReadinessTimer: () => void
|
||||
}): void {
|
||||
if (args.ws || args.socketCleanup) {
|
||||
args.reportClose()
|
||||
@@ -38,7 +38,7 @@ export function closeSharedControlSocket(args: {
|
||||
error: args.error
|
||||
})
|
||||
}
|
||||
args.clearReadyStableTimer()
|
||||
args.clearReadinessTimer()
|
||||
finishCloseAfterReadySubscriptions(args.subscriptions)
|
||||
closeSharedControlSocketState({
|
||||
readyWaiters: args.readyWaiters,
|
||||
|
||||
@@ -1,5 +1,14 @@
|
||||
import WebSocket from 'ws'
|
||||
import type { SharedControlConnectionState } from './remote-runtime-shared-control-types'
|
||||
import { remoteRuntimeUnavailableError } from '../requests/remote-runtime-request-frames'
|
||||
import type {
|
||||
RemoteRuntimeSharedControlConnectionOptions,
|
||||
SharedControlConnectionState,
|
||||
SharedControlSocketClosed
|
||||
} from './remote-runtime-shared-control-types'
|
||||
|
||||
/** Bounds OPEN→ready: pongs prove transport only, so a socket that never readies must still be retired (#37330).
|
||||
* Well above the host's 10s pre-auth reaper and one 15s caller wait, so a slow older host is not cut off. */
|
||||
const SHARED_CONTROL_READY_DEADLINE_MS = 30_000
|
||||
|
||||
export function scheduleSharedControlStableReset(args: {
|
||||
delayMs: number
|
||||
@@ -21,23 +30,46 @@ export function scheduleSharedControlStableReset(args: {
|
||||
return timer
|
||||
}
|
||||
|
||||
// Replace any prior stable-ready timer when scheduling.
|
||||
export class SharedControlReadyStableResetTimer {
|
||||
/** The current socket's one readiness timer: its OPEN→ready deadline, then its stable-ready backoff reset. */
|
||||
export class SharedControlReadinessTimer {
|
||||
private timer: ReturnType<typeof setTimeout> | null = null
|
||||
|
||||
constructor(private readonly delayMs: number) {}
|
||||
constructor(
|
||||
private readonly options: Pick<
|
||||
RemoteRuntimeSharedControlConnectionOptions,
|
||||
'reconnectStableResetMs' | 'readyDeadlineMs'
|
||||
>,
|
||||
private readonly args: {
|
||||
getState: () => SharedControlConnectionState
|
||||
getSocket: () => WebSocket | null
|
||||
resetAttempt: () => void
|
||||
retire: SharedControlSocketClosed
|
||||
}
|
||||
) {}
|
||||
|
||||
schedule(args: {
|
||||
getState: () => SharedControlConnectionState
|
||||
getSocket: () => WebSocket | null
|
||||
reset: () => void
|
||||
}): void {
|
||||
armReadyDeadline(generation: number): void {
|
||||
this.clear()
|
||||
const timer = setTimeout(() => {
|
||||
this.timer = null
|
||||
this.args.retire(
|
||||
remoteRuntimeUnavailableError(
|
||||
'Remote Orca runtime did not finish connecting; resetting the control connection.'
|
||||
),
|
||||
generation,
|
||||
'ready_timeout'
|
||||
)
|
||||
}, this.options.readyDeadlineMs ?? SHARED_CONTROL_READY_DEADLINE_MS)
|
||||
timer.unref?.()
|
||||
this.timer = timer
|
||||
}
|
||||
|
||||
scheduleStableReset(): void {
|
||||
this.clear()
|
||||
this.timer = scheduleSharedControlStableReset({
|
||||
delayMs: this.delayMs,
|
||||
getState: args.getState,
|
||||
getSocket: args.getSocket,
|
||||
reset: args.reset,
|
||||
delayMs: this.options.reconnectStableResetMs ?? 30_000,
|
||||
getState: this.args.getState,
|
||||
getSocket: this.args.getSocket,
|
||||
reset: this.args.resetAttempt,
|
||||
clearCurrent: () => {
|
||||
this.timer = null
|
||||
}
|
||||
|
||||
@@ -76,6 +76,7 @@ export type SharedControlSocketCloseCause =
|
||||
| 'socket_close'
|
||||
| 'socket_error'
|
||||
| 'protocol_error'
|
||||
| 'ready_timeout'
|
||||
| 'local'
|
||||
|
||||
export type SharedControlSocketClosed = (
|
||||
@@ -117,6 +118,8 @@ export type RemoteRuntimeSharedControlConnectionOptions = {
|
||||
/** Each socket that becomes ready, including after every reconnect. */
|
||||
onReady?: () => void
|
||||
reconnectStableResetMs?: number
|
||||
/** OPEN→ready bound before the socket is retired; defaults to 30s. */
|
||||
readyDeadlineMs?: number
|
||||
/** Every socket close with its cause, e.g. for the diagnostics trace. */
|
||||
onSocketClosed?: (record: SharedControlSocketCloseRecord) => void
|
||||
liveness?: RemoteRuntimeSocketLivenessOptions
|
||||
|
||||
@@ -20,6 +20,8 @@ export type RemoteRuntimeWebSocket = {
|
||||
}
|
||||
|
||||
export type RemoteRuntimeWebSocketCallbacks = {
|
||||
/** The opening handshake completed and the E2EE hello was sent. */
|
||||
onOpen?: (ws: WebSocket) => void
|
||||
onClose: (ws: WebSocket, code: number, reason: Buffer) => void
|
||||
onError: (ws: WebSocket, error: RemoteRuntimeClientError) => void
|
||||
onTextFrame: (ws: WebSocket, frame: string) => void
|
||||
@@ -45,6 +47,7 @@ export function openRemoteRuntimeWebSocket(
|
||||
let cleanedUp = false
|
||||
const onOpen = (): void => {
|
||||
ws.send(handshake.helloText())
|
||||
callbacks.onOpen?.(ws)
|
||||
}
|
||||
const onError = (error: Error): void => {
|
||||
callbacks.onError(
|
||||
|
||||
Reference in New Issue
Block a user