diff --git a/mobile/app/h/[hostId]/index.tsx b/mobile/app/h/[hostId]/index.tsx index 461ea90b746..6ddaf06f2db 100644 --- a/mobile/app/h/[hostId]/index.tsx +++ b/mobile/app/h/[hostId]/index.tsx @@ -42,7 +42,8 @@ import { useWorktreeResync } from '../../../src/transport/use-worktree-resync' import { startHostWorktreeRefresh } from '../../../src/worktree/host-worktree-refresh' import { useLastConnectedAt, - useReconnectAttempt + useReconnectAttempt, + useRpcUnresponsiveSince } from '../../../src/transport/client-context-connection-metrics' import { classifyConnection, @@ -141,6 +142,7 @@ export function HostScreen({ const { client, state: connState } = useHostClient(hostId) const reconnectAttempts = useReconnectAttempt(hostId) const lastConnectedAt = useLastConnectedAt(hostId) + const rpcUnresponsiveSince = useRpcUnresponsiveSince(hostId) const clientRef = useRef(null) const fetchWorktreesInFlightRef = useRef(false) // Why: useRef, not useMemo — React may discard memoized values, which would silently @@ -796,7 +798,8 @@ export function HostScreen({ const headerVerdict = classifyConnection({ state: connState, reconnectAttempts, - lastConnectedAt + lastConnectedAt, + rpcUnresponsiveSince }) return ( <> @@ -806,27 +809,26 @@ export function HostScreen({ {hostName || 'Host'} - {connState !== 'connected' && - (() => { - // Why: auth-failed has its own banner, so suppress the Reconnect button for that verdict. - const verdict = headerVerdict - const isError = isErrorVerdict(verdict) - const showReconnectButton = isError && hostId && verdict.kind !== 'auth-failed' - if (!showReconnectButton) { - return null - } - return ( - - void forceReconnectHost(hostId!).catch(showForceReconnectError) - } - hitSlop={8} - > - Reconnect - - ) - })()} + {(() => { + // Why: auth-failed has its own banner, so suppress the Reconnect button for that verdict. + const verdict = headerVerdict + const isError = isErrorVerdict(verdict) + const showReconnectButton = isError && hostId && verdict.kind !== 'auth-failed' + if (!showReconnectButton) { + return null + } + return ( + + void forceReconnectHost(hostId!).catch(showForceReconnectError) + } + hitSlop={8} + > + Reconnect + + ) + })()} ) })()} diff --git a/mobile/app/h/[hostId]/session/[worktreeId].tsx b/mobile/app/h/[hostId]/session/[worktreeId].tsx index 52547f5193d..f5c0eb0f127 100644 --- a/mobile/app/h/[hostId]/session/[worktreeId].tsx +++ b/mobile/app/h/[hostId]/session/[worktreeId].tsx @@ -62,7 +62,8 @@ import { useHostClient, useForceReconnect } from '../../../../src/transport/clie import { showForceReconnectError } from '../../../../src/transport/force-reconnect-feedback' import { useLastConnectedAt, - useReconnectAttempt + useReconnectAttempt, + useRpcUnresponsiveSince } from '../../../../src/transport/client-context-connection-metrics' import { classifyConnection, @@ -853,6 +854,7 @@ export default function SessionScreen() { const { client, state: connState } = useHostClient(hostId) const reconnectAttempts = useReconnectAttempt(hostId) const lastConnectedAt = useLastConnectedAt(hostId) + const rpcUnresponsiveSince = useRpcUnresponsiveSince(hostId) const forceReconnectHost = useForceReconnect() const worktreeName = useLiveWorktreeName({ client, @@ -4249,21 +4251,21 @@ export default function SessionScreen() { state: connState, reconnectAttempts, lastConnectedAt, + rpcUnresponsiveSince, endpoint: hostEndpoint }) const showConnectionRetry = connectionVerdict.kind === 'warning' || connectionVerdict.kind === 'unreachable' - const terminalSummary = - connState === 'connected' + const terminalSummary = showConnectionRetry + ? `${verdictDisplayLabel(connectionVerdict)} — tap to retry` + : connState === 'connected' ? showLoadingState ? 'Loading tabs' : visibleTabs.length === 1 ? '1 tab' : `${visibleTabs.length} tabs` - : showConnectionRetry - ? `${verdictDisplayLabel(connectionVerdict)} — tap to retry` - : MOBILE_SESSION_STATUS_LABELS[connState] + : MOBILE_SESSION_STATUS_LABELS[connState] // Why: iOS keyboard height includes the home-indicator inset; Android IME height does not. const keyboardLift = diff --git a/mobile/app/h/[hostId]/tasks.tsx b/mobile/app/h/[hostId]/tasks.tsx index c41b165cbf9..55f6121f6ba 100644 --- a/mobile/app/h/[hostId]/tasks.tsx +++ b/mobile/app/h/[hostId]/tasks.tsx @@ -35,10 +35,7 @@ import { import type { RpcClient } from '../../../src/transport/rpc-client' import type { RpcSuccess } from '../../../src/transport/types' import { useHostClient } from '../../../src/transport/client-context' -import { - useLastConnectedAt, - useReconnectAttempt -} from '../../../src/transport/client-context-connection-metrics' +import { useConnectionHealthInputs } from '../../../src/transport/client-context-connection-metrics' import { classifyConnection } from '../../../src/transport/connection-health' import { StatusDot } from '../../../src/components/StatusDot' import { ActionSheetModal } from '../../../src/components/ActionSheetModal' @@ -2079,8 +2076,7 @@ export default function MobileTasksScreen() { const router = useRouter() const insets = useSafeAreaInsets() const { client, state: connState } = useHostClient(hostId) - const reconnectAttempts = useReconnectAttempt(hostId) - const lastConnectedAt = useLastConnectedAt(hostId) + const connectionHealth = useConnectionHealthInputs(hostId) const clientRef = useRef(null) const reposRef = useRef([]) const loadGenerationRef = useRef(0) @@ -8655,8 +8651,7 @@ export default function MobileTasksScreen() { const headerVerdict = classifyConnection({ state: connState, - reconnectAttempts, - lastConnectedAt + ...connectionHealth }) const emptyLabel = connState !== 'connected' diff --git a/mobile/app/index.tsx b/mobile/app/index.tsx index 1eba85f4279..78f8516a04d 100644 --- a/mobile/app/index.tsx +++ b/mobile/app/index.tsx @@ -30,6 +30,7 @@ import { usePrimeHosts } from '../src/transport/client-context' import { showForceReconnectError } from '../src/transport/force-reconnect-feedback' +import { useRpcUnresponsiveByHost } from '../src/transport/client-context-connection-metrics' import { classifyConnection } from '../src/transport/connection-health' import { subscribeToDesktopNotifications } from '../src/notifications/mobile-notifications' import { @@ -312,6 +313,7 @@ export default function HomeScreen() { // Why: shared clients from the per-host store, not N independent WebSockets. See docs/mobile-shared-client-per-host.md. const hostIds = useMemo(() => hosts.map((h) => h.id), [hosts]) + const hostRpcUnresponsive = useRpcUnresponsiveByHost(hostIds) const allClients = useAllHostClients(hostIds) const hostPaths = useMemo( () => Object.fromEntries(allClients.map(({ hostId, path }) => [hostId, path])), @@ -781,6 +783,7 @@ export default function HomeScreen() { state, reconnectAttempts: attempts, lastConnectedAt, + rpcUnresponsiveSince: hostRpcUnresponsive[item.id] ?? null, endpoint: item.endpoint }) return ( diff --git a/mobile/src/transport/client-context-actions.ts b/mobile/src/transport/client-context-actions.ts index 62de79fbe7f..86900550d85 100644 --- a/mobile/src/transport/client-context-actions.ts +++ b/mobile/src/transport/client-context-actions.ts @@ -1,4 +1,4 @@ -import { useRpcClientContext } from './client-context' +import { useRpcClientContext } from './rpc-client-react-context' export const useCloseHost = () => useRpcClientContext().closeHost export const useForceReconnect = () => useRpcClientContext().forceReconnect diff --git a/mobile/src/transport/client-context-all-host-clients.ts b/mobile/src/transport/client-context-all-host-clients.ts new file mode 100644 index 00000000000..2bde1bf7b64 --- /dev/null +++ b/mobile/src/transport/client-context-all-host-clients.ts @@ -0,0 +1,55 @@ +import { useEffect, useMemo, useState } from 'react' +import { useRpcClientContext } from './rpc-client-react-context' +import type { RpcClient } from './rpc-client' +import type { MobileConnectionPath } from './stable-logical-rpc-client' +import type { ConnectionState } from './types' + +export function useAllHostClients(hostIds: string[]) { + const context = useRpcClientContext() + const key = useMemo(() => [...hostIds].sort().join(','), [hostIds]) + const [tick, setTick] = useState(0) + + useEffect(() => { + if (hostIds.length === 0) { + return + } + for (const id of hostIds) { + context.acquire(id) + } + const unsubscribes = hostIds.map((id) => + context.subscribeHostState(id, () => setTick((value) => value + 1)) + ) + unsubscribes.push(context.subscribeAllHosts(() => setTick((value) => value + 1))) + return () => { + for (const unsubscribe of unsubscribes) { + unsubscribe() + } + for (const id of hostIds) { + context.release(id) + } + } + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [key]) + + return useMemo(() => { + const clients: Array<{ + hostId: string + client: RpcClient + state: ConnectionState + path: MobileConnectionPath + }> = [] + for (const id of hostIds) { + const found = context.getAllClients().find((entry) => entry.hostId === id) + if (found) { + clients.push({ + hostId: id, + client: found.client, + state: context.getState(id), + path: context.getActivePath(id) + }) + } + } + return clients + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [key, tick]) +} diff --git a/mobile/src/transport/client-context-connection-metrics.ts b/mobile/src/transport/client-context-connection-metrics.ts index e41d2241d75..3dcf3f0543e 100644 --- a/mobile/src/transport/client-context-connection-metrics.ts +++ b/mobile/src/transport/client-context-connection-metrics.ts @@ -1,7 +1,9 @@ import { useEffect, useState } from 'react' -import { useRpcClientContext } from './client-context' +import { useRpcClientContext } from './rpc-client-react-context' import type { RpcClientContextValue } from './rpc-client-context-contract' +const RPC_UNRESPONSIVE_POLL_MS = 1_000 + export function useReconnectAttempt(hostId: string | undefined): number { return useHostMetric(hostId, (context, id) => context.getReconnectAttempt(id), 0) } @@ -10,10 +12,40 @@ export function useLastConnectedAt(hostId: string | undefined): number | null { return useHostMetric(hostId, (context, id) => context.getLastConnectedAt(id), null) } +export function useRpcUnresponsiveSince(hostId: string | undefined): number | null { + return useHostMetric(hostId, (context, id) => context.getRpcUnresponsiveSince(id), null, { + pollMs: RPC_UNRESPONSIVE_POLL_MS + }) +} + +export function useConnectionHealthInputs(hostId: string | undefined) { + return { + reconnectAttempts: useReconnectAttempt(hostId), + lastConnectedAt: useLastConnectedAt(hostId), + rpcUnresponsiveSince: useRpcUnresponsiveSince(hostId) + } +} + +export function useRpcUnresponsiveByHost(hostIds: string[]): Record { + const context = useRpcClientContext() + const [values, setValues] = useState>({}) + const hostKey = hostIds.join('\u0000') + useEffect(() => { + const read = () => { + setValues(Object.fromEntries(hostIds.map((id) => [id, context.getRpcUnresponsiveSince(id)]))) + } + read() + const interval = setInterval(read, RPC_UNRESPONSIVE_POLL_MS) + return () => clearInterval(interval) + }, [context, hostIds, hostKey]) + return values +} + function useHostMetric( hostId: string | undefined, read: (context: RpcClientContextValue, hostId: string) => T, - fallback: T + fallback: T, + options: { pollMs?: number } = {} ): T { const context = useRpcClientContext() const [, force] = useState(0) @@ -21,7 +53,16 @@ function useHostMetric( if (!hostId) { return } - return context.subscribeHostState(hostId, () => force((count) => count + 1)) - }, [context, hostId]) + const unsubscribe = context.subscribeHostState(hostId, () => force((count) => count + 1)) + const interval = options.pollMs + ? setInterval(() => force((count) => count + 1), options.pollMs) + : null + return () => { + unsubscribe() + if (interval) { + clearInterval(interval) + } + } + }, [context, hostId, options.pollMs]) return hostId ? read(context, hostId) : fallback } diff --git a/mobile/src/transport/client-context.tsx b/mobile/src/transport/client-context.tsx index ff368c25d7e..133a72dea13 100644 --- a/mobile/src/transport/client-context.tsx +++ b/mobile/src/transport/client-context.tsx @@ -1,13 +1,4 @@ -import { - createContext, - useCallback, - useContext, - useEffect, - useMemo, - useRef, - useState, - type ReactNode -} from 'react' +import { useCallback, useEffect, useMemo, useRef, useState, type ReactNode } from 'react' import type { RpcClient } from './rpc-client' import { connectionLogStore } from './connection-log-buffer' import { HostForceReconnectCoordinator, type HostReconnectEntry } from './host-force-reconnect' @@ -22,11 +13,12 @@ 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' +import { RpcApplicationResponsiveness } from './rpc-application-responsiveness' +import { RpcClientContextBoundary, useRpcClientContext } from './rpc-client-react-context' +export { useRpcClientContext } from './rpc-client-react-context' type StoreEntry = HostReconnectEntry & { state: ConnectionState } -const Ctx = createContext(null) - export function RpcClientProvider({ children }: { children: ReactNode }) { const storeRef = useRef>(new Map()) const stateListenersRef = useRef void>>>(new Map()) @@ -36,6 +28,16 @@ export function RpcClientProvider({ children }: { children: ReactNode }) { const forceReconnectCoordinatorRef = useRef(new HostForceReconnectCoordinator()) const primedHostsRef = useRef(new HostReconnectProfileCache()) + const responsivenessRef = useRef(new Map()) + + function getHostResponsiveness(hostId: string): RpcApplicationResponsiveness { + let responsiveness = responsivenessRef.current.get(hostId) + if (!responsiveness) { + responsiveness = new RpcApplicationResponsiveness() + responsivenessRef.current.set(hostId, responsiveness) + } + return responsiveness + } function notifyHostState(hostId: string, state: ConnectionState) { const set = stateListenersRef.current.get(hostId) @@ -57,6 +59,7 @@ export function RpcClientProvider({ children }: { children: ReactNode }) { pendingOpensRef.current.cancel(hostId) forceReconnectCoordinatorRef.current.cancel(hostId) primedHostsRef.current.delete(hostId) + responsivenessRef.current.delete(hostId) const entry = storeRef.current.get(hostId) entry?.unsubState() storeRef.current.delete(hostId) @@ -121,7 +124,8 @@ export function RpcClientProvider({ children }: { children: ReactNode }) { client = openHostLogicalClient( profile.host, onLog, - primedHostsRef.current.publisher(hostId, profile.version, getHostListLoadRevision) + primedHostsRef.current.publisher(hostId, profile.version, getHostListLoadRevision), + getHostResponsiveness(hostId) ) } catch { // Why: openHostLogicalClient can throw synchronously (bad public key / invalid URL); notify so the UI leaves 'connecting'. @@ -228,6 +232,10 @@ export function RpcClientProvider({ children }: { children: ReactNode }) { return storeRef.current.get(hostId)?.client.getLastConnectedAt() ?? null }, []) + const getRpcUnresponsiveSince = useCallback((hostId: string): number | null => { + return responsivenessRef.current.get(hostId)?.getUnresponsiveSince() ?? null + }, []) + const getActivePath = useCallback((hostId: string): MobileConnectionPath => { return clientActivePath(storeRef.current.get(hostId)?.client) }, []) @@ -298,6 +306,7 @@ export function RpcClientProvider({ children }: { children: ReactNode }) { getState, getReconnectAttempt, getLastConnectedAt, + getRpcUnresponsiveSince, getActivePath, subscribeHostState, getAllClients, @@ -312,6 +321,7 @@ export function RpcClientProvider({ children }: { children: ReactNode }) { getState, getReconnectAttempt, getLastConnectedAt, + getRpcUnresponsiveSince, getActivePath, subscribeHostState, getAllClients, @@ -320,15 +330,7 @@ export function RpcClientProvider({ children }: { children: ReactNode }) { ] ) - return {children} -} - -export function useRpcClientContext(): RpcClientContextValue { - const ctx = useContext(Ctx) - if (!ctx) { - throw new Error('useHostClient must be used inside ') - } - return ctx + return {children} } export function useHostClient(hostId: string | undefined): { @@ -391,56 +393,6 @@ export function useHostClient(hostId: string | undefined): { return { client: bound ? clientRef.current : null, state: boundState } } -export function useAllHostClients(hostIds: string[]) { - const ctx = useRpcClientContext() - const key = useMemo(() => [...hostIds].sort().join(','), [hostIds]) - const [tick, setTick] = useState(0) - - useEffect(() => { - if (hostIds.length === 0) { - return - } - for (const id of hostIds) { - ctx.acquire(id) - } - const unsubs: Array<() => void> = [] - for (const id of hostIds) { - unsubs.push(ctx.subscribeHostState(id, () => setTick((n) => n + 1))) - } - unsubs.push(ctx.subscribeAllHosts(() => setTick((n) => n + 1))) - return () => { - for (const u of unsubs) { - u() - } - for (const id of hostIds) { - ctx.release(id) - } - } - // eslint-disable-next-line react-hooks/exhaustive-deps - }, [key]) - - return useMemo(() => { - const out: Array<{ - hostId: string - client: RpcClient - state: ConnectionState - path: MobileConnectionPath - }> = [] - for (const id of hostIds) { - const all = ctx.getAllClients().find((entry) => entry.hostId === id) - if (all) { - out.push({ - hostId: id, - client: all.client, - state: ctx.getState(id), - path: ctx.getActivePath(id) - }) - } - } - return out - // eslint-disable-next-line react-hooks/exhaustive-deps - }, [key, tick]) -} - // Why: host-store's removeHost() must close the live client but has no React-side handle; this hook bridges to it. export { useCloseHost, useForceReconnect, usePrimeHosts } from './client-context-actions' +export { useAllHostClients } from './client-context-all-host-clients' diff --git a/mobile/src/transport/connection-health.ts b/mobile/src/transport/connection-health.ts index 6774b111ff5..7449506bfc3 100644 --- a/mobile/src/transport/connection-health.ts +++ b/mobile/src/transport/connection-health.ts @@ -46,6 +46,7 @@ export function classifyConnection(args: { state: ConnectionState reconnectAttempts: number lastConnectedAt: number | null + rpcUnresponsiveSince?: number | null // Optional pinned host endpoint — enables the Tailscale hint on // warning/unreachable verdicts. Callers without it get plain labels. endpoint?: string | null @@ -62,6 +63,9 @@ export function classifyConnection(args: { } if (state === 'connected') { + if (args.rpcUnresponsiveSince != null) { + return { kind: 'warning', label: 'Connected, not responding', hint } + } return { kind: 'normal', label: 'Connected' } } diff --git a/mobile/src/transport/existing-host-relay-routing-lifecycle.test.ts b/mobile/src/transport/existing-host-relay-routing-lifecycle.test.ts index 4db25072022..9bfe3f06da9 100644 --- a/mobile/src/transport/existing-host-relay-routing-lifecycle.test.ts +++ b/mobile/src/transport/existing-host-relay-routing-lifecycle.test.ts @@ -10,6 +10,9 @@ vi.mock('./host-store', () => hostStoreMock) vi.mock('./host-device-token-store', () => tokenStoreMock) vi.mock('./mobile-relay-host-overlay-store', () => overlayStoreMock) vi.mock('./host-list-load-sharing', () => ({ dropSharedHostListLoad: vi.fn() })) +vi.mock('./mobile-relay-credential-bundle', () => ({ + deleteMobileRelayCredentialBundleIfCurrent: vi.fn(async () => false) +})) import { MobileRelayUpgradeLifecycleRetiredError, @@ -136,6 +139,37 @@ describe('existing host relay endpoint lifecycle publication', () => { expect(writeBundle).not.toHaveBeenCalled() }) + it('conditionally cleans a bundle written after its endpoint lifecycle retires', async () => { + let releaseWrite: (() => void) | null = null + let markWriteStarted: (() => void) | null = null + const writeStarted = new Promise((resolve) => { + markWriteStarted = resolve + }) + const staleBundle = credentialBundle(1) + const cleanupBundle = vi.fn(async () => false) + const lifecycle = beginHostEndpointPublicationLifecycle(HOST.id) + const staleWrite = writeExistingHostRelayCredentialBundle( + HOST, + staleBundle, + async () => { + markWriteStarted?.() + await new Promise((resolve) => { + releaseWrite = resolve + }) + }, + lifecycle, + cleanupBundle + ) + await writeStarted + + beginHostEndpointPublicationLifecycle(HOST.id) + releaseWrite?.() + + await expect(staleWrite).rejects.toBeInstanceOf(MobileRelayUpgradeLifecycleRetiredError) + await vi.waitFor(() => expect(cleanupBundle).toHaveBeenCalledWith(staleBundle)) + expect(await cleanupBundle.mock.results[0]!.value).toBe(false) + }) + it('rechecks routing publication after its credential write', async () => { let releaseCredentialWrite: (() => void) | null = null let markCredentialWriteStarted: (() => void) | null = null diff --git a/mobile/src/transport/existing-host-relay-routing.ts b/mobile/src/transport/existing-host-relay-routing.ts index 1beb3e9d09e..0ae4a3afc76 100644 --- a/mobile/src/transport/existing-host-relay-routing.ts +++ b/mobile/src/transport/existing-host-relay-routing.ts @@ -10,7 +10,10 @@ import { } from './host-profile-publication' import * as hostListLoads from './host-list-load-sharing' import { saveMobileRelayHostOverlay } from './mobile-relay-host-overlay-store' -import type { MobileRelayCredentialBundle } from './mobile-relay-credential-bundle' +import { + deleteMobileRelayCredentialBundleIfCurrent, + type MobileRelayCredentialBundle +} from './mobile-relay-credential-bundle' export class MobileRelayUpgradeHostRemovedError extends Error {} export class MobileRelayUpgradeHostSupersededError extends Error {} @@ -20,10 +23,10 @@ export async function saveExistingHostRelayRouting( host: HostProfile, beforePublish?: () => Promise, endpointLifecycle = getHostEndpointPublicationLifecycle(host.id) -): Promise { +): Promise { const validated = HostProfileSchema.parse(host) await requireCurrentHostCredential(validated) - await serializeHostProfilePublication(validated.id, async () => { + return serializeHostProfilePublication(validated.id, async () => { requireCurrentEndpointGeneration(validated.id, endpointLifecycle) const existing = await requireCurrentHostMetadata(validated) requireCurrentPublication(validated, endpointLifecycle) @@ -45,7 +48,7 @@ export async function saveExistingHostRelayRouting( relayHostId, relay }) - hostListLoads.dropSharedHostListLoad() + return hostListLoads.dropSharedHostListLoad() }) } @@ -53,7 +56,8 @@ export async function writeExistingHostRelayCredentialBundle( host: HostProfile, bundle: MobileRelayCredentialBundle, writeBundle: (bundle: MobileRelayCredentialBundle) => Promise, - endpointLifecycle = getHostEndpointPublicationLifecycle(host.id) + endpointLifecycle = getHostEndpointPublicationLifecycle(host.id), + cleanupBundle = deleteMobileRelayCredentialBundleIfCurrent ): Promise { const validated = HostProfileSchema.parse(host) await requireCurrentHostCredential(validated) @@ -65,6 +69,12 @@ export async function writeExistingHostRelayCredentialBundle( } requireCurrentPublication(validated, endpointLifecycle) await writeBundle(bundle) + try { + requireCurrentPublication(validated, endpointLifecycle) + } catch (error) { + void cleanupBundle(bundle).catch(() => {}) + throw error + } }) } diff --git a/mobile/src/transport/force-reconnect-rpc-health.test.ts b/mobile/src/transport/force-reconnect-rpc-health.test.ts index 3f0add2a9e9..a9043a937cd 100644 --- a/mobile/src/transport/force-reconnect-rpc-health.test.ts +++ b/mobile/src/transport/force-reconnect-rpc-health.test.ts @@ -4,6 +4,7 @@ import type { ConnectionState, RpcResponse } from './types' import { verifyForceReconnectRpcHealth } from './force-reconnect-rpc-health' import { markRpcDeliveryUnknown } from './rpc-delivery-ambiguity' import { LogicalClientCutoverError } from './stable-logical-rpc-client' +import { RpcApplicationResponsiveness } from './rpc-application-responsiveness' describe('Force Reconnect RPC health', () => { afterEach(() => vi.useRealTimers()) @@ -66,6 +67,24 @@ describe('Force Reconnect RPC health', () => { expect(sendRequest).toHaveBeenCalledTimes(2) }) + it('rejects a successful probe until a real application response clears the shared latch', async () => { + const responsiveness = new RpcApplicationResponsiveness() + responsiveness.recordTimeout('browser.screenshot', 123) + const sendRequest = vi.fn(async () => ({ id: 'rpc-1', ok: true, result: {} }) as RpcResponse) + const client = { + sendRequest, + getState: () => 'connected' as const, + getRpcUnresponsiveSince: () => responsiveness.getUnresponsiveSince() + } as unknown as RpcClient + + await expect(verifyForceReconnectRpcHealth(client)).rejects.toThrow( + 'Application RPC channel is still not responding' + ) + responsiveness.recordResponse('worktree.list') + await expect(verifyForceReconnectRpcHealth(client)).resolves.toBeUndefined() + expect(sendRequest).toHaveBeenCalledTimes(2) + }) + it('fails when authorization retries reach auth-failed', async () => { const sendRequest = vi.fn(async () => { throw new Error('Unauthorized — pairing may be revoked') diff --git a/mobile/src/transport/force-reconnect-rpc-health.ts b/mobile/src/transport/force-reconnect-rpc-health.ts index 29262852845..fe09257cac4 100644 --- a/mobile/src/transport/force-reconnect-rpc-health.ts +++ b/mobile/src/transport/force-reconnect-rpc-health.ts @@ -20,6 +20,9 @@ export async function verifyForceReconnectRpcHealth( budgetSpansConnect: true, strictDeadline: true }) + if (client.getRpcUnresponsiveSince?.() != null) { + throw new Error('Application RPC channel is still not responding') + } return } catch (error) { const state = client.getState() diff --git a/mobile/src/transport/host-credential-delete.ts b/mobile/src/transport/host-credential-delete.ts new file mode 100644 index 00000000000..9158bc39496 --- /dev/null +++ b/mobile/src/transport/host-credential-delete.ts @@ -0,0 +1,19 @@ +import { deleteHostDeviceToken } from './host-device-token-store' +import { deleteMobileRelayCredentialBundle } from './mobile-relay-credential-bundle' +import { deleteMobileRelayDirectUpgradeJournal } from './mobile-relay-direct-upgrade-journal' + +export async function deleteHostCredentials(hostId: string): Promise { + await deleteHostDeviceToken(hostId) + await deleteMobileRelayCredentialBundle(hostId) + await deleteMobileRelayDirectUpgradeJournal(hostId) +} + +export function createRemovedHostCredentialDelete( + hostExists: (hostId: string) => Promise +): (hostId: string) => Promise { + return async (hostId) => { + if (!(await hostExists(hostId))) { + await deleteHostCredentials(hostId) + } + } +} diff --git a/mobile/src/transport/host-list-load-sharing.ts b/mobile/src/transport/host-list-load-sharing.ts index 016020dcc49..117daf6ce3c 100644 --- a/mobile/src/transport/host-list-load-sharing.ts +++ b/mobile/src/transport/host-list-load-sharing.ts @@ -25,9 +25,10 @@ export function shareHostListLoad(load: () => Promise): Promise void = () => {} + onHostUpdated: (host: HostProfile, sourceRevision?: number) => void = () => {}, + applicationResponsiveness = new RpcApplicationResponsiveness() ): RpcClient { // Why: the stable facade owns app-visible RPC/subscription state while the // direct socket remains a replaceable first physical generation. const logical = createStableLogicalRpcClient( - connect(host.endpoint, host.deviceToken, host.publicKeyB64, { onLog }), + connect(host.endpoint, host.deviceToken, host.publicKeyB64, { + onLog, + applicationResponsiveness + }), directPathForEndpoint(host, host.endpoint) ) if (Platform.OS === 'web') { return logical } - const endpointLifecycle = startMobileEndpointLifecycle(logical, host, onLog, onHostUpdated) + const endpointLifecycle = startMobileEndpointLifecycle( + logical, + host, + onLog, + onHostUpdated, + applicationResponsiveness + ) endpointLifecycle.setForeground(AppState.currentState === 'active') const appStateSubscription = AppState.addEventListener('change', (state) => { endpointLifecycle.setForeground(state === 'active') diff --git a/mobile/src/transport/host-profile-publication.ts b/mobile/src/transport/host-profile-publication.ts index 559dcf7ce40..239de3f8bf5 100644 --- a/mobile/src/transport/host-profile-publication.ts +++ b/mobile/src/transport/host-profile-publication.ts @@ -29,10 +29,13 @@ export function recordDurableHostIdentity( }) } -export function retireHostProfilePublication(hostId: string): void { +export function retireHostProfilePublication(hostId: string): Promise | null { revisionByHost.set(hostId, getHostProfilePublicationRevision(hostId) + 1) endpointGenerationByHost.set(hostId, (endpointGenerationByHost.get(hostId) ?? 0) + 1) publishedIdentityByHost.delete(hostId) + const pending = pendingByHost.get(hostId) ?? null + pendingByHost.delete(hostId) + return pending } export function beginHostEndpointPublicationLifecycle( @@ -77,13 +80,22 @@ export function publishHostProfileTransaction( saveHost: (host: HostProfile) => Promise ): Promise { return serializeHostProfilePublication(host.id, async () => { + const publicationRevision = getHostProfilePublicationRevision(host.id) await beforeHostSave?.() + requirePublicationRevision(host.id, publicationRevision) await saveHost(host) - revisionByHost.set(host.id, getHostProfilePublicationRevision(host.id) + 1) + requirePublicationRevision(host.id, publicationRevision) + revisionByHost.set(host.id, publicationRevision + 1) recordDurableHostIdentity(host) }) } +function requirePublicationRevision(hostId: string, expected: number): void { + if (getHostProfilePublicationRevision(hostId) !== expected) { + throw new Error('host profile publication was retired') + } +} + /** Test-only: reset publication state between module-level storage cases. */ export function resetHostProfilePublicationForTests(): void { pendingByHost.clear() diff --git a/mobile/src/transport/host-reconnect-profile-cache.test.ts b/mobile/src/transport/host-reconnect-profile-cache.test.ts index 13fa446d0c5..8d818808c00 100644 --- a/mobile/src/transport/host-reconnect-profile-cache.test.ts +++ b/mobile/src/transport/host-reconnect-profile-cache.test.ts @@ -48,4 +48,36 @@ describe('HostReconnectProfileCache', () => { version: cachedVersion + 1 }) }) + + it('publishes relay routing at the exact revision returned by its save', () => { + const cache = new HostReconnectProfileCache() + let currentRevision = 4 + const initialVersion = cache.prime(HOST, currentRevision) + const publish = cache.publisher(HOST.id, initialVersion, () => currentRevision) + const relayHost = { + ...HOST, + relayHostId: 'AbCdEf0123_-xyZ9', + endpoints: [ + { id: 'direct-primary', kind: 'lan' as const, url: HOST.endpoint }, + { + id: 'relay-primary', + kind: 'relay' as const, + url: 'wss://relay.invalid/v1/connect/host' + } + ], + relay: { + v: 1 as const, + directorUrl: 'https://relay.invalid', + cellUrl: 'https://relay.invalid', + assignmentEpoch: 1, + relayHostId: 'AbCdEf0123_-xyZ9', + e2eeFraming: 2 as const + } + } + + currentRevision = 5 + publish(relayHost, 5) + + expect(cache.get(HOST.id, 5)).toEqual(relayHost) + }) }) diff --git a/mobile/src/transport/host-reconnect-profile-cache.ts b/mobile/src/transport/host-reconnect-profile-cache.ts index ed6f980303b..12e72928a7a 100644 --- a/mobile/src/transport/host-reconnect-profile-cache.ts +++ b/mobile/src/transport/host-reconnect-profile-cache.ts @@ -3,6 +3,7 @@ import type { HostProfile } from './types' type CachedHostProfile = { host: HostProfile version: number + publicationVersion: number sourceRevision: number } @@ -15,15 +16,22 @@ export type HostOpenProfile = { export class HostReconnectProfileCache { private readonly profiles = new Map() private readonly latestVersions = new Map() + private readonly publicationVersions = new Map() prime(host: HostProfile, sourceRevision: number): number { + const previous = this.profiles.get(host.id) const current = this.freshProfile(host.id, sourceRevision) const version = current && reconnectProfileMatches(current.host, host) ? current.version : (this.latestVersions.get(host.id) ?? 0) + 1 this.latestVersions.set(host.id, version) - this.profiles.set(host.id, { host, version, sourceRevision }) + const publicationVersion = + previous && reconnectProfileMatches(previous.host, host) + ? previous.publicationVersion + : (this.publicationVersions.get(host.id) ?? 0) + 1 + this.publicationVersions.set(host.id, publicationVersion) + this.profiles.set(host.id, { host, version, publicationVersion, sourceRevision }) return version } @@ -74,28 +82,39 @@ export class HostReconnectProfileCache { primeFromVersion( host: HostProfile, - sourceVersion: number, + sourcePublicationVersion: number, sourceRevision: number ): number | null { - if (this.version(host.id, sourceRevision) !== sourceVersion) { + if ((this.publicationVersions.get(host.id) ?? 0) !== sourcePublicationVersion) { return null } - return this.prime(host, sourceRevision) + const version = (this.latestVersions.get(host.id) ?? 0) + 1 + const publicationVersion = sourcePublicationVersion + 1 + this.latestVersions.set(host.id, version) + this.publicationVersions.set(host.id, publicationVersion) + this.profiles.set(host.id, { host, version, publicationVersion, sourceRevision }) + return version } publisher( hostId: string, initialVersion: number, getCurrentRevision: () => number - ): (host: HostProfile) => void { - let sourceVersion = initialVersion - return (host) => { - if (host.id !== hostId) { + ): (host: HostProfile, sourceRevision?: number) => void { + const initial = this.profiles.get(hostId) + let sourcePublicationVersion = + initial?.version === initialVersion + ? initial.publicationVersion + : (this.publicationVersions.get(hostId) ?? 0) + return (host, publishedSourceRevision) => { + const currentRevision = getCurrentRevision() + const sourceRevision = publishedSourceRevision ?? currentRevision + if (host.id !== hostId || sourceRevision !== currentRevision) { return } - const nextVersion = this.primeFromVersion(host, sourceVersion, getCurrentRevision()) + const nextVersion = this.primeFromVersion(host, sourcePublicationVersion, sourceRevision) if (nextVersion !== null) { - sourceVersion = nextVersion + sourcePublicationVersion = this.publicationVersions.get(hostId) ?? sourcePublicationVersion } } } @@ -111,6 +130,7 @@ export class HostReconnectProfileCache { delete(hostId: string): void { this.profiles.delete(hostId) + this.publicationVersions.set(hostId, (this.publicationVersions.get(hostId) ?? 0) + 1) } private freshProfile(hostId: string, currentRevision: number): CachedHostProfile | undefined { diff --git a/mobile/src/transport/host-store-retired-publication.test.ts b/mobile/src/transport/host-store-retired-publication.test.ts new file mode 100644 index 00000000000..1b2ddd20d20 --- /dev/null +++ b/mobile/src/transport/host-store-retired-publication.test.ts @@ -0,0 +1,74 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const storage = vi.hoisted(() => ({ getItem: vi.fn(), setItem: vi.fn(), removeItem: vi.fn() })) +const secureStore = vi.hoisted(() => ({ + deleteItemAsync: vi.fn(), + getItemAsync: vi.fn(), + setItemAsync: vi.fn() +})) +const scheduleCleanup = vi.hoisted(() => vi.fn()) + +vi.mock('@react-native-async-storage/async-storage', () => ({ default: storage })) +vi.mock('expo-secure-store', () => ({ + WHEN_UNLOCKED_THIS_DEVICE_ONLY: 'WHEN_UNLOCKED_THIS_DEVICE_ONLY', + ...secureStore +})) +vi.mock('react-native', () => ({ Platform: { OS: 'ios' } })) +vi.mock('./host-credential-cleanup', () => ({ + scheduleHostCredentialCleanup: (...args: unknown[]) => scheduleCleanup(...args), + retryPendingHostCredentialCleanups: vi.fn() +})) + +import { removeHost, resetHostStoreForTests, saveHost } from './host-store' +import { publishHostProfileTransaction } from './host-profile-publication' + +const HOST = { + id: 'host-1', + name: 'Host 1', + endpoint: 'ws://127.0.0.1:1', + publicKeyB64: 'key', + lastConnected: 0 +} +const HOSTS_STORAGE_KEY = 'orca:hosts' + +describe('host removal with a retired publication', () => { + let storedHostsRaw: string + + beforeEach(() => { + vi.clearAllMocks() + resetHostStoreForTests() + storedHostsRaw = JSON.stringify([HOST]) + storage.getItem.mockImplementation(async (key: string) => + key === HOSTS_STORAGE_KEY ? storedHostsRaw : null + ) + storage.setItem.mockImplementation(async (key: string, raw: string) => { + if (key === HOSTS_STORAGE_KEY) { + storedHostsRaw = raw + } + }) + secureStore.getItemAsync.mockResolvedValue('token-1') + scheduleCleanup.mockResolvedValue(undefined) + }) + + it('removes metadata without waiting for a stalled retired publication', async () => { + let markPublicationStarted: (() => void) | null = null + const publicationStarted = new Promise((resolve) => { + markPublicationStarted = resolve + }) + const stalledPublication = publishHostProfileTransaction( + { ...HOST, deviceToken: 'token-1' }, + async () => { + markPublicationStarted?.() + await new Promise(() => {}) + }, + saveHost + ) + void stalledPublication.catch(() => {}) + await publicationStarted + + await expect(removeHost(HOST.id)).resolves.toBeUndefined() + + expect(JSON.parse(storedHostsRaw)).toEqual([]) + expect(scheduleCleanup).toHaveBeenCalledOnce() + }) +}) diff --git a/mobile/src/transport/host-store.ts b/mobile/src/transport/host-store.ts index 6a8e3669ad8..ca1d0bb025c 100644 --- a/mobile/src/transport/host-store.ts +++ b/mobile/src/transport/host-store.ts @@ -19,8 +19,6 @@ import { saveMobileRelayHostOverlay, updateMobileRelayHostOverlayDirectEndpoint } from './mobile-relay-host-overlay-store' -import { deleteMobileRelayCredentialBundle } from './mobile-relay-credential-bundle' -import { deleteMobileRelayDirectUpgradeJournal } from './mobile-relay-direct-upgrade-journal' import { scheduleOrphanedMobileRelayCleanup } from './mobile-relay-orphan-cleanup' import { recordDurableHostIdentity, @@ -28,21 +26,12 @@ import { retireHostProfilePublication, serializeHostProfilePublication } from './host-profile-publication' -import { - deleteHostDeviceToken, - readHostDeviceToken, - writeHostDeviceToken -} from './host-device-token-store' +import { readHostDeviceToken, writeHostDeviceToken } from './host-device-token-store' +import { createRemovedHostCredentialDelete, deleteHostCredentials } from './host-credential-delete' const STORAGE_KEY = 'orca:hosts' type StoredHostIdentity = Pick -async function deleteHostCredentials(hostId: string): Promise { - await deleteHostDeviceToken(hostId) - await deleteMobileRelayCredentialBundle(hostId) - await deleteMobileRelayDirectUpgradeJournal(hostId) -} - // Why: Keychain reads are slow (50-200ms) and loadHosts() runs on every screen mount; cache per-hostId in memory, invalidate on save/remove. const tokenCache = new Map() // Why: serialize RMW of the shared hosts JSON; without a queue concurrent writers drop writes (resurrect a removed host, drop a rename). @@ -139,6 +128,10 @@ export async function loadStoredHostIdentity(hostId: string): Promise + Boolean(await loadStoredHostIdentity(hostId)) +) + export async function resolvePairingHostIdentity( publicKeyB64: string, newHostId: string @@ -196,10 +189,6 @@ function toStored(host: HostProfile): StoredHostProfile { } export async function saveHost(host: HostProfile): Promise { - await persistHost(host) -} - -async function persistHost(host: HostProfile): Promise { const validated = HostProfileSchema.parse(host) const stored = toStored(validated) const duplicateHostIds = new Set() @@ -256,10 +245,10 @@ async function persistHost(host: HostProfile): Promise { } export async function removeHost(hostId: string): Promise { + const retiredPublication = retireHostProfilePublication(hostId) await serializeHostProfilePublication(hostId, async () => { await mutateStoredHosts((hosts) => hosts.filter((h) => h.id !== hostId)) tokenCache.delete(hostId) - retireHostProfilePublication(hostId) try { await removeMobileRelayHostOverlay(hostId) hostListLoads.dropSharedHostListLoad() @@ -268,11 +257,17 @@ export async function removeHost(hostId: string): Promise { } // Why: keychain delete can stall/reject; await only the durable cleanup intent so removeHost can't freeze the UI. try { - await scheduleHostCredentialCleanup(hostId, deleteHostCredentials) + await scheduleHostCredentialCleanup(hostId, deleteRemovedHostCredentials) } catch { // Metadata is already committed; orphan-token recovery is best-effort. } }) + if (retiredPublication) { + // Why: a retired native write may land after the first delete was attempted. + void retiredPublication.then(() => + scheduleHostCredentialCleanup(hostId, deleteRemovedHostCredentials).catch(() => {}) + ) + } } export async function retryPendingHostCredentialCleanup(): Promise<{ diff --git a/mobile/src/transport/mobile-endpoint-lifecycle.ts b/mobile/src/transport/mobile-endpoint-lifecycle.ts index d06c28d9568..2ece7890cdd 100644 --- a/mobile/src/transport/mobile-endpoint-lifecycle.ts +++ b/mobile/src/transport/mobile-endpoint-lifecycle.ts @@ -20,6 +20,7 @@ import { upgradeDirectMobileRelay } from './mobile-relay-direct-upgrade' import { retireMobileRelayDirectUpgradeJournalForRelayHost } from './mobile-relay-direct-upgrade-journal' import { MobileRelayDirectUpgradeController } from './mobile-relay-direct-upgrade-controller' import type { StableLogicalRpcClient } from './stable-logical-rpc-client' +import { RpcApplicationResponsiveness } from './rpc-application-responsiveness' type EndpointLifecycle = { setForeground(foreground: boolean): void @@ -34,23 +35,41 @@ export function startMobileEndpointLifecycle( logical: StableLogicalRpcClient, initialHost: HostProfile, onLog: ConnectionLogSink, - onHostUpdated: (host: HostProfile) => void = () => {} + onHostUpdated: (host: HostProfile, sourceRevision?: number) => void = () => {}, + applicationResponsiveness = new RpcApplicationResponsiveness() ): EndpointLifecycle { let stopped = false let foreground = true let owner: EndpointOwner + let publishedSourceRevision: number | undefined const endpointLifecycle = beginHostEndpointPublicationLifecycle(initialHost.id) const publishHostUpdate = (host: HostProfile): void => { if (!stopped) { - onHostUpdated(host) + onHostUpdated(host, publishedSourceRevision) } } + const saveHost = async (host: HostProfile, beforePublish?: () => Promise) => { + publishedSourceRevision = await saveExistingHostRelayRouting( + host, + beforePublish, + endpointLifecycle + ) + return publishedSourceRevision + } const startSupervisor = async (host: HostProfile): Promise => { if (stopped) { return } - const supervisor = createSupervisor(logical, host, onLog, publishHostUpdate, endpointLifecycle) + const supervisor = createSupervisor( + logical, + host, + onLog, + publishHostUpdate, + endpointLifecycle, + saveHost, + applicationResponsiveness + ) owner.stop() owner = supervisor supervisor.setForeground(foreground) @@ -60,7 +79,15 @@ export function startMobileEndpointLifecycle( if (initialHost.relay) { // Why: relay publication supersedes any committed journal retained by the retired direct owner. void retireMobileRelayDirectUpgradeJournalForRelayHost(initialHost.id).catch(() => {}) - owner = createSupervisor(logical, initialHost, onLog, publishHostUpdate, endpointLifecycle) + owner = createSupervisor( + logical, + initialHost, + onLog, + publishHostUpdate, + endpointLifecycle, + saveHost, + applicationResponsiveness + ) void owner.start() } else { owner = new MobileRelayDirectUpgradeController(logical, initialHost, { @@ -70,8 +97,7 @@ export function startMobileEndpointLifecycle( host, dependencies: { randomBytes: ExpoCrypto.getRandomBytes, - saveHost: (updatedHost, beforePublish) => - saveExistingHostRelayRouting(updatedHost, beforePublish, endpointLifecycle) + saveHost } }), onUpgraded: async ({ host }) => { @@ -99,10 +125,16 @@ function createSupervisor( host: HostProfile, onLog: ConnectionLogSink, onHostUpdated: (host: HostProfile) => void, - endpointLifecycle: HostEndpointPublicationLifecycle + endpointLifecycle: HostEndpointPublicationLifecycle, + saveHost: (host: HostProfile, beforePublish?: () => Promise) => Promise, + applicationResponsiveness: RpcApplicationResponsiveness ): MobileEndpointSupervisor { return new MobileEndpointSupervisor(logical, host, { - openDirect: (endpoint) => connect(endpoint, host.deviceToken, host.publicKeyB64, { onLog }), + openDirect: (endpoint) => + connect(endpoint, host.deviceToken, host.publicKeyB64, { + onLog, + applicationResponsiveness + }), openRelay: (relay, credential, confirmReqId) => connectMobileRelayRpcSession({ relay, @@ -110,7 +142,8 @@ function createSupervisor( resumeCredentialVersion: credential.version, resumeConfirmReqId: confirmReqId, deviceToken: host.deviceToken, - desktopPublicKeyB64: host.publicKeyB64 + desktopPublicKeyB64: host.publicKeyB64, + applicationResponsiveness }), resolveRelay: resolveMobileRelayEndpoint, readBundle: readMobileRelayCredentialBundle, @@ -121,8 +154,7 @@ function createSupervisor( writeMobileRelayCredentialBundle, endpointLifecycle ), - saveHost: (updatedHost, beforePublish) => - saveExistingHostRelayRouting(updatedHost, beforePublish, endpointLifecycle), + saveHost, onHostUpdated, now: Date.now, randomBytes: ExpoCrypto.getRandomBytes, diff --git a/mobile/src/transport/mobile-endpoint-supervisor-contract.ts b/mobile/src/transport/mobile-endpoint-supervisor-contract.ts index 11bbfd5d1b9..b69c0756744 100644 --- a/mobile/src/transport/mobile-endpoint-supervisor-contract.ts +++ b/mobile/src/transport/mobile-endpoint-supervisor-contract.ts @@ -15,7 +15,7 @@ export type MobileEndpointSupervisorDependencies = { resolveRelay: typeof resolveMobileRelayEndpoint readBundle: (hostId: string) => Promise writeBundle: (bundle: MobileRelayCredentialBundle) => Promise - saveHost: (host: HostProfile, beforePublish?: () => Promise) => Promise + saveHost: (host: HostProfile, beforePublish?: () => Promise) => Promise onHostUpdated?: (host: HostProfile) => void now: () => number randomBytes: (length: number) => Uint8Array diff --git a/mobile/src/transport/mobile-endpoint-supervisor-support.ts b/mobile/src/transport/mobile-endpoint-supervisor-support.ts index 1653cb3d7d9..eb0fb4389ad 100644 --- a/mobile/src/transport/mobile-endpoint-supervisor-support.ts +++ b/mobile/src/transport/mobile-endpoint-supervisor-support.ts @@ -20,7 +20,7 @@ export function relayWebSocketUrl(relay: { cellUrl: string; relayHostId: string export async function persistRelayHost( host: HostProfile, relay: MobileRelayEndpoint, - saveHost: (host: HostProfile, beforePublish?: () => Promise) => Promise, + saveHost: (host: HostProfile, beforePublish?: () => Promise) => Promise, beforePublish?: () => Promise ): Promise { const endpoints = [ diff --git a/mobile/src/transport/mobile-relay-credential-bundle.ts b/mobile/src/transport/mobile-relay-credential-bundle.ts index a08fa052459..97a81010461 100644 --- a/mobile/src/transport/mobile-relay-credential-bundle.ts +++ b/mobile/src/transport/mobile-relay-credential-bundle.ts @@ -4,6 +4,7 @@ import type { DeviceCredentialInstalled } from '../../../src/shared/mobile-relay import type { MobileRelayPairingJournal } from './mobile-relay-pairing-journal' import { deletePairingKeychainItem, + deletePairingKeychainItemIfMatches, readPairingKeychainItem, writePairingKeychainItem } from './pairing-keychain' @@ -101,6 +102,19 @@ export async function deleteMobileRelayCredentialBundle(hostId: string): Promise await deletePairingKeychainItem(credentialKey(hostId)) } +export function deleteMobileRelayCredentialBundleIfCurrent( + bundle: MobileRelayCredentialBundle +): Promise { + if (Platform.OS === 'web') { + return Promise.resolve(false) + } + const validated = MobileRelayCredentialBundleSchema.parse(bundle) + return deletePairingKeychainItemIfMatches( + credentialKey(validated.hostId), + JSON.stringify(validated) + ) +} + function requireNativeSecretStore(): void { if (Platform.OS === 'web') { throw new Error('Orca Relay credentials require a native secret store') diff --git a/mobile/src/transport/mobile-relay-pending-requests.ts b/mobile/src/transport/mobile-relay-pending-requests.ts index ba98e1f35b1..f64f34f9396 100644 --- a/mobile/src/transport/mobile-relay-pending-requests.ts +++ b/mobile/src/transport/mobile-relay-pending-requests.ts @@ -5,6 +5,7 @@ export type MobileRelayPendingRequest = { resolve: (response: RpcResponse) => void reject: (error: Error) => void timer: ReturnType + method: string } export function rejectMobileRelayPendingRequests( diff --git a/mobile/src/transport/mobile-relay-rpc-session-contract.ts b/mobile/src/transport/mobile-relay-rpc-session-contract.ts new file mode 100644 index 00000000000..a80ff9f9ee6 --- /dev/null +++ b/mobile/src/transport/mobile-relay-rpc-session-contract.ts @@ -0,0 +1,24 @@ +import type { + DeviceResumeConfirmed, + MobileRelayEndpoint +} from '../../../src/shared/mobile-relay-credential-contract' +import type { RpcApplicationResponsiveness } from './rpc-application-responsiveness' +import type { RpcClient } from './rpc-client' + +export type MobileRelayRpcSession = RpcClient & { + getLeaseExpiresAt(): number | null + getResumeConfirmation(): DeviceResumeConfirmed | null + getFailure(): Error | null +} + +export type MobileRelayRpcSessionOptions = { + relay: MobileRelayEndpoint + resumeToken: string + resumeCredentialVersion: number + resumeConfirmReqId: string + deviceToken: string + desktopPublicKeyB64: string + requestTimeoutMs?: number + createSocket?: (url: string) => WebSocket + applicationResponsiveness?: RpcApplicationResponsiveness +} diff --git a/mobile/src/transport/mobile-relay-rpc-session.test.ts b/mobile/src/transport/mobile-relay-rpc-session.test.ts index 006e0ea02c0..7718a43235d 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.test.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.test.ts @@ -277,6 +277,41 @@ describe('mobile relay RPC session', () => { } }) + it('keeps Relay application stalls latched through probes and recycles a repeated stall', async () => { + const { session } = await authenticateSession() + vi.useFakeTimers() + try { + const first = session + .sendRequest('browser.screenshot', {}, { timeoutMs: 100 }) + .catch((error: unknown) => error) + await vi.advanceTimersByTimeAsync(100) + await expect(first).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' } }) + ) + + expect(session.getRpcUnresponsiveSince?.()).not.toBeNull() + expect(session.getState()).toBe('connected') + const second = session + .sendRequest('browser.screenshot', {}, { timeoutMs: 100 }) + .catch((error: unknown) => error) + await vi.advanceTimersByTimeAsync(100) + + await expect(second).resolves.toMatchObject({ + message: 'relay RPC timed out: browser.screenshot' + }) + expect(session.getState()).toBe('disconnected') + expect(fakes.close).toHaveBeenCalledOnce() + } finally { + vi.useRealTimers() + } + }) + it('counts a late timed-out reply as relay control-plane liveness', 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 230fe1f4a29..806bf395aca 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.ts @@ -1,7 +1,6 @@ import { PairingGetEndpointsResultSchema, - type DeviceResumeConfirmed, - type MobileRelayEndpoint + type DeviceResumeConfirmed } from '../../../src/shared/mobile-relay-credential-contract' import { MobileRelayE2eeLink } from './mobile-relay-e2ee-link' import { MobileRelayRpcStreams } from './mobile-relay-rpc-streams' @@ -15,32 +14,27 @@ import { import { RpcControlProbeFollowUp } from './rpc-control-probe-follow-up' import { openRpcRequestBudget, resolvePostConnectRequestTimeout } from './rpc-request-budget' import { isRpcResponse } from './rpc-response-shape' -import type { RpcClient } from './rpc-client' import type { ConnectionState, RpcResponse } from './types' import { TimedOutControlRequestIndex } from './timed-out-control-request-index' +import { RpcApplicationResponseTracker } from './rpc-application-response-tracker' +import type { + MobileRelayRpcSession, + MobileRelayRpcSessionOptions +} from './mobile-relay-rpc-session-contract' +export type { MobileRelayRpcSession } from './mobile-relay-rpc-session-contract' const CONTROL_PROBE_TIMEOUT_MS = 8_000 const CONTROL_PROBE_INTERVAL_MS = 20_000 -export type MobileRelayRpcSession = RpcClient & { - getLeaseExpiresAt(): number | null - getResumeConfirmation(): DeviceResumeConfirmed | null - getFailure(): Error | null -} - -export function connectMobileRelayRpcSession(args: { - relay: MobileRelayEndpoint - resumeToken: string - resumeCredentialVersion: number - resumeConfirmReqId: string - deviceToken: string - desktopPublicKeyB64: string - requestTimeoutMs?: number - createSocket?: (url: string) => WebSocket -}): MobileRelayRpcSession { +export function connectMobileRelayRpcSession( + args: MobileRelayRpcSessionOptions +): MobileRelayRpcSession { const requestTimeoutMs = args.requestTimeoutMs ?? 30_000 const pending = new Map() const timedOutControlRequestIds = new TimedOutControlRequestIndex() + const applicationResponseTracker = new RpcApplicationResponseTracker( + args.applicationResponsiveness + ) const stateListeners = new Set<(state: ConnectionState) => void>() let state: ConnectionState = 'connecting' let requestCounter = 0 @@ -105,6 +99,7 @@ export function connectMobileRelayRpcSession(args: { getState: () => state, getReconnectAttempt: () => 0, getLastConnectedAt: () => lastConnectedAt, + getRpcUnresponsiveSince: () => applicationResponseTracker.getUnresponsiveSince(), onStateChange(listener) { stateListeners.add(listener) return () => stateListeners.delete(listener) @@ -173,11 +168,15 @@ export function connectMobileRelayRpcSession(args: { // 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 (applicationResponseTracker.recordTimeout(id, method, state === 'connected')) { + fail(error) + return + } if (probeAfterTimeout) { probeControlPlane(true) } }, timeoutMs) - pending.set(id, { resolve, reject, timer }) + pending.set(id, { resolve, reject, timer, method }) if (!sendFrame({ id, method, params })) { clearTimeout(timer) pending.delete(id) @@ -201,6 +200,7 @@ export function connectMobileRelayRpcSession(args: { return } const request = pending.get(value.id) + const lateApplicationResponse = applicationResponseTracker.consumeLateResponse(value.id) if (!value.ok && value.error.code === 'unauthorized') { const error = new MobileE2EEAuthenticationError() if (request) { @@ -215,6 +215,7 @@ export function connectMobileRelayRpcSession(args: { if ( request || timedOutControlRequestIds.consume(value.id) || + lateApplicationResponse || streams.isControlResponse(value) ) { controlResponseSequence += 1 @@ -222,6 +223,7 @@ export function connectMobileRelayRpcSession(args: { if (request) { clearTimeout(request.timer) pending.delete(value.id) + applicationResponseTracker.recordResponse(request.method) request.resolve(value) return } diff --git a/mobile/src/transport/rpc-application-response-tracker.ts b/mobile/src/transport/rpc-application-response-tracker.ts new file mode 100644 index 00000000000..9d4b2d11db0 --- /dev/null +++ b/mobile/src/transport/rpc-application-response-tracker.ts @@ -0,0 +1,45 @@ +import { RpcApplicationResponsiveness } from './rpc-application-responsiveness' +import { TimedOutControlRequestIndex } from './timed-out-control-request-index' + +export class RpcApplicationResponseTracker { + private readonly timedOutRequestIds = new TimedOutControlRequestIndex() + + constructor( + private readonly responsiveness = new RpcApplicationResponsiveness(), + private readonly hooks: { onLatched?: (method: string) => void; onRecovered?: () => void } = {} + ) {} + + recordTimeout(id: string, method: string, connected: boolean): boolean { + if (!connected) { + return false + } + const result = this.responsiveness.recordTimeout(method) + if (result.latched) { + this.hooks.onLatched?.(method) + } + if (result.latched || result.recycle) { + this.timedOutRequestIds.remember(id) + } + return result.recycle + } + + recordResponse(method: string): void { + if (this.responsiveness.recordResponse(method)) { + this.hooks.onRecovered?.() + } + } + + consumeLateResponse(id: string): boolean { + if (!this.timedOutRequestIds.consume(id)) { + return false + } + if (this.responsiveness.recordApplicationResponse()) { + this.hooks.onRecovered?.() + } + return true + } + + getUnresponsiveSince(): number | null { + return this.responsiveness.getUnresponsiveSince() + } +} diff --git a/mobile/src/transport/rpc-application-responsiveness.ts b/mobile/src/transport/rpc-application-responsiveness.ts new file mode 100644 index 00000000000..6cd8bfe85e9 --- /dev/null +++ b/mobile/src/transport/rpc-application-responsiveness.ts @@ -0,0 +1,38 @@ +const RPC_TIMEOUT_RECYCLE_STREAK = 2 + +export class RpcApplicationResponsiveness { + private unresponsiveSince: number | null = null + private timeoutStreak = 0 + + recordResponse(method: string, now = Date.now()): boolean { + if (isRpcHealthProbeMethod(method)) { + return false + } + return this.recordApplicationResponse(now) + } + + recordApplicationResponse(_now = Date.now()): boolean { + const recovered = this.unresponsiveSince !== null + this.unresponsiveSince = null + this.timeoutStreak = 0 + return recovered + } + + recordTimeout(method: string, now = Date.now()): { latched: boolean; recycle: boolean } { + if (isRpcHealthProbeMethod(method)) { + return { latched: false, recycle: false } + } + this.timeoutStreak += 1 + const latched = this.unresponsiveSince === null + this.unresponsiveSince ??= now + return { latched, recycle: this.timeoutStreak >= RPC_TIMEOUT_RECYCLE_STREAK } + } + + getUnresponsiveSince(): number | null { + return this.unresponsiveSince + } +} + +export function isRpcHealthProbeMethod(method: string): boolean { + return method === 'status.get' +} diff --git a/mobile/src/transport/rpc-client-application-responsiveness.test.ts b/mobile/src/transport/rpc-client-application-responsiveness.test.ts new file mode 100644 index 00000000000..0769ec821a6 --- /dev/null +++ b/mobile/src/transport/rpc-client-application-responsiveness.test.ts @@ -0,0 +1,124 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { classifyConnection } from './connection-health' +import { connect } from './rpc-client' + +vi.mock('./e2ee', () => ({ + generateKeyPair: () => ({ + publicKey: new Uint8Array(32), + secretKey: new Uint8Array(32) + }), + deriveSharedKey: () => new Uint8Array(32), + publicKeyFromBase64: () => new Uint8Array(32), + publicKeyToBase64: () => 'client-public-key', + encrypt: (plaintext: string) => `encrypted:${plaintext}`, + decrypt: (raw: string) => raw.replace(/^encrypted:/, ''), + decryptBytes: (bytes: Uint8Array) => bytes +})) + +class MockWebSocket { + static readonly CONNECTING = 0 + static readonly OPEN = 1 + static readonly CLOSING = 2 + static readonly CLOSED = 3 + + readonly CONNECTING = MockWebSocket.CONNECTING + readonly OPEN = MockWebSocket.OPEN + readonly CLOSING = MockWebSocket.CLOSING + readonly CLOSED = MockWebSocket.CLOSED + readyState = MockWebSocket.CONNECTING + onopen: (() => void) | null = null + onclose: (() => void) | null = null + onmessage: ((event: { data: unknown }) => void) | null = null + onerror: (() => void) | null = null + sent: string[] = [] + close = vi.fn(() => { + this.readyState = MockWebSocket.CLOSED + this.onclose?.() + }) + + constructor() { + sockets.push(this) + } + + send(payload: string): void { + this.sent.push(payload) + } + + openAndAuthenticate(): void { + this.readyState = MockWebSocket.OPEN + this.onopen?.() + this.receive(JSON.stringify({ type: 'e2ee_ready' })) + this.receive('encrypted:{"type":"e2ee_authenticated"}') + } + + receive(payload: unknown): void { + this.onmessage?.({ data: payload }) + } +} + +const sockets: MockWebSocket[] = [] +const originalWebSocket = globalThis.WebSocket + +function sentRequest(socket: MockWebSocket, method: string): { id: string } { + for (const payload of socket.sent) { + const request = JSON.parse(payload.replace(/^encrypted:/, '')) as { + id: string + method: string + } + if (request.method === method) { + return request + } + } + throw new Error(`Request not sent: ${method}`) +} + +describe('direct RPC application responsiveness', () => { + beforeEach(() => { + vi.useFakeTimers() + sockets.length = 0 + globalThis.WebSocket = MockWebSocket as unknown as typeof WebSocket + }) + + afterEach(() => { + vi.useRealTimers() + globalThis.WebSocket = originalWebSocket + }) + + it('keeps application stalls latched through probes and recycles a repeated stall', async () => { + const client = connect('ws://desktop.invalid', 'token', 'server-key') + const socket = sockets[0]! + socket.openAndAuthenticate() + const first = client + .sendRequest('browser.screenshot', {}, { timeoutMs: 100 }) + .catch((error: unknown) => error) + + await vi.advanceTimersByTimeAsync(100) + await expect(first).resolves.toMatchObject({ message: 'Request timed out: browser.screenshot' }) + const probe = sentRequest(socket, 'status.get') + socket.receive(`encrypted:${JSON.stringify({ id: probe.id, ok: true, result: {} })}`) + await vi.advanceTimersByTimeAsync(0) + + const rpcUnresponsiveSince = client.getRpcUnresponsiveSince?.() + expect(rpcUnresponsiveSince).not.toBeNull() + expect( + classifyConnection({ + state: client.getState(), + reconnectAttempts: client.getReconnectAttempt(), + lastConnectedAt: client.getLastConnectedAt(), + rpcUnresponsiveSince + }) + ).toMatchObject({ kind: 'warning', label: 'Connected, not responding' }) + + const second = client + .sendRequest('browser.screenshot', {}, { timeoutMs: 100 }) + .catch((error: unknown) => error) + await vi.advanceTimersByTimeAsync(100) + + await expect(second).resolves.toMatchObject({ + message: 'Request timed out: browser.screenshot' + }) + expect(socket.close).toHaveBeenCalledOnce() + expect(client.getState()).toBe('reconnecting') + client.close() + }) +}) diff --git a/mobile/src/transport/rpc-client-context-contract.ts b/mobile/src/transport/rpc-client-context-contract.ts index 56f2c076820..e578fcf8b73 100644 --- a/mobile/src/transport/rpc-client-context-contract.ts +++ b/mobile/src/transport/rpc-client-context-contract.ts @@ -10,6 +10,7 @@ export type RpcClientContextValue = { getState: (hostId: string) => ConnectionState getReconnectAttempt: (hostId: string) => number getLastConnectedAt: (hostId: string) => number | null + getRpcUnresponsiveSince: (hostId: string) => number | null getActivePath: (hostId: string) => MobileConnectionPath subscribeHostState: (hostId: string, listener: (state: ConnectionState) => void) => () => void getAllClients: () => { hostId: string; client: RpcClient }[] diff --git a/mobile/src/transport/rpc-client-contract.ts b/mobile/src/transport/rpc-client-contract.ts new file mode 100644 index 00000000000..b2244eb72bd --- /dev/null +++ b/mobile/src/transport/rpc-client-contract.ts @@ -0,0 +1,48 @@ +import type { BrowserScreencastFrame } from './browser-screencast-protocol' +import type { RpcApplicationResponsiveness } from './rpc-application-responsiveness' +import type { ConnectionLogSink, ConnectionState, RpcResponse } from './types' + +export type SendRequestOptions = { + timeoutMs?: number + /** Spend `timeoutMs` across connect-wait and request instead of restarting after connect. */ + budgetSpansConnect?: boolean + /** Do not grant the normal post-connect acknowledgement floor. */ + strictDeadline?: boolean + /** Reject immediately instead of replaying a disconnected request after reconnect. */ + failWhenDisconnected?: boolean +} + +export type SubscribeOptions = { + onBinaryFrame?: (frame: BrowserScreencastFrame) => void +} + +export type RpcClient = { + sendRequest: ( + method: string, + params?: unknown, + options?: SendRequestOptions + ) => Promise + subscribe: ( + method: string, + params: unknown, + onData: (result: unknown) => void, + options?: SubscribeOptions + ) => () => void + updateTerminalSubscriptionViewport: ( + terminal: string, + viewport: { cols: number; rows: number } + ) => void + getState: () => ConnectionState + getReconnectAttempt: () => number + getLastConnectedAt: () => number | null + getRpcUnresponsiveSince?: () => number | null + onStateChange: (listener: (state: ConnectionState) => void) => () => void + notifyForeground: () => void + close: () => void +} + +export type ConnectOptions = { + onStateChange?: (state: ConnectionState) => void + onLog?: ConnectionLogSink + applicationResponsiveness?: RpcApplicationResponsiveness +} diff --git a/mobile/src/transport/rpc-client-react-context.tsx b/mobile/src/transport/rpc-client-react-context.tsx new file mode 100644 index 00000000000..d6033f23036 --- /dev/null +++ b/mobile/src/transport/rpc-client-react-context.tsx @@ -0,0 +1,22 @@ +import { createContext, useContext, type ReactNode } from 'react' +import type { RpcClientContextValue } from './rpc-client-context-contract' + +const RpcClientReactContext = createContext(null) + +export function RpcClientContextBoundary({ + value, + children +}: { + value: RpcClientContextValue + children: ReactNode +}) { + return {children} +} + +export function useRpcClientContext(): RpcClientContextValue { + const context = useContext(RpcClientReactContext) + if (!context) { + throw new Error('useHostClient must be used inside ') + } + return context +} diff --git a/mobile/src/transport/rpc-client.ts b/mobile/src/transport/rpc-client.ts index 067a6908697..7df5bdf499f 100644 --- a/mobile/src/transport/rpc-client.ts +++ b/mobile/src/transport/rpc-client.ts @@ -1,10 +1,4 @@ -import type { - RpcResponse, - RpcSuccess, - ConnectionState, - ConnectionLogLevel, - ConnectionLogSink -} from './types' +import type { RpcResponse, RpcSuccess, ConnectionState, ConnectionLogLevel } from './types' import { generateKeyPair, deriveSharedKey, @@ -22,6 +16,13 @@ import { decodeBrowserScreencastFrame, type BrowserScreencastFrame } from './browser-screencast-protocol' +import type { + ConnectOptions, + RpcClient, + SendRequestOptions, + SubscribeOptions +} from './rpc-client-contract' +export type { ConnectOptions, RpcClient, SendRequestOptions } from './rpc-client-contract' import { buildStreamUnsubscribe, buildTerminalUnsubscribeParams, @@ -44,10 +45,12 @@ import { } from './rpc-stream-response-shape' import { websocketPayloadToUint8 } from './websocket-payload-bytes' import { TimedOutControlRequestIndex } from './timed-out-control-request-index' +import { RpcApplicationResponseTracker } from './rpc-application-response-tracker' type PendingRequest = { resolve: (response: RpcResponse) => void reject: (error: Error) => void + method?: string } type ConnectWaiter = { @@ -56,28 +59,6 @@ type ConnectWaiter = { timeout: ReturnType | null } -export type SendRequestOptions = { - timeoutMs?: number - /** Spend `timeoutMs` across connect-wait AND the request instead of giving each - * phase its own. Interactive chat writes need it: they run as sequential loops - * under one shared budget, so a per-phase clock lets the composer sit `sending` - * for a multiple of the stated ceiling. Off by default — the long-running - * callers (worktree create, dictation finish, credit reset) sized their budgets - * against the post-connect clock, and squeezing them to the floor after a slow - * reconnect would fail sends that used to land. */ - budgetSpansConnect?: boolean - /** Keep the connect-plus-request deadline strict instead of granting the - * normal post-connect acknowledgement floor. */ - strictDeadline?: boolean - /** Reject immediately when not connected — a send parked in the connect wait - * replays stale terminal bytes into the PTY after reconnect. */ - failWhenDisconnected?: boolean -} - -type SubscribeOptions = { - onBinaryFrame?: (frame: BrowserScreencastFrame) => void -} - type StreamingListener = (result: unknown) => void type StreamRequest = { @@ -90,33 +71,6 @@ type StreamRequest = { sent?: 'awaiting' | 'received' } -export type RpcClient = { - sendRequest: ( - method: string, - params?: unknown, - options?: SendRequestOptions - ) => Promise - subscribe: ( - method: string, - params: unknown, - onData: StreamingListener, - options?: SubscribeOptions - ) => () => void - updateTerminalSubscriptionViewport: ( - terminal: string, - viewport: { cols: number; rows: number } - ) => void - getState: () => ConnectionState - // 0 means never failed (reset once the handshake authenticates); the UI escalates "Reconnecting…" to "Can't connect" past a threshold. - getReconnectAttempt: () => number - // Last 'connected' timestamp (ms epoch); null = never connected. Lets the UI tell "never reachable" from "transient blip". - getLastConnectedAt: () => number | null - onStateChange: (listener: (state: ConnectionState) => void) => () => void - // Why: app-resume hook — iOS/Android can kill the TCP path while backgrounded; call on AppState 'active' to recover. - notifyForeground: () => void - close: () => void -} - // Why: tiered backoff — fast early entries recover blips; the slow tail avoids burning a SYN every 4s on an unreachable desktop. const RECONNECT_DELAYS = [500, 1000, 2000, 4000, 8000, 15_000, 30_000, 60_000] // Why: ≈6 min of failure before the re-pair banner; MUST stay aligned with connection-health.ts UNREACHABLE_ATTEMPTS. @@ -139,12 +93,6 @@ const WEBSOCKET_CONNECTING_STATE = 0 // Why: RN auto-pongs pings natively, so JS needs an app-level probe to detect half-open sockets. const ACTIVITY_PROBE_INTERVAL_MS = 20_000 -export type ConnectOptions = { - onStateChange?: (state: ConnectionState) => void - // Fires for every lifecycle event so the UI can show where 'Connecting…' is stuck (e.g. broken Tailscale route). - onLog?: ConnectionLogSink -} - export function connect( endpoint: string, deviceToken: string, @@ -197,6 +145,14 @@ export function connect( const pending = new Map() const timedOutControlRequestIds = new TimedOutControlRequestIndex() + const applicationResponseTracker = new RpcApplicationResponseTracker( + options.applicationResponsiveness, + { + onLatched: (method) => emitLog('warn', 'RPC channel not responding', `${method} timed out`), + onRecovered: () => + emitLog('success', 'RPC channel recovered', 'An application request completed') + } + ) const streamListeners = new Map() const terminalStreamListeners = new Map() const terminalStreamIdsByRequest = new Map>() @@ -513,9 +469,12 @@ export function connect( if (!isRpcResponse(response)) { return } + const request = pending.get(response.id) + const lateApplicationResponse = applicationResponseTracker.consumeLateResponse(response.id) if ( - pending.has(response.id) || + request || timedOutControlRequestIds.consume(response.id) || + lateApplicationResponse || consumeFirstStreamControlResponse(streamListeners.get(response.id)) ) { controlResponseSequence++ @@ -526,7 +485,6 @@ export function connect( handleAuthRejection('Unauthorized — pairing may be revoked') return } - const isStreaming = response.ok && (response as RpcSuccess).streaming === true if (isStreaming) { @@ -1057,6 +1015,11 @@ export function connect( reject(markRpcDeliveryUnknown(new Error(`Request timed out: ${method}`))) if (requestWs === ws) { timedOutControlRequestIds.remember(id) + if (applicationResponseTracker.recordTimeout(id, method, true)) { + emitLog('error', 'RPC channel unresponsive', 'Recycling the connection') + forceSocketReconnect(requestWs) + return + } } runActivityProbe(requestWs, true) }, timeoutMs) @@ -1064,12 +1027,14 @@ export function connect( pending.set(id, { resolve: (response) => { clearTimeout(timeout) + applicationResponseTracker.recordResponse(method) resolve(response) }, reject: (error) => { clearTimeout(timeout) reject(error) - } + }, + method }) if (!sendEncrypted({ id, deviceToken, method, params })) { @@ -1168,6 +1133,10 @@ export function connect( return lastConnectedAt }, + getRpcUnresponsiveSince(): number | null { + return applicationResponseTracker.getUnresponsiveSince() + }, + onStateChange(listener: (state: ConnectionState) => void): () => void { stateListeners.add(listener) return () => stateListeners.delete(listener) diff --git a/mobile/src/transport/stable-logical-rpc-client.ts b/mobile/src/transport/stable-logical-rpc-client.ts index 1fd19c9c01f..4e8b985160d 100644 --- a/mobile/src/transport/stable-logical-rpc-client.ts +++ b/mobile/src/transport/stable-logical-rpc-client.ts @@ -134,6 +134,7 @@ export function createStableLogicalRpcClient( getState: () => state, getReconnectAttempt: () => activeSession.getReconnectAttempt(), getLastConnectedAt: () => activeSession.getLastConnectedAt(), + getRpcUnresponsiveSince: () => activeSession.getRpcUnresponsiveSince?.() ?? null, onStateChange(listener) { stateListeners.add(listener) return () => stateListeners.delete(listener)