mirror of
https://github.com/stablyai/orca.git
synced 2026-10-27 00:02:48 +00:00
fix(mobile): close remaining stalled RPC paths
This commit is contained in:
@@ -269,7 +269,8 @@ describe('useHostClient', () => {
|
||||
|
||||
expect(stale.closeMock).toHaveBeenCalledOnce()
|
||||
expect(fresh.sendRequest).toHaveBeenCalledWith('status.get', undefined, {
|
||||
timeoutMs: 15_000
|
||||
timeoutMs: 15_000,
|
||||
budgetSpansConnect: true
|
||||
})
|
||||
expect(completed).toBe(false)
|
||||
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import type { RpcClient, SendRequestOptions } from './rpc-client'
|
||||
import { verifyForceReconnectRpcHealth } from './force-reconnect-rpc-health'
|
||||
import { LogicalClientCutoverError } from './stable-logical-rpc-client'
|
||||
|
||||
describe('Force Reconnect RPC health', () => {
|
||||
afterEach(() => vi.useRealTimers())
|
||||
|
||||
it('spends one 15-second budget across a cutover and the replacement probe', async () => {
|
||||
vi.useFakeTimers()
|
||||
let attempt = 0
|
||||
const timeouts: number[] = []
|
||||
const sendRequest = vi.fn(
|
||||
(_method: string, _params?: unknown, options?: SendRequestOptions): Promise<never> => {
|
||||
const timeoutMs = options?.timeoutMs ?? 0
|
||||
timeouts.push(timeoutMs)
|
||||
const currentAttempt = ++attempt
|
||||
return new Promise((_, reject) => {
|
||||
setTimeout(
|
||||
() =>
|
||||
reject(currentAttempt === 1 ? new LogicalClientCutoverError() : new Error('stalled')),
|
||||
currentAttempt === 1 ? 14_000 : timeoutMs
|
||||
)
|
||||
})
|
||||
}
|
||||
)
|
||||
const client = { sendRequest } as unknown as RpcClient
|
||||
let outcome = 'pending'
|
||||
const verification = verifyForceReconnectRpcHealth(client).catch((error: Error) => {
|
||||
outcome = error.message
|
||||
})
|
||||
|
||||
await vi.advanceTimersByTimeAsync(14_999)
|
||||
expect(outcome).toBe('pending')
|
||||
expect(timeouts).toEqual([15_000, 1_000])
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
await verification
|
||||
|
||||
expect(outcome).toBe('stalled')
|
||||
expect(sendRequest).toHaveBeenCalledTimes(2)
|
||||
expect(sendRequest).toHaveBeenLastCalledWith('status.get', undefined, {
|
||||
timeoutMs: 1_000,
|
||||
budgetSpansConnect: true
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -1,9 +1,24 @@
|
||||
import type { RpcClient } from './rpc-client'
|
||||
import { isLogicalClientCutoverError } from './stable-logical-rpc-client'
|
||||
|
||||
const FORCE_RECONNECT_HEALTH_TIMEOUT_MS = 15_000
|
||||
|
||||
export async function verifyForceReconnectRpcHealth(client: RpcClient): Promise<void> {
|
||||
await client
|
||||
.sendRequest('status.get', undefined, { timeoutMs: FORCE_RECONNECT_HEALTH_TIMEOUT_MS })
|
||||
.catch(() => undefined)
|
||||
const deadline = Date.now() + FORCE_RECONNECT_HEALTH_TIMEOUT_MS
|
||||
let lastCutover: unknown = null
|
||||
for (;;) {
|
||||
const timeoutMs = deadline - Date.now()
|
||||
if (timeoutMs <= 0) {
|
||||
throw lastCutover ?? new Error('Force Reconnect health check timed out')
|
||||
}
|
||||
try {
|
||||
await client.sendRequest('status.get', undefined, { timeoutMs, budgetSpansConnect: true })
|
||||
return
|
||||
} catch (error) {
|
||||
if (!isLogicalClientCutoverError(error)) {
|
||||
throw error
|
||||
}
|
||||
lastCutover = error
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -234,6 +234,11 @@ describe('mobile relay RPC session', () => {
|
||||
message: 'relay RPC timed out: terminal.send',
|
||||
unknown: true
|
||||
})
|
||||
expect(session.getFailure()).toMatchObject({
|
||||
message: 'relay RPC timed out: terminal.send'
|
||||
})
|
||||
expect(session.getState()).toBe('disconnected')
|
||||
expect(fakes.close).toHaveBeenCalledOnce()
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
|
||||
@@ -152,7 +152,9 @@ export function connectMobileRelayRpcSession(args: {
|
||||
const timer = setTimeout(() => {
|
||||
pending.delete(id)
|
||||
// Why: the frame was written long ago — the desktop may have processed it.
|
||||
reject(markRpcDeliveryUnknown(new Error(`relay RPC timed out: ${method}`)))
|
||||
const error = markRpcDeliveryUnknown(new Error(`relay RPC timed out: ${method}`))
|
||||
reject(error)
|
||||
fail(error)
|
||||
}, timeoutMs)
|
||||
pending.set(id, { resolve, reject, timer })
|
||||
if (!sendFrame({ id, method, params })) {
|
||||
|
||||
@@ -56,6 +56,14 @@ class MockWebSocket {
|
||||
}
|
||||
}
|
||||
|
||||
function sentRequest(socket: MockWebSocket, method: string): { id: string } {
|
||||
const payload = socket.sent.find((value) => value.includes(`"method":"${method}"`))
|
||||
if (!payload) {
|
||||
throw new Error(`No ${method} request sent`)
|
||||
}
|
||||
return JSON.parse(payload.replace(/^encrypted:/, '')) as { id: string }
|
||||
}
|
||||
|
||||
const mockSockets: MockWebSocket[] = []
|
||||
const originalWebSocket = globalThis.WebSocket
|
||||
|
||||
@@ -171,4 +179,26 @@ describe('mobile rpc-client request deadline', () => {
|
||||
await request.catch(() => undefined)
|
||||
}
|
||||
})
|
||||
|
||||
it('does not reconnect when another control response proves the socket live', async () => {
|
||||
const client = connect('ws://desktop.invalid', 'token', 'server-key')
|
||||
const socket = mockSockets[0]!
|
||||
socket.open()
|
||||
const stalled = client.sendRequest('browser.screenshot', {}, { timeoutMs: 100 })
|
||||
const stalledOutcome = stalled.catch((error: unknown) => error)
|
||||
const healthy = client.sendRequest('status.get', undefined, { timeoutMs: 100 })
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
const healthRequest = sentRequest(socket, 'status.get')
|
||||
socket.receive(`encrypted:${JSON.stringify({ id: healthRequest.id, ok: true, result: {} })}`)
|
||||
|
||||
await expect(healthy).resolves.toMatchObject({ ok: true })
|
||||
await vi.advanceTimersByTimeAsync(100)
|
||||
await expect(stalledOutcome).resolves.toMatchObject({
|
||||
message: 'Request timed out: browser.screenshot'
|
||||
})
|
||||
expect(socket.close).not.toHaveBeenCalled()
|
||||
expect(client.getState()).toBe('connected')
|
||||
|
||||
client.close()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -234,7 +234,7 @@ describe('mobile rpc-client connection timeout', () => {
|
||||
client.close()
|
||||
})
|
||||
|
||||
it('routes browser screencast binary frames to the browser subscriber', async () => {
|
||||
it('routes browser frames without treating them as control liveness', async () => {
|
||||
const client = connect('ws://desktop.invalid', 'token', 'server-key')
|
||||
const socket = mockSockets[0]!
|
||||
const frames: unknown[] = []
|
||||
@@ -260,16 +260,16 @@ describe('mobile rpc-client connection timeout', () => {
|
||||
})}`
|
||||
)
|
||||
|
||||
socket.emitCloseOnClose = false
|
||||
client.notifyForeground()
|
||||
const probe = sentRequest(socket, 'status.get')
|
||||
socket.receive(`encrypted:${JSON.stringify({ id: probe.id, ok: true, result: {} })}`)
|
||||
socket.receive(encodeBrowserFrame())
|
||||
await Promise.resolve()
|
||||
await Promise.resolve()
|
||||
await vi.advanceTimersByTimeAsync(8_000)
|
||||
|
||||
expect(frames).toHaveLength(1)
|
||||
expect(socket.close).not.toHaveBeenCalled()
|
||||
expect(socket.close).toHaveBeenCalled()
|
||||
expect(client.getState()).toBe('reconnecting')
|
||||
expect(frames[0]).toMatchObject({
|
||||
seq: 7,
|
||||
format: 'jpeg',
|
||||
|
||||
@@ -1024,19 +1024,18 @@ export function connect(
|
||||
}
|
||||
|
||||
const requestWs = ws
|
||||
const requestControlResponseSequence = controlResponseSequence
|
||||
return new Promise((resolve, reject) => {
|
||||
const id = nextId()
|
||||
const timeoutMs = resolvePostConnectRequestTimeout(budget, REQUEST_TIMEOUT_MS)
|
||||
const timeout = setTimeout(() => {
|
||||
pending.delete(id)
|
||||
console.log('[net] sendRequest TIMEOUT', {
|
||||
method,
|
||||
timeoutMs,
|
||||
state
|
||||
})
|
||||
console.log('[net] sendRequest TIMEOUT', { method, timeoutMs, state })
|
||||
// Why: the frame was written 30s ago — the host may have processed it.
|
||||
reject(markRpcDeliveryUnknown(new Error(`Request timed out: ${method}`)))
|
||||
forceSocketReconnect(requestWs)
|
||||
if (controlResponseSequence === requestControlResponseSequence) {
|
||||
forceSocketReconnect(requestWs)
|
||||
}
|
||||
}, timeoutMs)
|
||||
|
||||
pending.set(id, {
|
||||
|
||||
@@ -6,6 +6,7 @@ import {
|
||||
createStableLogicalRpcClient,
|
||||
LogicalClientCutoverError
|
||||
} from './stable-logical-rpc-client'
|
||||
import { verifyForceReconnectRpcHealth } from './force-reconnect-rpc-health'
|
||||
|
||||
class FakeSession implements RpcClient {
|
||||
readonly sendRequest =
|
||||
@@ -100,6 +101,26 @@ describe('stable logical RPC client', () => {
|
||||
pending.resolve(success('late'))
|
||||
})
|
||||
|
||||
it('retries Force Reconnect health verification on the active session after cutover', async () => {
|
||||
const oldSession = new FakeSession('connected')
|
||||
const nextSession = new FakeSession('connected')
|
||||
const pending = deferred<RpcResponse>()
|
||||
oldSession.sendRequest.mockReturnValue(pending.promise)
|
||||
nextSession.sendRequest.mockResolvedValue(success('relay'))
|
||||
const client = createStableLogicalRpcClient(oldSession, 'lan')
|
||||
|
||||
const verifying = verifyForceReconnectRpcHealth(client)
|
||||
await Promise.resolve()
|
||||
await client.migrateTo(nextSession, 'relay')
|
||||
|
||||
await expect(verifying).resolves.toBeUndefined()
|
||||
expect(nextSession.sendRequest).toHaveBeenCalledWith('status.get', undefined, {
|
||||
timeoutMs: expect.any(Number),
|
||||
budgetSpansConnect: true
|
||||
})
|
||||
pending.resolve(success('late'))
|
||||
})
|
||||
|
||||
it('keeps replies that commit before cutover and carries viewport state into replay', async () => {
|
||||
const oldSession = new FakeSession('connected')
|
||||
const nextSession = new FakeSession('connected')
|
||||
|
||||
Reference in New Issue
Block a user