From 693c43a4a6dba78709cafcb804ba3dcc35aeef2f Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Sat, 1 Aug 2026 02:21:13 -0700 Subject: [PATCH] fix(mobile): probe timed-out RPCs before reconnecting --- .../mobile-endpoint-hysteresis-policy.ts | 16 ++ .../transport/mobile-endpoint-lifecycle.ts | 11 +- ...ile-endpoint-supervisor-retirement.test.ts | 222 ++++++++++++++++++ .../transport/mobile-endpoint-supervisor.ts | 21 +- .../mobile-relay-rpc-session.test.ts | 39 ++- .../src/transport/mobile-relay-rpc-session.ts | 7 +- .../rpc-client-request-deadline.test.ts | 42 ++++ mobile/src/transport/rpc-client.test.ts | 31 --- mobile/src/transport/rpc-client.ts | 6 +- 9 files changed, 338 insertions(+), 57 deletions(-) create mode 100644 mobile/src/transport/mobile-endpoint-hysteresis-policy.ts create mode 100644 mobile/src/transport/mobile-endpoint-supervisor-retirement.test.ts diff --git a/mobile/src/transport/mobile-endpoint-hysteresis-policy.ts b/mobile/src/transport/mobile-endpoint-hysteresis-policy.ts new file mode 100644 index 00000000000..037c724c086 --- /dev/null +++ b/mobile/src/transport/mobile-endpoint-hysteresis-policy.ts @@ -0,0 +1,16 @@ +import { MobileEndpointHysteresis } from './mobile-endpoint-hysteresis' + +export type { MobileEndpointHysteresis } from './mobile-endpoint-hysteresis' + +const DIRECT_OBSERVATION_MS = 30_000 +const MINIMUM_DWELL_MS = 60_000 +const FAILURE_COOLDOWN_MS = 60_000 + +export function createMobileEndpointHysteresis(now: number): MobileEndpointHysteresis { + return new MobileEndpointHysteresis(now, { + directSuccessesRequired: 3, + directObservationMs: DIRECT_OBSERVATION_MS, + failureCooldownMs: FAILURE_COOLDOWN_MS, + minimumDwellMs: MINIMUM_DWELL_MS + }) +} diff --git a/mobile/src/transport/mobile-endpoint-lifecycle.ts b/mobile/src/transport/mobile-endpoint-lifecycle.ts index 0a9e90e57df..c8f9852de0e 100644 --- a/mobile/src/transport/mobile-endpoint-lifecycle.ts +++ b/mobile/src/transport/mobile-endpoint-lifecycle.ts @@ -31,12 +31,17 @@ export function startMobileEndpointLifecycle( let stopped = false let foreground = true let owner: EndpointOwner + const publishHostUpdate = (host: HostProfile): void => { + if (!stopped) { + onHostUpdated(host) + } + } const startSupervisor = async (host: HostProfile): Promise => { if (stopped) { return } - const supervisor = createSupervisor(logical, host, onLog, onHostUpdated) + const supervisor = createSupervisor(logical, host, onLog, publishHostUpdate) owner.stop() owner = supervisor supervisor.setForeground(foreground) @@ -44,7 +49,7 @@ export function startMobileEndpointLifecycle( } if (initialHost.relay) { - owner = createSupervisor(logical, initialHost, onLog, onHostUpdated) + owner = createSupervisor(logical, initialHost, onLog, publishHostUpdate) void owner.start() } else { owner = new MobileRelayDirectUpgradeController(logical, initialHost, { @@ -55,7 +60,7 @@ export function startMobileEndpointLifecycle( dependencies: { randomBytes: ExpoCrypto.getRandomBytes } }), onUpgraded: async ({ host }) => { - onHostUpdated(host) + publishHostUpdate(host) await startSupervisor(host) } }) diff --git a/mobile/src/transport/mobile-endpoint-supervisor-retirement.test.ts b/mobile/src/transport/mobile-endpoint-supervisor-retirement.test.ts new file mode 100644 index 00000000000..8a1954980fb --- /dev/null +++ b/mobile/src/transport/mobile-endpoint-supervisor-retirement.test.ts @@ -0,0 +1,222 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { MobileRelayCredentialBundle } from './mobile-relay-credential-bundle' +import { RelayOuterError } from './mobile-relay-e2ee-link' +import type { MobileRelayRpcSession } from './mobile-relay-rpc-session' +import { + MobileEndpointSupervisor, + type MobileEndpointSupervisorDependencies +} from './mobile-endpoint-supervisor' +import type { RpcClient } from './rpc-client' +import type { MobileConnectionPath, StableLogicalRpcClient } from './stable-logical-rpc-client' +import type { ConnectionState, HostProfile, RpcResponse } from './types' + +vi.mock('react-native', () => ({ Platform: { OS: 'ios' } })) +vi.mock('expo-secure-store', () => ({ WHEN_UNLOCKED_THIS_DEVICE_ONLY: 'when-unlocked' })) +vi.mock('expo-crypto', () => ({ getRandomBytes: (length: number) => new Uint8Array(length) })) + +class FakeSession implements RpcClient { + readonly sendRequest = vi.fn( + async (_method: string, _params?: unknown): Promise => ({ + id: 'rpc-1', + ok: true, + result: {}, + _meta: { runtimeId: 'runtime-1' } + }) + ) + readonly subscribe = vi.fn(() => () => {}) + readonly updateTerminalSubscriptionViewport = vi.fn() + readonly notifyForeground = vi.fn() + readonly close = vi.fn() + private readonly listeners = new Set<(state: ConnectionState) => void>() + + constructor(private state: ConnectionState) {} + + getState = () => this.state + getReconnectAttempt = () => 0 + getLastConnectedAt = () => null + onStateChange = (listener: (state: ConnectionState) => void) => { + this.listeners.add(listener) + return () => this.listeners.delete(listener) + } + + publishState(state: ConnectionState): void { + this.state = state + for (const listener of this.listeners) { + listener(state) + } + } +} + +class FakeRelaySession extends FakeSession implements MobileRelayRpcSession { + constructor( + state: ConnectionState, + private readonly failure: Error | null = null + ) { + super(state) + } + getLeaseExpiresAt = () => Date.now() + 120_000 + getResumeConfirmation = () => null + getFailure = () => this.failure +} + +class FakeLogicalClient extends FakeSession implements StableLogicalRpcClient { + private generation = 1 + + constructor( + state: ConnectionState, + private path: MobileConnectionPath + ) { + super(state) + } + + migrateTo = vi.fn(async (session: RpcClient, path: MobileConnectionPath) => { + if (session.getState() !== 'connected') { + session.close() + throw new Error(`replacement session ${session.getState()}`) + } + this.path = path + this.generation += 1 + this.publishState('connected') + }) + suspendActiveSession = vi.fn(() => this.publishState('disconnected')) + getActivePath = () => this.path + getGeneration = () => this.generation +} + +const relay = { + v: 1 as const, + directorUrl: 'https://relay.onorca.dev', + cellUrl: 'https://relay-c1.onorca.dev', + assignmentEpoch: 7, + relayHostId: 'AbCdEf0123_-xyZ9', + e2eeFraming: 2 as const +} +const host: HostProfile = { + id: 'host-1', + name: 'Blue Whale', + endpoint: 'ws://192.168.1.10:6768', + deviceToken: 'device-token', + publicKeyB64: 'A'.repeat(44), + lastConnected: 1, + relayHostId: relay.relayHostId, + relay +} +const bundle: MobileRelayCredentialBundle = { + v: 1, + hostId: host.id, + deviceToken: host.deviceToken, + current: { + token: 'A'.repeat(43), + hash: 'B'.repeat(43), + version: 2, + expiresAt: Number.MAX_SAFE_INTEGER + } +} + +function dependencies( + overrides: Partial = {} +): MobileEndpointSupervisorDependencies { + return { + openDirect: vi.fn(() => new FakeSession('connected')), + openRelay: vi.fn(() => new FakeRelaySession('connected')), + resolveRelay: vi.fn(async ({ relay }) => relay), + readBundle: vi.fn(async () => bundle), + writeBundle: vi.fn(async () => {}), + saveHost: vi.fn(async () => {}), + now: Date.now, + randomBytes: (length) => new Uint8Array(length).fill(1), + setTimer: setTimeout, + clearTimer: clearTimeout, + ...overrides + } +} + +function mockCredentialRotation(logical: FakeLogicalClient): void { + let installed: Record | null = null + logical.sendRequest.mockImplementation(async (method, params) => { + const request = params as { installReqId?: string; reqId?: string } + if (method === 'pairing.provisionRelay') { + installed = { + v: 1, + reqId: request.reqId, + authorizationMode: 'authenticated-direct', + currentVersion: 3, + resumeExpiresAt: Date.now() + 300_000, + graceExpiresAt: Date.now() + 60_000 + } + return { id: 'rpc-2', ok: true, result: installed, _meta: { runtimeId: 'runtime-1' } } + } + return { + id: 'rpc-1', + ok: true, + result: { + v: 1, + relay, + installStatus: installed + ? { v: 1, reqId: request.installReqId, state: 'committed', result: installed } + : { v: 1, reqId: request.installReqId, state: 'not-found' } + }, + _meta: { runtimeId: 'runtime-1' } + } + }) +} + +describe('mobile endpoint supervisor retirement', () => { + beforeEach(() => { + vi.useFakeTimers() + vi.setSystemTime(new Date('2026-07-13T12:00:00Z')) + }) + + afterEach(() => vi.useRealTimers()) + + it('does not publish a resolved relay profile after stopping', async () => { + const logical = new FakeLogicalClient('disconnected', 'lan') + let finishResolve: ((value: typeof relay) => void) | undefined + const resolvePending = new Promise((resolve) => { + finishResolve = resolve + }) + const onHostUpdated = vi.fn() + const deps = dependencies({ + openRelay: vi.fn(() => new FakeRelaySession('disconnected', new RelayOuterError(4409))), + resolveRelay: vi.fn(() => resolvePending), + onHostUpdated + }) + const supervisor = new MobileEndpointSupervisor(logical, host, deps) + + const starting = supervisor.start() + await vi.waitFor(() => expect(deps.resolveRelay).toHaveBeenCalledOnce()) + supervisor.stop() + finishResolve?.(relay) + await starting + + expect(deps.saveHost).not.toHaveBeenCalled() + expect(onHostUpdated).not.toHaveBeenCalled() + }) + + it('does not publish a rotated relay profile after stopping', async () => { + const logical = new FakeLogicalClient('connected', 'lan') + let finishCredentialWrite: (() => void) | undefined + const credentialWritePending = new Promise((resolve) => { + finishCredentialWrite = resolve + }) + const writeBundle = vi + .fn<(value: MobileRelayCredentialBundle) => Promise>() + .mockResolvedValue() + .mockResolvedValueOnce() + .mockReturnValueOnce(credentialWritePending) + const onHostUpdated = vi.fn() + mockCredentialRotation(logical) + const deps = dependencies({ writeBundle, onHostUpdated }) + const supervisor = new MobileEndpointSupervisor(logical, host, deps) + + await supervisor.start() + logical.publishState('connected') + await vi.waitFor(() => expect(writeBundle).toHaveBeenCalledTimes(2)) + supervisor.stop() + finishCredentialWrite?.() + await vi.advanceTimersByTimeAsync(0) + + expect(deps.saveHost).not.toHaveBeenCalled() + expect(onHostUpdated).not.toHaveBeenCalled() + }) +}) diff --git a/mobile/src/transport/mobile-endpoint-supervisor.ts b/mobile/src/transport/mobile-endpoint-supervisor.ts index b16e77a0580..2af0d278583 100644 --- a/mobile/src/transport/mobile-endpoint-supervisor.ts +++ b/mobile/src/transport/mobile-endpoint-supervisor.ts @@ -2,7 +2,10 @@ import { openAuthenticatedDirectEndpoint } from './mobile-direct-endpoint-probe' import type { MobileEndpointSupervisorDependencies } from './mobile-endpoint-supervisor-contract' import { RelayReconnectController } from './mobile-relay-reconnect-controller' import { RelayLeaseRotationTimer } from './mobile-relay-lease-rotation-timer' -import { MobileEndpointHysteresis } from './mobile-endpoint-hysteresis' +import { + createMobileEndpointHysteresis, + type MobileEndpointHysteresis +} from './mobile-endpoint-hysteresis-policy' import { encodeBase64Url, isDirectorResolutionFailure, @@ -21,9 +24,6 @@ import type { HostProfile } from './types' export type { MobileEndpointSupervisorDependencies } from './mobile-endpoint-supervisor-contract' const DIRECT_PROBE_INTERVAL_MS = 15_000 -const DIRECT_OBSERVATION_MS = 30_000 -const MINIMUM_DWELL_MS = 60_000 -const FAILURE_COOLDOWN_MS = 60_000 export class MobileEndpointSupervisor { private bundle: MobileRelayCredentialBundle | null = null @@ -43,12 +43,7 @@ export class MobileEndpointSupervisor { private host: HostProfile, private readonly dependencies: MobileEndpointSupervisorDependencies ) { - this.hysteresis = new MobileEndpointHysteresis(dependencies.now(), { - directSuccessesRequired: 3, - directObservationMs: DIRECT_OBSERVATION_MS, - failureCooldownMs: FAILURE_COOLDOWN_MS, - minimumDwellMs: MINIMUM_DWELL_MS - }) + this.hysteresis = createMobileEndpointHysteresis(dependencies.now()) this.relayReconnect = new RelayReconnectController(dependencies, this.recoverRelay.bind(this)) this.leaseRotation = new RelayLeaseRotationTimer(dependencies, () => { this.relayRotationPending = true @@ -179,6 +174,9 @@ export class MobileEndpointSupervisor { relay: this.host.relay, resumeToken: credential.token }) + if (this.stopped) { + return { ok: false, error: new Error('relay state missing') } + } this.host = await persistRelayHost(this.host, resolved, this.dependencies.saveHost) this.dependencies.onHostUpdated?.(this.host) return await this.openAndMigrateRelay(credential) @@ -300,6 +298,9 @@ export class MobileEndpointSupervisor { writeBundle: this.dependencies.writeBundle, randomBytes: this.dependencies.randomBytes }) + if (this.stopped) { + return + } this.bundle = result.bundle // Why: a scheduled rotation can finish after the old credential enters the rejection gate. credentialRefreshed = true diff --git a/mobile/src/transport/mobile-relay-rpc-session.test.ts b/mobile/src/transport/mobile-relay-rpc-session.test.ts index ad8b5d87a2b..0c0577cd86e 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.test.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.test.ts @@ -218,7 +218,7 @@ describe('mobile relay RPC session', () => { await expect(pending.catch((error: unknown) => isRpcDeliveryUnknown(error))).resolves.toBe(true) }) - it('marks a relay RPC timeout delivery-unknown', async () => { + it('probes before demoting a timed-out relay RPC', async () => { const { session } = await authenticateSession() vi.useFakeTimers() try { @@ -234,9 +234,14 @@ 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('connected') + expect(fakes.close).not.toHaveBeenCalled() + expect( + fakes.sendText.mock.calls.map(([payload]) => JSON.parse(payload as string)).at(-1) + ).toMatchObject({ method: 'status.get' }) + + await vi.advanceTimersByTimeAsync(8_000) + expect(session.getFailure()).toMatchObject({ message: 'relay RPC timed out: status.get' }) expect(session.getState()).toBe('disconnected') expect(fakes.close).toHaveBeenCalledOnce() } finally { @@ -244,6 +249,32 @@ describe('mobile relay RPC session', () => { } }) + it('keeps a relay session when its post-timeout probe answers', async () => { + const { session } = await authenticateSession() + vi.useFakeTimers() + try { + const request = session.sendRequest('browser.screenshot', {}, { timeoutMs: 100 }) + const outcome = request.catch((error: unknown) => error) + + await vi.advanceTimersByTimeAsync(100) + await expect(outcome).resolves.toMatchObject({ + message: 'relay RPC timed out: browser.screenshot' + }) + const probe = fakes.sendText.mock.calls + .map(([payload]) => JSON.parse(payload as string) as { id: string; method: string }) + .find(({ method }) => method === 'status.get')! + fakes.linkOptions!.onText( + JSON.stringify({ id: probe.id, ok: true, result: {}, _meta: { runtimeId: 'runtime-1' } }) + ) + + await vi.advanceTimersByTimeAsync(8_000) + expect(session.getState()).toBe('connected') + expect(fakes.close).not.toHaveBeenCalled() + } finally { + vi.useRealTimers() + } + }) + it('keeps a Relay session when a fresh timeout probe proves it live', async () => { const { session } = await authenticateSession() vi.useFakeTimers() diff --git a/mobile/src/transport/mobile-relay-rpc-session.ts b/mobile/src/transport/mobile-relay-rpc-session.ts index 2b509fa9f88..3b804f645d6 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.ts @@ -152,18 +152,13 @@ export function connectMobileRelayRpcSession(args: { return Promise.reject(new Error('relay session not connected')) } const id = nextId() - const requestControlResponseSequence = controlResponseSequence return new Promise((resolve, reject) => { const timer = setTimeout(() => { pending.delete(id) // Why: the frame was written long ago — the desktop may have processed it. const error = markRpcDeliveryUnknown(new Error(`relay RPC timed out: ${method}`)) reject(error) - if (controlResponseSequence === requestControlResponseSequence) { - fail(error) - } else { - probeControlPlane() - } + probeControlPlane() }, timeoutMs) pending.set(id, { resolve, reject, timer }) if (!sendFrame({ id, method, params })) { diff --git a/mobile/src/transport/rpc-client-request-deadline.test.ts b/mobile/src/transport/rpc-client-request-deadline.test.ts index bfb22171ec1..227d4ed32af 100644 --- a/mobile/src/transport/rpc-client-request-deadline.test.ts +++ b/mobile/src/transport/rpc-client-request-deadline.test.ts @@ -181,6 +181,48 @@ describe('mobile rpc-client request deadline', () => { } }) + it('probes before demoting a request that exceeds its deadline', async () => { + const client = connect('ws://desktop.invalid', 'token', 'server-key') + const socket = mockSockets[0]! + socket.open() + const request = client.sendRequest('speech.dictation.finish', {}, { timeoutMs: 123 }) + const outcome = track(request) + + await vi.advanceTimersByTimeAsync(123) + expect(outcome.read()).toBe('Request timed out: speech.dictation.finish') + const probe = sentRequest(socket, 'status.get') + expect(socket.close).not.toHaveBeenCalled() + expect(client.getState()).toBe('connected') + + socket.receive(`encrypted:${JSON.stringify({ id: probe.id, ok: true, result: {} })}`) + await vi.advanceTimersByTimeAsync(8_000) + expect(socket.close).not.toHaveBeenCalled() + expect(client.getState()).toBe('connected') + + client.close() + await request.catch(() => undefined) + }) + + it('demotes when the post-timeout control probe also stalls', async () => { + const client = connect('ws://desktop.invalid', 'token', 'server-key') + const socket = mockSockets[0]! + socket.open() + const request = client.sendRequest('browser.screenshot', {}, { timeoutMs: 123 }) + const outcome = track(request) + + await vi.advanceTimersByTimeAsync(123) + expect(outcome.read()).toBe('Request timed out: browser.screenshot') + expect(sentRequest(socket, 'status.get')).toBeDefined() + expect(client.getState()).toBe('connected') + + await vi.advanceTimersByTimeAsync(8_000) + expect(socket.close).toHaveBeenCalled() + expect(client.getState()).toBe('reconnecting') + + client.close() + 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]! diff --git a/mobile/src/transport/rpc-client.test.ts b/mobile/src/transport/rpc-client.test.ts index c9294d4c935..8303a45263e 100644 --- a/mobile/src/transport/rpc-client.test.ts +++ b/mobile/src/transport/rpc-client.test.ts @@ -535,37 +535,6 @@ describe('mobile rpc-client connection timeout', () => { client.close() }) - it('honors per-request timeout overrides', async () => { - const client = connect('ws://desktop.invalid', 'token', 'server-key') - const socket = mockSockets[0]! - - socket.open() - socket.receive(JSON.stringify({ type: 'e2ee_ready' })) - socket.receive('encrypted:{"type":"e2ee_authenticated"}') - socket.emitCloseOnClose = false - - const request = client.sendRequest( - 'speech.dictation.finish', - { dictationId: 'd1' }, - { - timeoutMs: 123 - } - ) - await Promise.resolve() - - await vi.advanceTimersByTimeAsync(122) - await expect( - Promise.race([request.then(() => 'settled'), Promise.resolve('pending')]) - ).resolves.toBe('pending') - - await vi.advanceTimersByTimeAsync(1) - await expect(request).rejects.toThrow('Request timed out: speech.dictation.finish') - expect(socket.close).toHaveBeenCalled() - expect(client.getState()).toBe('reconnecting') - - client.close() - }) - it('applies per-request timeout overrides while waiting for reconnect', async () => { const client = connect('ws://desktop.invalid', 'token', 'server-key') const socket = mockSockets[0]! diff --git a/mobile/src/transport/rpc-client.ts b/mobile/src/transport/rpc-client.ts index f585afdb14a..1c3aca20455 100644 --- a/mobile/src/transport/rpc-client.ts +++ b/mobile/src/transport/rpc-client.ts @@ -768,8 +768,8 @@ export function connect( } // Why: stream frames can flow while control RPC is wedged; only a control response satisfies this probe. - function runActivityProbe() { - if (state !== 'connected' || !ws || activityProbeInFlight) { + function runActivityProbe(expectedWs: WebSocket | null = ws) { + if (state !== 'connected' || !ws || ws !== expectedWs || activityProbeInFlight) { return } activityProbeInFlight = true @@ -1037,7 +1037,7 @@ export function connect( // Why: the frame was written 30s ago — the host may have processed it. reject(markRpcDeliveryUnknown(new Error(`Request timed out: ${method}`))) if (controlResponseSequence === requestControlResponseSequence) { - forceSocketReconnect(requestWs) + runActivityProbe(requestWs) } }, timeoutMs)