fix(mobile): detect unresponsive application RPCs

This commit is contained in:
Brennan Benson
2026-08-01 11:58:48 -07:00
parent 7ee1184a0e
commit 203ffd241b
37 changed files with 893 additions and 253 deletions
+25 -23
View File
@@ -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<RpcClient | null>(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'}
</Text>
</View>
{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 (
<Pressable
style={styles.reconnectButton}
onPress={() =>
void forceReconnectHost(hostId!).catch(showForceReconnectError)
}
hitSlop={8}
>
<Text style={styles.reconnectButtonText}>Reconnect</Text>
</Pressable>
)
})()}
{(() => {
// 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 (
<Pressable
style={styles.reconnectButton}
onPress={() =>
void forceReconnectHost(hostId!).catch(showForceReconnectError)
}
hitSlop={8}
>
<Text style={styles.reconnectButtonText}>Reconnect</Text>
</Pressable>
)
})()}
</>
)
})()}
@@ -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 =
+3 -8
View File
@@ -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<RpcClient | null>(null)
const reposRef = useRef<RepoSummary[]>([])
const loadGenerationRef = useRef(0)
@@ -8655,8 +8651,7 @@ export default function MobileTasksScreen() {
const headerVerdict = classifyConnection({
state: connState,
reconnectAttempts,
lastConnectedAt
...connectionHealth
})
const emptyLabel =
connState !== 'connected'
+3
View File
@@ -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 (
@@ -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
@@ -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])
}
@@ -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<string, number | null> {
const context = useRpcClientContext()
const [values, setValues] = useState<Record<string, number | null>>({})
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<T>(
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<T>(
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
}
+25 -73
View File
@@ -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<RpcClientContextValue | null>(null)
export function RpcClientProvider({ children }: { children: ReactNode }) {
const storeRef = useRef<Map<string, StoreEntry>>(new Map())
const stateListenersRef = useRef<Map<string, Set<(state: ConnectionState) => 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<string, RpcApplicationResponsiveness>())
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 <Ctx.Provider value={value}>{children}</Ctx.Provider>
}
export function useRpcClientContext(): RpcClientContextValue {
const ctx = useContext(Ctx)
if (!ctx) {
throw new Error('useHostClient must be used inside <RpcClientProvider>')
}
return ctx
return <RpcClientContextBoundary value={value}>{children}</RpcClientContextBoundary>
}
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'
@@ -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' }
}
@@ -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<void>((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<void>((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
@@ -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<void>,
endpointLifecycle = getHostEndpointPublicationLifecycle(host.id)
): Promise<void> {
): Promise<number> {
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<void>,
endpointLifecycle = getHostEndpointPublicationLifecycle(host.id)
endpointLifecycle = getHostEndpointPublicationLifecycle(host.id),
cleanupBundle = deleteMobileRelayCredentialBundleIfCurrent
): Promise<void> {
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
}
})
}
@@ -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')
@@ -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()
@@ -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<void> {
await deleteHostDeviceToken(hostId)
await deleteMobileRelayCredentialBundle(hostId)
await deleteMobileRelayDirectUpgradeJournal(hostId)
}
export function createRemovedHostCredentialDelete(
hostExists: (hostId: string) => Promise<boolean>
): (hostId: string) => Promise<void> {
return async (hostId) => {
if (!(await hostExists(hostId))) {
await deleteHostCredentials(hostId)
}
}
}
@@ -25,9 +25,10 @@ export function shareHostListLoad(load: () => Promise<HostProfile[]>): Promise<H
}
/** Call after every durable host write so no later read is served a pre-write pass. */
export function dropSharedHostListLoad(): void {
export function dropSharedHostListLoad(): number {
revision += 1
inflight = null
return revision
}
/** Stop offering a possibly stalled pass without invalidating durable snapshot revisions. */
+14 -3
View File
@@ -4,23 +4,34 @@ import { createStableLogicalRpcClient } from './stable-logical-rpc-client'
import type { ConnectionLogSink, HostProfile } from './types'
import { directPathForEndpoint } from './mobile-direct-endpoint-probe'
import { startMobileEndpointLifecycle } from './mobile-endpoint-lifecycle'
import { RpcApplicationResponsiveness } from './rpc-application-responsiveness'
export function openHostLogicalClient(
host: HostProfile,
onLog: ConnectionLogSink,
onHostUpdated: (host: HostProfile) => 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')
@@ -29,10 +29,13 @@ export function recordDurableHostIdentity(
})
}
export function retireHostProfilePublication(hostId: string): void {
export function retireHostProfilePublication(hostId: string): Promise<void> | 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<void>
): Promise<void> {
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()
@@ -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)
})
})
@@ -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<string, CachedHostProfile>()
private readonly latestVersions = new Map<string, number>()
private readonly publicationVersions = new Map<string, number>()
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 {
@@ -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<void>((resolve) => {
markPublicationStarted = resolve
})
const stalledPublication = publishHostProfileTransaction(
{ ...HOST, deviceToken: 'token-1' },
async () => {
markPublicationStarted?.()
await new Promise<void>(() => {})
},
saveHost
)
void stalledPublication.catch(() => {})
await publicationStarted
await expect(removeHost(HOST.id)).resolves.toBeUndefined()
expect(JSON.parse(storedHostsRaw)).toEqual([])
expect(scheduleCleanup).toHaveBeenCalledOnce()
})
})
+14 -19
View File
@@ -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<HostProfile, 'endpoint' | 'publicKeyB64'>
async function deleteHostCredentials(hostId: string): Promise<void> {
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<string, string>()
// 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<StoredHost
return stored ? { endpoint: stored.endpoint, publicKeyB64: stored.publicKeyB64 } : null
}
const deleteRemovedHostCredentials = createRemovedHostCredentialDelete(async (hostId) =>
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<void> {
await persistHost(host)
}
async function persistHost(host: HostProfile): Promise<void> {
const validated = HostProfileSchema.parse(host)
const stored = toStored(validated)
const duplicateHostIds = new Set<string>()
@@ -256,10 +245,10 @@ async function persistHost(host: HostProfile): Promise<void> {
}
export async function removeHost(hostId: string): Promise<void> {
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<void> {
}
// 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<{
@@ -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<void>) => {
publishedSourceRevision = await saveExistingHostRelayRouting(
host,
beforePublish,
endpointLifecycle
)
return publishedSourceRevision
}
const startSupervisor = async (host: HostProfile): Promise<void> => {
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<void>) => Promise<number>,
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,
@@ -15,7 +15,7 @@ export type MobileEndpointSupervisorDependencies = {
resolveRelay: typeof resolveMobileRelayEndpoint
readBundle: (hostId: string) => Promise<MobileRelayCredentialBundle | null>
writeBundle: (bundle: MobileRelayCredentialBundle) => Promise<void>
saveHost: (host: HostProfile, beforePublish?: () => Promise<void>) => Promise<void>
saveHost: (host: HostProfile, beforePublish?: () => Promise<void>) => Promise<unknown>
onHostUpdated?: (host: HostProfile) => void
now: () => number
randomBytes: (length: number) => Uint8Array
@@ -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<void>) => Promise<void>,
saveHost: (host: HostProfile, beforePublish?: () => Promise<void>) => Promise<unknown>,
beforePublish?: () => Promise<void>
): Promise<HostProfile> {
const endpoints = [
@@ -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<boolean> {
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')
@@ -5,6 +5,7 @@ export type MobileRelayPendingRequest = {
resolve: (response: RpcResponse) => void
reject: (error: Error) => void
timer: ReturnType<typeof setTimeout>
method: string
}
export function rejectMobileRelayPendingRequests(
@@ -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
}
@@ -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()
@@ -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<string, MobileRelayPendingRequest>()
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
}
@@ -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()
}
}
@@ -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'
}
@@ -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()
})
})
@@ -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 }[]
@@ -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<RpcResponse>
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
}
@@ -0,0 +1,22 @@
import { createContext, useContext, type ReactNode } from 'react'
import type { RpcClientContextValue } from './rpc-client-context-contract'
const RpcClientReactContext = createContext<RpcClientContextValue | null>(null)
export function RpcClientContextBoundary({
value,
children
}: {
value: RpcClientContextValue
children: ReactNode
}) {
return <RpcClientReactContext.Provider value={value}>{children}</RpcClientReactContext.Provider>
}
export function useRpcClientContext(): RpcClientContextValue {
const context = useContext(RpcClientReactContext)
if (!context) {
throw new Error('useHostClient must be used inside <RpcClientProvider>')
}
return context
}
+34 -65
View File
@@ -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<typeof setTimeout> | 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<RpcResponse>
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<string, PendingRequest>()
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<string, StreamRequest>()
const terminalStreamListeners = new Map<number, StreamingListener>()
const terminalStreamIdsByRequest = new Map<string, Set<number>>()
@@ -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)
@@ -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)