From 4e578fce259d29fece00bdab4a7828544dbc704c Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Sat, 1 Aug 2026 03:07:14 -0700 Subject: [PATCH] fix(mobile): fence reconnect liveness races --- .../client-context-connection-metrics.ts | 3 +- mobile/src/transport/client-context.test.ts | 74 ++++++- mobile/src/transport/client-context.tsx | 193 +++++++++--------- .../transport/existing-host-relay-routing.ts | 28 +++ mobile/src/transport/host-force-reconnect.ts | 15 +- .../transport/host-reconnect-profile-cache.ts | 28 ++- mobile/src/transport/host-store.test.ts | 38 +++- mobile/src/transport/host-store.ts | 22 +- .../transport/mobile-endpoint-lifecycle.ts | 4 +- .../mobile-relay-direct-upgrade.test.ts | 2 +- .../transport/mobile-relay-direct-upgrade.ts | 9 +- .../mobile-relay-rpc-session.test.ts | 60 ++++++ .../src/transport/mobile-relay-rpc-session.ts | 48 ++++- .../transport/rpc-client-context-contract.ts | 18 ++ .../rpc-client-request-deadline.test.ts | 22 ++ mobile/src/transport/rpc-client.ts | 50 ++--- .../transport/rpc-stream-response-shape.ts | 21 ++ 17 files changed, 455 insertions(+), 180 deletions(-) create mode 100644 mobile/src/transport/existing-host-relay-routing.ts create mode 100644 mobile/src/transport/rpc-client-context-contract.ts create mode 100644 mobile/src/transport/rpc-stream-response-shape.ts diff --git a/mobile/src/transport/client-context-connection-metrics.ts b/mobile/src/transport/client-context-connection-metrics.ts index 391c0cb0568..e41d2241d75 100644 --- a/mobile/src/transport/client-context-connection-metrics.ts +++ b/mobile/src/transport/client-context-connection-metrics.ts @@ -1,5 +1,6 @@ import { useEffect, useState } from 'react' -import { useRpcClientContext, type RpcClientContextValue } from './client-context' +import { useRpcClientContext } from './client-context' +import type { RpcClientContextValue } from './rpc-client-context-contract' export function useReconnectAttempt(hostId: string | undefined): number { return useHostMetric(hostId, (context, id) => context.getReconnectAttempt(id), 0) diff --git a/mobile/src/transport/client-context.test.ts b/mobile/src/transport/client-context.test.ts index 84d01e272d2..70006816021 100644 --- a/mobile/src/transport/client-context.test.ts +++ b/mobile/src/transport/client-context.test.ts @@ -365,6 +365,30 @@ describe('useHostClient', () => { harness.unmount() }) + it('ignores relay metadata published by a superseded lifecycle', async () => { + const stale = makeFakeClient('connected') + const fresh = makeFakeClient('connecting') + fresh.sendRequest = vi.fn(async () => ({ id: 'health', ok: true, result: {} })) + connectMock.mockReturnValueOnce(stale).mockReturnValueOnce(fresh) + loadHostsMock.mockResolvedValue([HOST]) + + const harness = await renderHarness(HOST.id) + const updatedHost = { ...HOST, endpoint: 'ws://127.0.0.1:2' } + harness.primeHosts([updatedHost]) + const publishHostUpdate = connectMock.mock.calls[0]?.[2] as + | ((host: HostProfile) => void) + | undefined + publishHostUpdate?.({ + ...HOST, + endpoints: [{ id: 'direct-primary', kind: 'lan', url: HOST.endpoint }] + }) + + await act(async () => harness.forceReconnect(HOST.id)) + + expect(connectMock.mock.calls[1]?.[0]).toEqual(updatedHost) + harness.unmount() + }) + it('supersedes a pending old-profile open when Force Reconnect uses a saved endpoint', async () => { let resolveOldLookup: ((hosts: HostProfile[]) => void) | null = null const oldLookup = new Promise((resolve) => { @@ -423,17 +447,43 @@ describe('useHostClient', () => { harness.unmount() }) - it('queues a changed endpoint behind an older reconnect health probe', async () => { + it('opens a changed endpoint without waiting for a superseded host lookup', async () => { + const stale = makeFakeClient('connected') + const fresh = makeFakeClient('connecting') + fresh.sendRequest = vi.fn(async () => ({ id: 'health', ok: true, result: {} })) + loadHostsMock.mockResolvedValueOnce([HOST]).mockReturnValueOnce(new Promise(() => {})) + connectMock.mockReturnValueOnce(stale).mockReturnValueOnce(fresh) + + const harness = await renderHarness(HOST.id) + void harness.forceReconnect(HOST.id) + await act(async () => { + await Promise.resolve() + await Promise.resolve() + }) + + const updatedHost = { ...HOST, endpoint: 'ws://127.0.0.1:2' } + harness.primeHosts([updatedHost]) + await act(async () => harness.forceReconnect(HOST.id)) + + expect(connectMock).toHaveBeenCalledTimes(2) + expect(connectMock.mock.calls[1]?.[0]).toEqual(updatedHost) + harness.unmount() + }) + + it('supersedes an older reconnect health probe immediately', async () => { const stale = makeFakeClient('connected') const oldReplacement = makeFakeClient('connecting') const newReplacement = makeFakeClient('connecting') - let resolveOldHealthCheck: (() => void) | null = null + let rejectOldHealthCheck: ((error: Error) => void) | null = null oldReplacement.sendRequest = vi.fn( () => - new Promise((resolve) => { - resolveOldHealthCheck = () => resolve({ id: 'old-health', ok: true, result: {} }) + new Promise((_resolve, reject) => { + rejectOldHealthCheck = reject }) ) + oldReplacement.closeMock.mockImplementation(() => { + rejectOldHealthCheck?.(new Error('Client closed')) + }) newReplacement.sendRequest = vi.fn(async () => ({ id: 'new-health', ok: true, result: {} })) connectMock .mockReturnValueOnce(stale) @@ -452,14 +502,12 @@ describe('useHostClient', () => { harness.primeHosts([updatedHost]) const second = harness.forceReconnect(HOST.id) expect(second).not.toBe(first) - expect(connectMock).toHaveBeenCalledTimes(2) + expect(connectMock).toHaveBeenCalledTimes(3) - resolveOldHealthCheck?.() await act(async () => { await Promise.all([first, second]) }) - expect(connectMock).toHaveBeenCalledTimes(3) expect(connectMock.mock.calls[2]?.[0]).toEqual(updatedHost) expect(oldReplacement.closeMock).toHaveBeenCalledOnce() expect(newReplacement.sendRequest).toHaveBeenCalledWith('status.get', undefined, { @@ -471,9 +519,11 @@ describe('useHostClient', () => { harness.unmount() }) - it('keeps a host disconnected when it closes with a changed-profile reconnect queued', async () => { + it('keeps a host disconnected when it closes during a changed-profile reconnect', async () => { const stale = makeFakeClient('connected') const oldReplacement = makeFakeClient('connecting') + const newReplacement = makeFakeClient('connecting') + newReplacement.sendRequest = vi.fn(async () => ({ id: 'new-health', ok: true, result: {} })) let rejectOldHealthCheck: ((error: Error) => void) | null = null oldReplacement.sendRequest = vi.fn( () => @@ -484,7 +534,10 @@ describe('useHostClient', () => { oldReplacement.closeMock.mockImplementation(() => { rejectOldHealthCheck?.(new Error('Client closed')) }) - connectMock.mockReturnValueOnce(stale).mockReturnValueOnce(oldReplacement) + connectMock + .mockReturnValueOnce(stale) + .mockReturnValueOnce(oldReplacement) + .mockReturnValueOnce(newReplacement) loadHostsMock.mockResolvedValue([HOST]) const harness = await renderHarness(HOST.id) @@ -500,8 +553,9 @@ describe('useHostClient', () => { await act(async () => Promise.all([first, queued])) - expect(connectMock).toHaveBeenCalledTimes(2) + expect(connectMock).toHaveBeenCalledTimes(3) expect(oldReplacement.closeMock).toHaveBeenCalledOnce() + expect(newReplacement.closeMock).toHaveBeenCalledOnce() expect(harness.hook.client).toBeNull() expect(harness.hook.state).toBe('disconnected') diff --git a/mobile/src/transport/client-context.tsx b/mobile/src/transport/client-context.tsx index e22caac9d60..5aeec659a36 100644 --- a/mobile/src/transport/client-context.tsx +++ b/mobile/src/transport/client-context.tsx @@ -13,12 +13,13 @@ import { import type { RpcClient } from './rpc-client' import { connectionLogStore } from './connection-log-buffer' import { HostForceReconnectCoordinator, type HostReconnectEntry } from './host-force-reconnect' -import { HostReconnectProfileCache } from './host-reconnect-profile-cache' +import { HostReconnectProfileCache, type HostOpenProfile } from './host-reconnect-profile-cache' import { subscribeConnectionRevivalTriggers } from './connection-revival-triggers' import { HostClientOpenRegistry } from './host-client-open-registry' import { loadHosts } from './host-store' import { openHostLogicalClient } from './host-logical-client' import { clientActivePath } from './rpc-client-active-path' +import type { RpcClientContextValue } from './rpc-client-context-contract' import type { MobileConnectionPath } from './stable-logical-rpc-client' import type { ConnectionLogSink, ConnectionState, HostProfile } from './types' @@ -26,23 +27,6 @@ type StoreEntry = HostReconnectEntry & { state: ConnectionState } -export type RpcClientContextValue = { - acquire: (hostId: string, host?: HostProfile) => RpcClient | null - release: (hostId: string) => void - forceReconnect: (hostId: string) => Promise - closeHost: (hostId: string) => void - getState: (hostId: string) => ConnectionState - getReconnectAttempt: (hostId: string) => number - // Why: ms-epoch of the last 'connected' (null if never this session); UI escalates "Reconnecting…" into a re-pair prompt. - getLastConnectedAt: (hostId: string) => number | null - getActivePath: (hostId: string) => MobileConnectionPath - subscribeHostState: (hostId: string, listener: (state: ConnectionState) => void) => () => void - getAllClients: () => Array<{ hostId: string; client: RpcClient }> - subscribeAllHosts: (listener: () => void) => () => void - // Why: lets the home screen feed already-loaded HostProfiles so we don't pay loadHosts() latency twice. - primeHosts: (hosts: HostProfile[]) => void -} - const Ctx = createContext(null) export function RpcClientProvider({ children }: { children: ReactNode }) { @@ -86,85 +70,90 @@ export function RpcClientProvider({ children }: { children: ReactNode }) { notifyAllHosts() }, []) - const openEntry = useCallback(async (hostId: string): Promise => { - const profileVersion = primedHostsRef.current.version(hostId) - const existing = pendingOpensRef.current.getActivePromise(hostId, profileVersion) - if (existing) { - await existing - return storeRef.current.get(hostId) ?? null - } - let resolve: () => void = () => {} - const promise = new Promise((res) => { - resolve = res - }) - const pendingOpen = pendingOpensRef.current.register(hostId, profileVersion, promise) - - try { - // Why: prefer the primed cache so we don't serialize a second Keychain pass on cold start. - let host = primedHostsRef.current.get(hostId) - if (!host) { - try { - const hosts = await loadHosts() - host = hosts.find((h) => h.id === hostId) - } catch { - // Why: cold-start Keychain failure (iOS mid-unlock / Android Keystore race); surface 'disconnected' so the user can Reconnect. - notifyHostState(hostId, 'disconnected') - notifyAllHosts() - return null - } - if (!host) { - // Why: silent return leaves screens on a permanent spinner (STA-1511); surface 'disconnected' so they show a retry affordance. - notifyHostState(hostId, 'disconnected') - notifyAllHosts() - return null - } + const openEntry = useCallback( + async (hostId: string, requestedProfile?: HostOpenProfile): Promise => { + const profileVersion = requestedProfile?.version ?? primedHostsRef.current.version(hostId) + const existing = pendingOpensRef.current.getActivePromise(hostId, profileVersion) + if (existing) { + await existing + return storeRef.current.get(hostId) ?? null } - - if (pendingOpen.cancelled) { - return null - } - - // Re-check after any await — another acquire() may have completed. - const after = storeRef.current.get(hostId) - if (after) { - return after - } - - let client: RpcClient - const onLog: ConnectionLogSink = (entry) => connectionLogStore.append(hostId, entry) - try { - client = openHostLogicalClient(host, onLog, (updatedHost) => - primedHostsRef.current.prime(updatedHost) - ) - } catch { - // Why: openHostLogicalClient can throw synchronously (bad public key / invalid URL); notify so the UI leaves 'connecting'. - notifyHostState(hostId, 'disconnected') - notifyAllHosts() - return null - } - const unsubState = client.onStateChange((state) => { - const cur = storeRef.current.get(hostId) - if (!cur) { - return - } - cur.state = state - notifyHostState(hostId, state) + let resolve: () => void = () => {} + const promise = new Promise((res) => { + resolve = res }) - const entry: StoreEntry = { - client, - state: client.getState(), - refCount: 0, - unsubState + const pendingOpen = pendingOpensRef.current.register(hostId, profileVersion, promise) + + try { + // Why: prefer the primed cache so we don't serialize a second Keychain pass on cold start. + let host = requestedProfile?.host ?? primedHostsRef.current.get(hostId) + if (!host) { + try { + const hosts = await loadHosts() + host = hosts.find((h) => h.id === hostId) + } catch { + // Why: cold-start Keychain failure (iOS mid-unlock / Android Keystore race); surface 'disconnected' so the user can Reconnect. + notifyHostState(hostId, 'disconnected') + notifyAllHosts() + return null + } + if (!host) { + // Why: silent return leaves screens on a permanent spinner (STA-1511); surface 'disconnected' so they show a retry affordance. + notifyHostState(hostId, 'disconnected') + notifyAllHosts() + return null + } + } + + if (pendingOpen.cancelled) { + return null + } + + // Re-check after any await — another acquire() may have completed. + const after = storeRef.current.get(hostId) + if (after) { + return after + } + + let client: RpcClient + const onLog: ConnectionLogSink = (entry) => connectionLogStore.append(hostId, entry) + try { + client = openHostLogicalClient( + host, + onLog, + primedHostsRef.current.publisher(hostId, profileVersion) + ) + } catch { + // Why: openHostLogicalClient can throw synchronously (bad public key / invalid URL); notify so the UI leaves 'connecting'. + notifyHostState(hostId, 'disconnected') + notifyAllHosts() + return null + } + const unsubState = client.onStateChange((state) => { + const cur = storeRef.current.get(hostId) + if (!cur) { + return + } + cur.state = state + notifyHostState(hostId, state) + }) + const entry: StoreEntry = { + client, + state: client.getState(), + refCount: 0, + unsubState + } + storeRef.current.set(hostId, entry) + notifyHostState(hostId, entry.state) + notifyAllHosts() + return entry + } finally { + pendingOpensRef.current.deleteIfCurrent(hostId, pendingOpen) + resolve() } - storeRef.current.set(hostId, entry) - notifyHostState(hostId, entry.state) - notifyAllHosts() - return entry - } finally { - pendingOpensRef.current.deleteIfCurrent(hostId, pendingOpen) - resolve() - } - }, []) + }, + [] + ) // Synchronous get-or-open: returns an existing client immediately, else kicks off an async open and returns null this tick. const acquire = useCallback( @@ -205,10 +194,14 @@ export function RpcClientProvider({ children }: { children: ReactNode }) { }, []) const forceReconnect = useCallback( - (hostId: string): Promise => - forceReconnectCoordinatorRef.current.run({ + (hostId: string): Promise => { + const profile = { + host: primedHostsRef.current.get(hostId), + version: primedHostsRef.current.version(hostId) + } + return forceReconnectCoordinatorRef.current.run({ hostId, - profileVersion: primedHostsRef.current.version(hostId), + profileVersion: profile.version, getEntry: () => storeRef.current.get(hostId), getListenerCount: () => stateListenersRef.current.get(hostId)?.size ?? 0, removeEntry: () => { @@ -216,8 +209,10 @@ export function RpcClientProvider({ children }: { children: ReactNode }) { notifyHostState(hostId, 'disconnected') notifyAllHosts() }, - openReplacement: () => openEntry(hostId) - }), + cancelPendingOpen: () => pendingOpensRef.current.cancel(hostId), + openReplacement: () => openEntry(hostId, profile) + }) + }, [openEntry] ) diff --git a/mobile/src/transport/existing-host-relay-routing.ts b/mobile/src/transport/existing-host-relay-routing.ts new file mode 100644 index 00000000000..91c33afcc79 --- /dev/null +++ b/mobile/src/transport/existing-host-relay-routing.ts @@ -0,0 +1,28 @@ +import { HostProfileSchema, type HostProfile } from './types' +import { mutateStoredHosts } from './host-store' +import * as hostListLoads from './host-list-load-sharing' +import { saveMobileRelayHostOverlay } from './mobile-relay-host-overlay-store' + +export class MobileRelayUpgradeHostRemovedError extends Error {} + +export async function saveExistingHostRelayRouting(host: HostProfile): Promise { + const validated = HostProfileSchema.parse(host) + await mutateStoredHosts((hosts) => { + const existing = hosts.find(({ id }) => id === validated.id) + if (!existing || existing.publicKeyB64 !== validated.publicKeyB64) { + throw new MobileRelayUpgradeHostRemovedError('mobile relay upgrade host was removed') + } + return hosts + }) + if (!validated.endpoints || !validated.relayHostId || !validated.relay) { + throw new Error('mobile relay upgrade routing metadata missing') + } + await saveMobileRelayHostOverlay({ + v: 2, + hostId: validated.id, + endpoints: validated.endpoints, + relayHostId: validated.relayHostId, + relay: validated.relay + }) + hostListLoads.dropSharedHostListLoad() +} diff --git a/mobile/src/transport/host-force-reconnect.ts b/mobile/src/transport/host-force-reconnect.ts index daf0fd5a228..fb16671f9bf 100644 --- a/mobile/src/transport/host-force-reconnect.ts +++ b/mobile/src/transport/host-force-reconnect.ts @@ -13,6 +13,7 @@ type HostReconnectOperation = { getEntry: () => HostReconnectEntry | undefined getListenerCount: () => number removeEntry: () => void + cancelPendingOpen: () => void openReplacement: () => Promise } @@ -38,17 +39,17 @@ export class HostForceReconnectCoordinator { } run(operation: HostReconnectOperation): Promise { - const generation = this.generation(operation.hostId) const pending = this.pendingByHost.get(operation.hostId) + let generation = this.generation(operation.hostId) if (pending?.profileVersion === operation.profileVersion && pending.generation === generation) { return pending.promise } - // Why: a changed endpoint must supersede the in-flight profile without racing its health probe. - const reconnect = pending - ? pending.promise - .catch(() => undefined) - .then(() => this.replaceAndVerify(operation, generation)) - : this.replaceAndVerify(operation, generation) + if (pending) { + generation += 1 + this.generations.set(operation.hostId, generation) + operation.cancelPendingOpen() + } + const reconnect = this.replaceAndVerify(operation, generation) this.pendingByHost.set(operation.hostId, { profileVersion: operation.profileVersion, generation, diff --git a/mobile/src/transport/host-reconnect-profile-cache.ts b/mobile/src/transport/host-reconnect-profile-cache.ts index 227e9151c15..5c6b50ec517 100644 --- a/mobile/src/transport/host-reconnect-profile-cache.ts +++ b/mobile/src/transport/host-reconnect-profile-cache.ts @@ -5,11 +5,16 @@ type CachedHostProfile = { version: number } +export type HostOpenProfile = { + host: HostProfile | undefined + version: number +} + export class HostReconnectProfileCache { private readonly profiles = new Map() private readonly latestVersions = new Map() - prime(host: HostProfile): void { + prime(host: HostProfile): number { const current = this.profiles.get(host.id) const version = current && reconnectProfileMatches(current.host, host) @@ -17,6 +22,27 @@ export class HostReconnectProfileCache { : (this.latestVersions.get(host.id) ?? 0) + 1 this.latestVersions.set(host.id, version) this.profiles.set(host.id, { host, version }) + return version + } + + primeFromVersion(host: HostProfile, sourceVersion: number): number | null { + if (this.version(host.id) !== sourceVersion) { + return null + } + return this.prime(host) + } + + publisher(hostId: string, initialVersion: number): (host: HostProfile) => void { + let sourceVersion = initialVersion + return (host) => { + if (host.id !== hostId) { + return + } + const nextVersion = this.primeFromVersion(host, sourceVersion) + if (nextVersion !== null) { + sourceVersion = nextVersion + } + } } get(hostId: string): HostProfile | undefined { diff --git a/mobile/src/transport/host-store.test.ts b/mobile/src/transport/host-store.test.ts index 997e8a3d42f..dc50e5f8e0c 100644 --- a/mobile/src/transport/host-store.test.ts +++ b/mobile/src/transport/host-store.test.ts @@ -36,15 +36,17 @@ vi.mock('./host-credential-cleanup', () => ({ import { loadHosts, - MobileRelayUpgradeHostRemovedError, removeHost, resolvePairingHostIdentity, resetHostStoreForTests, saveHost, - saveExistingHostRelayUpgrade, updateHostNameAndEndpoint, updateLastConnected } from './host-store' +import { + MobileRelayUpgradeHostRemovedError, + saveExistingHostRelayRouting +} from './existing-host-relay-routing' import { resetMobileRelayHostOverlayStoreForTests } from './mobile-relay-host-overlay-store' const HOSTS_STORAGE_KEY = 'orca:hosts' @@ -247,13 +249,43 @@ describe('host-store list mutations', () => { storedHostsRaw = JSON.stringify([HOST_TWO]) await expect( - saveExistingHostRelayUpgrade({ ...HOST_ONE, deviceToken: 'token-1' }) + saveExistingHostRelayRouting({ ...HOST_ONE, deviceToken: 'token-1' }) ).rejects.toBeInstanceOf(MobileRelayUpgradeHostRemovedError) expect(JSON.parse(storedHostsRaw)).toEqual([HOST_TWO]) expect(secureStoreMock.setItemAsync).not.toHaveBeenCalled() }) + it('publishes relay routing without overwriting a newer endpoint edit', async () => { + const relayHostId = 'AbCdEf0123_-xyZ9' + await updateHostNameAndEndpoint(HOST_ONE.id, { endpoint: 'ws://127.0.0.1:9' }) + + await saveExistingHostRelayRouting({ + ...HOST_ONE, + deviceToken: 'token-1', + endpoints: [ + { id: 'direct-primary', kind: 'lan', url: HOST_ONE.endpoint }, + { id: 'relay-primary', kind: 'relay', url: 'wss://relay.onorca.dev/v1/connect/host' } + ], + relayHostId, + relay: { + v: 1, + directorUrl: 'https://relay.onorca.dev', + cellUrl: 'https://relay.onorca.dev', + assignmentEpoch: 1, + relayHostId, + e2eeFraming: 2 + } + }) + + expect(JSON.parse(storedHostsRaw)).toContainEqual( + expect.objectContaining({ id: HOST_ONE.id, endpoint: 'ws://127.0.0.1:9' }) + ) + expect(JSON.parse(storedOverlayRaw ?? '[]')).toContainEqual( + expect.objectContaining({ hostId: HOST_ONE.id, relayHostId }) + ) + }) + it('awaits cleanup scheduling after metadata commit', async () => { let resolveSchedule: (() => void) | null = null scheduleCleanupMock.mockReturnValue( diff --git a/mobile/src/transport/host-store.ts b/mobile/src/transport/host-store.ts index 43657fffe76..6c1ab5a0c63 100644 --- a/mobile/src/transport/host-store.ts +++ b/mobile/src/transport/host-store.ts @@ -186,14 +186,16 @@ async function readStoredHostsForMutation(): Promise { } } -async function mutateStoredHosts( +export async function mutateStoredHosts( update: (hosts: StoredHostProfile[]) => StoredHostProfile[] ): Promise { const mutation = hostListMutation.then(async () => { const current = await readStoredHostsForMutation() const next = update(current) - await AsyncStorage.setItem(STORAGE_KEY, JSON.stringify(next)) - hostListLoads.dropSharedHostListLoad() + if (next !== current) { + await AsyncStorage.setItem(STORAGE_KEY, JSON.stringify(next)) + hostListLoads.dropSharedHostListLoad() + } }) hostListMutation = mutation.catch(() => {}) return mutation @@ -209,17 +211,11 @@ function toStored(host: HostProfile): StoredHostProfile { } } -export class MobileRelayUpgradeHostRemovedError extends Error {} - export async function saveHost(host: HostProfile): Promise { - await persistHost(host, false) + await persistHost(host) } -export async function saveExistingHostRelayUpgrade(host: HostProfile): Promise { - await persistHost(host, true) -} - -async function persistHost(host: HostProfile, requireExisting: boolean): Promise { +async function persistHost(host: HostProfile): Promise { const validated = HostProfileSchema.parse(host) const stored = toStored(validated) const duplicateHostIds = new Set() @@ -238,10 +234,6 @@ async function persistHost(host: HostProfile, requireExisting: boolean): Promise .filter(({ id }) => !duplicateHostIds.has(id)) .map((candidate) => (candidate.id === stored.id ? stored : candidate)) } - if (requireExisting) { - // Why: an in-flight relay upgrade must not resurrect a host the user removed. - throw new MobileRelayUpgradeHostRemovedError('mobile relay upgrade host was removed') - } return [...hosts.filter(({ id }) => !duplicateHostIds.has(id)), stored] }) // Why: write metadata before the keychain token so a crash leaves recoverable orphaned metadata, not an orphaned token that persists forever. diff --git a/mobile/src/transport/mobile-endpoint-lifecycle.ts b/mobile/src/transport/mobile-endpoint-lifecycle.ts index c8f9852de0e..c03f710538c 100644 --- a/mobile/src/transport/mobile-endpoint-lifecycle.ts +++ b/mobile/src/transport/mobile-endpoint-lifecycle.ts @@ -8,7 +8,7 @@ import { readMobileRelayCredentialBundle, writeMobileRelayCredentialBundle } from './mobile-relay-credential-bundle' -import { saveHost } from './host-store' +import { saveExistingHostRelayRouting } from './existing-host-relay-routing' import { upgradeDirectMobileRelay } from './mobile-relay-direct-upgrade' import { MobileRelayDirectUpgradeController } from './mobile-relay-direct-upgrade-controller' import type { StableLogicalRpcClient } from './stable-logical-rpc-client' @@ -99,7 +99,7 @@ function createSupervisor( resolveRelay: resolveMobileRelayEndpoint, readBundle: readMobileRelayCredentialBundle, writeBundle: writeMobileRelayCredentialBundle, - saveHost, + saveHost: saveExistingHostRelayRouting, onHostUpdated, now: Date.now, randomBytes: ExpoCrypto.getRandomBytes, diff --git a/mobile/src/transport/mobile-relay-direct-upgrade.test.ts b/mobile/src/transport/mobile-relay-direct-upgrade.test.ts index d77848b6e17..832febfe190 100644 --- a/mobile/src/transport/mobile-relay-direct-upgrade.test.ts +++ b/mobile/src/transport/mobile-relay-direct-upgrade.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it, vi } from 'vitest' import type { MobileRelayEndpoint } from '../../../src/shared/mobile-relay-credential-contract' -import { MobileRelayUpgradeHostRemovedError } from './host-store' +import { MobileRelayUpgradeHostRemovedError } from './existing-host-relay-routing' import { createMobileRelayDirectUpgradeJournal, type MobileRelayDirectUpgradeJournal diff --git a/mobile/src/transport/mobile-relay-direct-upgrade.ts b/mobile/src/transport/mobile-relay-direct-upgrade.ts index b804d6fb610..a2d907bc8a5 100644 --- a/mobile/src/transport/mobile-relay-direct-upgrade.ts +++ b/mobile/src/transport/mobile-relay-direct-upgrade.ts @@ -5,7 +5,10 @@ import { type DeviceCredentialInstalled, type PairingGetEndpointsResult } from '../../../src/shared/mobile-relay-credential-contract' -import { MobileRelayUpgradeHostRemovedError, saveExistingHostRelayUpgrade } from './host-store' +import { + MobileRelayUpgradeHostRemovedError, + saveExistingHostRelayRouting +} from './existing-host-relay-routing' import { persistRelayHost } from './mobile-endpoint-supervisor-support' import { MobileRelayCredentialBundleSchema, @@ -33,7 +36,7 @@ type Dependencies = { writeJournal: typeof writeMobileRelayDirectUpgradeJournal clearJournal: typeof deleteMobileRelayDirectUpgradeJournal writeBundle: typeof writeMobileRelayCredentialBundle - saveHost: typeof saveExistingHostRelayUpgrade + saveHost: typeof saveExistingHostRelayRouting deleteBundle: typeof deleteMobileRelayCredentialBundle randomBytes: (length: number) => Uint8Array } @@ -51,7 +54,7 @@ export async function upgradeDirectMobileRelay(args: { writeJournal: writeMobileRelayDirectUpgradeJournal, clearJournal: deleteMobileRelayDirectUpgradeJournal, writeBundle: writeMobileRelayCredentialBundle, - saveHost: saveExistingHostRelayUpgrade, + saveHost: saveExistingHostRelayRouting, deleteBundle: deleteMobileRelayCredentialBundle, randomBytes: ExpoCrypto.getRandomBytes, ...args.dependencies diff --git a/mobile/src/transport/mobile-relay-rpc-session.test.ts b/mobile/src/transport/mobile-relay-rpc-session.test.ts index 0c0577cd86e..406293ed058 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.test.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.test.ts @@ -275,6 +275,66 @@ describe('mobile relay RPC session', () => { } }) + it('counts a late timed-out reply as relay control-plane liveness', 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(0) + const timedOutRequest = fakes.sendText.mock.calls + .map(([payload]) => JSON.parse(payload as string) as { id: string; method: string }) + .find(({ method }) => method === 'browser.screenshot')! + + await vi.advanceTimersByTimeAsync(100) + await expect(outcome).resolves.toMatchObject({ + message: 'relay RPC timed out: browser.screenshot' + }) + fakes.linkOptions!.onText( + JSON.stringify({ id: timedOutRequest.id, ok: true, result: {}, _meta: {} }) + ) + + await vi.advanceTimersByTimeAsync(8_000) + expect(session.getState()).toBe('connected') + expect(fakes.close).not.toHaveBeenCalled() + } finally { + session.close() + vi.useRealTimers() + } + }) + + it('periodically demotes a silent half-open Relay session', async () => { + vi.useFakeTimers() + try { + const { session } = await authenticateSession() + + await vi.advanceTimersByTimeAsync(20_000) + expect( + fakes.sendText.mock.calls.map(([payload]) => JSON.parse(payload as string)).at(-1) + ).toMatchObject({ method: 'status.get' }) + expect(session.getState()).toBe('connected') + + await vi.advanceTimersByTimeAsync(8_000) + expect(session.getState()).toBe('disconnected') + expect(fakes.close).toHaveBeenCalledOnce() + } finally { + vi.useRealTimers() + } + }) + + it('stops periodic Relay probing when the session closes', async () => { + vi.useFakeTimers() + try { + const { session } = await authenticateSession() + session.close() + + await vi.advanceTimersByTimeAsync(20_000) + expect(fakes.sendText).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 3b804f645d6..6b72419f472 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.ts @@ -13,6 +13,7 @@ import type { RpcClient } from './rpc-client' import type { ConnectionState, RpcResponse } from './types' const CONTROL_PROBE_TIMEOUT_MS = 8_000 +const CONTROL_PROBE_INTERVAL_MS = 20_000 type PendingRequest = { resolve: (response: RpcResponse) => void @@ -38,11 +39,13 @@ export function connectMobileRelayRpcSession(args: { }): MobileRelayRpcSession { const requestTimeoutMs = args.requestTimeoutMs ?? 30_000 const pending = new Map() + const timedOutControlRequestIds = new Set() const stateListeners = new Set<(state: ConnectionState) => void>() let state: ConnectionState = 'connecting' let requestCounter = 0 let controlResponseSequence = 0 let controlProbeInFlight = false + let controlProbeTimer: ReturnType | null = null let lastConnectedAt: number | null = null let leaseExpiresAt: number | null = null let resumeConfirmation: DeviceResumeConfirmed | null = null @@ -102,12 +105,16 @@ export function connectMobileRelayRpcSession(args: { stateListeners.add(listener) return () => stateListeners.delete(listener) }, - notifyForeground: () => {}, + notifyForeground() { + startControlProbeTimer() + probeControlPlane() + }, close() { if (closed) { return } closed = true + stopControlProbeTimer() link.close() rejectPending(new Error('Client closed')) streams.clear() @@ -137,6 +144,7 @@ export function connectMobileRelayRpcSession(args: { resumeConfirmation = result.resumeConfirmation lastConnectedAt = Date.now() publishState('connected') + startControlProbeTimer() } catch (error) { fail(asError(error)) } @@ -155,6 +163,7 @@ export function connectMobileRelayRpcSession(args: { return new Promise((resolve, reject) => { const timer = setTimeout(() => { pending.delete(id) + timedOutControlRequestIds.add(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) @@ -184,8 +193,10 @@ export function connectMobileRelayRpcSession(args: { return } const request = pending.get(value.id) - if (request) { + if (request || timedOutControlRequestIds.delete(value.id)) { controlResponseSequence += 1 + } + if (request) { clearTimeout(request.timer) pending.delete(value.id) request.resolve(value) @@ -206,25 +217,41 @@ export function connectMobileRelayRpcSession(args: { const probeControlResponseSequence = controlResponseSequence void sendRpc('status.get', undefined, CONTROL_PROBE_TIMEOUT_MS).then( () => { - controlProbeInFlight = false + finishControlProbe() }, (error: unknown) => { - controlProbeInFlight = false + const controlResponded = controlResponseSequence > probeControlResponseSequence + finishControlProbe() if (closed || state !== 'connected') { return } - if ( - !isRpcDeliveryUnknown(error) || - controlResponseSequence === probeControlResponseSequence - ) { + if (!isRpcDeliveryUnknown(error) || !controlResponded) { fail(asError(error)) - } else { - probeControlPlane() } } ) } + function startControlProbeTimer(): void { + if (closed || state !== 'connected' || controlProbeTimer) { + return + } + controlProbeTimer = setInterval(probeControlPlane, CONTROL_PROBE_INTERVAL_MS) + } + + function stopControlProbeTimer(): void { + if (controlProbeTimer) { + clearInterval(controlProbeTimer) + controlProbeTimer = null + } + finishControlProbe() + } + + function finishControlProbe(): void { + controlProbeInFlight = false + timedOutControlRequestIds.clear() + } + function waitForConnected(timeoutMs = requestTimeoutMs): Promise { if (state === 'connected') { return Promise.resolve() @@ -268,6 +295,7 @@ export function connectMobileRelayRpcSession(args: { return } closed = true + stopControlProbeTimer() failure = error link.close() rejectPending(error) diff --git a/mobile/src/transport/rpc-client-context-contract.ts b/mobile/src/transport/rpc-client-context-contract.ts new file mode 100644 index 00000000000..af39285445a --- /dev/null +++ b/mobile/src/transport/rpc-client-context-contract.ts @@ -0,0 +1,18 @@ +import type { RpcClient } from './rpc-client' +import type { MobileConnectionPath } from './stable-logical-rpc-client' +import type { ConnectionState, HostProfile } from './types' + +export type RpcClientContextValue = { + acquire: (hostId: string, host?: HostProfile) => RpcClient | null + release: (hostId: string) => void + forceReconnect: (hostId: string) => Promise + closeHost: (hostId: string) => void + getState: (hostId: string) => ConnectionState + getReconnectAttempt: (hostId: string) => number + getLastConnectedAt: (hostId: string) => number | null + getActivePath: (hostId: string) => MobileConnectionPath + subscribeHostState: (hostId: string, listener: (state: ConnectionState) => void) => () => void + getAllClients: () => Array<{ hostId: string; client: RpcClient }> + subscribeAllHosts: (listener: () => void) => () => void + primeHosts: (hosts: HostProfile[]) => void +} diff --git a/mobile/src/transport/rpc-client-request-deadline.test.ts b/mobile/src/transport/rpc-client-request-deadline.test.ts index 227d4ed32af..08dbe7dec89 100644 --- a/mobile/src/transport/rpc-client-request-deadline.test.ts +++ b/mobile/src/transport/rpc-client-request-deadline.test.ts @@ -203,6 +203,28 @@ describe('mobile rpc-client request deadline', () => { await request.catch(() => undefined) }) + it('counts a late timed-out reply as control-plane liveness', async () => { + const client = connect('ws://desktop.invalid', 'token', 'server-key') + const socket = mockSockets[0]! + socket.open() + const request = client.sendRequest('browser.screenshot', {}, { timeoutMs: 100 }) + const outcome = track(request) + await vi.advanceTimersByTimeAsync(0) + const timedOutRequest = sentRequest(socket, 'browser.screenshot') + + await vi.advanceTimersByTimeAsync(100) + expect(outcome.read()).toBe('Request timed out: browser.screenshot') + expect(sentRequest(socket, 'status.get')).toBeDefined() + socket.receive(`encrypted:${JSON.stringify({ id: timedOutRequest.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]! diff --git a/mobile/src/transport/rpc-client.ts b/mobile/src/transport/rpc-client.ts index 1c3aca20455..57a60ccb990 100644 --- a/mobile/src/transport/rpc-client.ts +++ b/mobile/src/transport/rpc-client.ts @@ -36,6 +36,10 @@ import { import { markRpcDeliveryUnknown } from './rpc-delivery-ambiguity' import { openRpcRequestBudget, resolvePostConnectRequestTimeout } from './rpc-request-budget' import { isRpcResponse } from './rpc-response-shape' +import { + isStreamingSubscriptionReadyResult, + isTerminalSubscribedResult +} from './rpc-stream-response-shape' import { websocketPayloadToUint8 } from './websocket-payload-bytes' type PendingRequest = { @@ -186,6 +190,7 @@ export function connect( const serverPublicKey = publicKeyFromBase64(serverPublicKeyB64) const pending = new Map() + const timedOutControlRequestIds = new Set() const streamListeners = new Map() const terminalStreamListeners = new Map() const terminalStreamIdsByRequest = new Map>() @@ -502,7 +507,7 @@ export function connect( if (!isRpcResponse(response)) { return } - if (pending.has(response.id)) { + if (pending.has(response.id) || timedOutControlRequestIds.delete(response.id)) { controlResponseSequence++ } @@ -658,6 +663,7 @@ export function connect( handshakeTimer = null } stopActivityProbe() + finishActivityProbe() if (intentionallyClosed) { console.log('[net] handleSocketClosed — intentional close') setState('disconnected') @@ -779,9 +785,10 @@ export function connect( let timedOut = false const timeout = setTimeout(() => { timedOut = true - activityProbeInFlight = false pending.delete(id) - if (controlResponseSequence > probeControlResponseSequence) { + const controlResponded = controlResponseSequence > probeControlResponseSequence + finishActivityProbe() + if (controlResponded) { return } console.log('[net] activity-probe TIMEOUT — forcing reconnect', { state }) @@ -792,24 +799,29 @@ export function connect( if (timedOut) { return } - activityProbeInFlight = false clearTimeout(timeout) + finishActivityProbe() }, reject: () => { if (timedOut) { return } - activityProbeInFlight = false clearTimeout(timeout) + finishActivityProbe() } }) if (!sendEncrypted({ id, deviceToken, method: 'status.get' })) { - activityProbeInFlight = false clearTimeout(timeout) pending.delete(id) + finishActivityProbe() } } + function finishActivityProbe(): void { + activityProbeInFlight = false + timedOutControlRequestIds.clear() + } + function forceSocketReconnect(socket: WebSocket | null): void { if (!socket || socket !== ws) { return @@ -1037,6 +1049,9 @@ 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) { + if (requestWs === ws) { + timedOutControlRequestIds.add(id) + } runActivityProbe(requestWs) } }, timeoutMs) @@ -1191,6 +1206,7 @@ export function connect( handshakeTimer = null } stopActivityProbe() + finishActivityProbe() if (ws) { ws.close() ws = null @@ -1202,25 +1218,3 @@ export function connect( } } } - -function isTerminalSubscribedResult( - value: unknown -): value is { type: 'subscribed'; streamId: number } { - return ( - !!value && - typeof value === 'object' && - (value as { type?: unknown }).type === 'subscribed' && - typeof (value as { streamId?: unknown }).streamId === 'number' - ) -} - -function isStreamingSubscriptionReadyResult( - value: unknown -): value is { type: 'ready'; subscriptionId: string } { - return ( - !!value && - typeof value === 'object' && - (value as { type?: unknown }).type === 'ready' && - typeof (value as { subscriptionId?: unknown }).subscriptionId === 'string' - ) -} diff --git a/mobile/src/transport/rpc-stream-response-shape.ts b/mobile/src/transport/rpc-stream-response-shape.ts new file mode 100644 index 00000000000..cf09acdb6ca --- /dev/null +++ b/mobile/src/transport/rpc-stream-response-shape.ts @@ -0,0 +1,21 @@ +export function isTerminalSubscribedResult( + value: unknown +): value is { type: 'subscribed'; streamId: number } { + return ( + !!value && + typeof value === 'object' && + (value as { type?: unknown }).type === 'subscribed' && + typeof (value as { streamId?: unknown }).streamId === 'number' + ) +} + +export function isStreamingSubscriptionReadyResult( + value: unknown +): value is { type: 'ready'; subscriptionId: string } { + return ( + !!value && + typeof value === 'object' && + (value as { type?: unknown }).type === 'ready' && + typeof (value as { subscriptionId?: unknown }).subscriptionId === 'string' + ) +}