starred/orca
1
0
Fork 0
mirror of https://github.com/stablyai/orca.git synced 2026-10-29 16:17:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mobile): harden relay recovery and reconnects

Browse Source
This commit is contained in:
Brennan Benson
2026-08-01 14:00:27 -07:00
parent 19f668cf7c
commit d417246893
21 changed files with 645 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mobile/app/h/[hostId]/edit.tsx
+11 -6
View File
@@ -17,11 +17,12 @@ import { colors, radii, spacing, typography } from '../../../src/theme/mobile-th
import { loadHosts, updateHostNameAndEndpoint } from '../../../src/transport/host-store'
import { getHostListLoadRevision } from '../../../src/transport/host-list-load-sharing'
import { displayHostEndpoint } from '../../../src/transport/host-endpoint'
import { resolveHostEndpointEdit } from '../../../src/transport/host-endpoint-edit'
import {
hostProfileAfterEdit,
resolveHostEndpointEdit
} from '../../../src/transport/host-endpoint-edit'
import { useForceReconnect, usePrimeHosts } from '../../../src/transport/client-context'
useForceReconnect,
useForceReconnectAfterEdit,
usePrimeHosts
} from '../../../src/transport/client-context'
import { showForceReconnectError } from '../../../src/transport/force-reconnect-feedback'
import type { HostProfile } from '../../../src/transport/types'
@@ -31,6 +32,7 @@ export default function EditHostScreen() {
const { hostId } = useLocalSearchParams<{ hostId: string }>()
const primeHosts = usePrimeHosts()
const forceReconnectHost = useForceReconnect()
const forceReconnectHostAfterEdit = useForceReconnectAfterEdit()
const [host, setHost] = useState<HostProfile | null>(null)
const [loadError, setLoadError] = useState<string | null>(null)
@@ -109,7 +111,7 @@ export default function EditHostScreen() {
...(willRename ? { name: nextName } : {}),
...(nextEndpoint !== undefined ? { endpoint: nextEndpoint } : {})
}
let reconnectProfile = hostProfileAfterEdit(host, updates)
let reconnectProfile: HostProfile | null = null
savingRef.current = true
setSaving(true)
setSaveError(null)
@@ -157,7 +159,10 @@ export default function EditHostScreen() {
// Why: reconnect is a follow-on side effect of a save that already
// committed — its failure or a hang must not be reported as a save
// failure or block navigating back.
void forceReconnectHost(host.id, reconnectProfile).catch(showForceReconnectError)
const reconnect = reconnectProfile
? forceReconnectHost(host.id, reconnectProfile)
: forceReconnectHostAfterEdit(host.id, host, updates)
void reconnect.catch(showForceReconnectError)
}
}
mobile/src/host-edit-route-accessibility.test.ts
+2
View File
@@ -5,6 +5,7 @@ import EditHostScreen from '../app/h/[hostId]/edit'
const dependencies = vi.hoisted(() => ({
back: vi.fn(),
forceReconnectHostAfterEdit: vi.fn(),
forceReconnectHost: vi.fn(),
loadHosts: vi.fn(),
primeHosts: vi.fn(),
@@ -43,6 +44,7 @@ vi.mock('./transport/host-store', () => ({
vi.mock('./transport/client-context', () => ({
useForceReconnect: () => dependencies.forceReconnectHost,
useForceReconnectAfterEdit: () => dependencies.forceReconnectHostAfterEdit,
usePrimeHosts: () => dependencies.primeHosts
}))
mobile/src/host-edit-save-flow.test.ts
+31 -4
View File
@@ -7,6 +7,7 @@ const dependencies = vi.hoisted(() => ({
alert: vi.fn(),
back: vi.fn(),
forceReconnectHost: vi.fn(),
forceReconnectHostAfterEdit: vi.fn(),
getHostListLoadRevision: vi.fn(() => 0),
loadHosts: vi.fn(),
primeHosts: vi.fn(),
@@ -51,6 +52,7 @@ vi.mock('./transport/host-list-load-sharing', () => ({
vi.mock('./transport/client-context', () => ({
useForceReconnect: () => dependencies.forceReconnectHost,
useForceReconnectAfterEdit: () => dependencies.forceReconnectHostAfterEdit,
usePrimeHosts: () => dependencies.primeHosts
}))
@@ -147,6 +149,7 @@ describe('edit host handleSave', () => {
dependencies.alert.mockReset()
dependencies.back.mockReset()
dependencies.forceReconnectHost.mockReset().mockResolvedValue(undefined)
dependencies.forceReconnectHostAfterEdit.mockReset().mockResolvedValue(undefined)
dependencies.getHostListLoadRevision.mockReset().mockReturnValue(0)
dependencies.loadHosts.mockReset().mockResolvedValue([HOST_FIXTURE])
dependencies.primeHosts.mockReset()
@@ -264,8 +267,23 @@ describe('edit host handleSave', () => {
await pressSave(renderer)
expect(dependencies.primeHosts).not.toHaveBeenCalled()
expect(dependencies.forceReconnectHost).toHaveBeenCalledWith('host-1', {
...HOST_FIXTURE,
expect(dependencies.forceReconnectHost).not.toHaveBeenCalled()
expect(dependencies.forceReconnectHostAfterEdit).toHaveBeenCalledWith('host-1', HOST_FIXTURE, {
endpoint: 'ws://192.168.1.20:6768'
})
expect(dependencies.back).toHaveBeenCalledTimes(1)
act(() => renderer.unmount())
})
it('merges the endpoint edit into the current cache when the host reload omits it', async () => {
dependencies.loadHosts.mockResolvedValueOnce([HOST_FIXTURE]).mockResolvedValueOnce([])
const renderer = await renderEditHostRoute()
setFieldValue(renderer, 'Address', '192.168.1.20:6768')
await pressSave(renderer)
expect(dependencies.forceReconnectHost).not.toHaveBeenCalled()
expect(dependencies.forceReconnectHostAfterEdit).toHaveBeenCalledWith('host-1', HOST_FIXTURE, {
endpoint: 'ws://192.168.1.20:6768'
})
expect(dependencies.back).toHaveBeenCalledTimes(1)
@@ -297,8 +315,16 @@ describe('edit host handleSave', () => {
...HOST_FIXTURE,
endpoint: 'ws://192.168.1.20:6768',
endpoints: [
{ id: 'direct-primary', kind: 'lan' as const, url: 'ws://192.168.1.20:6768' },
{ id: 'relay-primary', kind: 'relay' as const, url: 'wss://relay.invalid/host-1' }
{
id: 'direct-primary',
kind: 'lan' as const,
url: 'ws://192.168.1.20:6768'
},
{
id: 'relay-primary',
kind: 'relay' as const,
url: 'wss://relay.invalid/host-1'
}
]
}
dependencies.loadHosts.mockResolvedValueOnce([HOST_FIXTURE]).mockResolvedValueOnce([relayHost])
@@ -370,6 +396,7 @@ describe('edit host load() error states', () => {
dependencies.alert.mockReset()
dependencies.back.mockReset()
dependencies.forceReconnectHost.mockReset().mockResolvedValue(undefined)
dependencies.forceReconnectHostAfterEdit.mockReset().mockResolvedValue(undefined)
dependencies.getHostListLoadRevision.mockReset().mockReturnValue(0)
dependencies.loadHosts.mockReset().mockResolvedValue([HOST_FIXTURE])
dependencies.primeHosts.mockReset()
mobile/src/transport/client-context-actions.ts
+1
View File
@@ -2,4 +2,5 @@ import { useRpcClientContext } from './rpc-client-react-context'
export const useCloseHost = () => useRpcClientContext().closeHost
export const useForceReconnect = () => useRpcClientContext().forceReconnect
export const useForceReconnectAfterEdit = () => useRpcClientContext().forceReconnectAfterEdit
export const usePrimeHosts = () => useRpcClientContext().primeHosts
mobile/src/transport/client-context.tsx
+43 -9
View File
@@ -193,13 +193,8 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
entry.refCount = Math.max(0, entry.refCount - 1)
}, [])
const forceReconnect = useCallback(
(hostId: string, requestedHost?: HostProfile): Promise<void> => {
const profile = primedHostsRef.current.reconnectProfile(
hostId,
getHostListLoadRevision(),
requestedHost
)
const runForceReconnect = useCallback(
(hostId: string, profile: HostOpenProfile): Promise<void> => {
return forceReconnectCoordinatorRef.current.run({
hostId,
profileVersion: profile.version,
@@ -220,6 +215,35 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
[openEntry]
)
const forceReconnect = useCallback(
(hostId: string, requestedHost?: HostProfile): Promise<void> => {
const profile = primedHostsRef.current.reconnectProfile(
hostId,
getHostListLoadRevision(),
requestedHost
)
return runForceReconnect(hostId, profile)
},
[runForceReconnect]
)
const forceReconnectAfterEdit = useCallback(
(
hostId: string,
fallbackHost: HostProfile,
updates: { name?: string; endpoint?: string }
): Promise<void> => {
const profile = primedHostsRef.current.reconnectEditedProfile(
hostId,
getHostListLoadRevision(),
fallbackHost,
updates
)
return runForceReconnect(hostId, profile)
},
[runForceReconnect]
)
const getState = useCallback((hostId: string): ConnectionState => {
return storeRef.current.get(hostId)?.state ?? 'disconnected'
}, [])
@@ -262,7 +286,10 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
[]
)
const getAllClients = useCallback((): Array<{ hostId: string; client: RpcClient }> => {
const getAllClients = useCallback((): Array<{
hostId: string
client: RpcClient
}> => {
const out: Array<{ hostId: string; client: RpcClient }> = []
for (const [hostId, entry] of storeRef.current) {
out.push({ hostId, client: entry.client })
@@ -302,6 +329,7 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
acquire,
release,
forceReconnect,
forceReconnectAfterEdit,
closeHost: closeEntry,
getState,
getReconnectAttempt,
@@ -317,6 +345,7 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
acquire,
release,
forceReconnect,
forceReconnectAfterEdit,
closeEntry,
getState,
getReconnectAttempt,
@@ -394,5 +423,10 @@ export function useHostClient(hostId: string | undefined): {
}
// 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 {
useCloseHost,
useForceReconnect,
useForceReconnectAfterEdit,
usePrimeHosts
} from './client-context-actions'
export { useAllHostClients } from './client-context-all-host-clients'
mobile/src/transport/host-endpoint-edit.ts
+3 -4
View File
@@ -6,6 +6,8 @@ import {
} from './host-endpoint'
import type { HostProfile } from './types'
export type HostProfileEdit = { name?: string; endpoint?: string }
export type HostEndpointEditResolution =
| { kind: 'unchanged'; endpoint: string }
| { kind: 'changed'; endpoint: string }
@@ -44,10 +46,7 @@ export function resolveHostEndpointEdit(
}
}
export function hostProfileAfterEdit(
host: HostProfile,
updates: { name?: string; endpoint?: string }
): HostProfile {
export function hostProfileAfterEdit(host: HostProfile, updates: HostProfileEdit): HostProfile {
const updatedEndpoint = updates.endpoint
const endpoints =
updatedEndpoint === undefined || !host.endpoints
mobile/src/transport/host-profile-publication.ts
+10 -3
View File
@@ -14,6 +14,10 @@ export function getHostProfilePublicationRevision(hostId: string): number {
return revisionByHost.get(hostId) ?? 0
}
export function recordHostProfileMutation(hostId: string): void {
revisionByHost.set(hostId, getHostProfilePublicationRevision(hostId) + 1)
}
export function getPublishedHostIdentity(
hostId: string
): Pick<HostProfile, 'deviceToken' | 'publicKeyB64'> | undefined {
@@ -43,7 +47,10 @@ export function beginHostEndpointPublicationLifecycle(
): HostEndpointPublicationLifecycle {
const generation = (endpointGenerationByHost.get(hostId) ?? 0) + 1
endpointGenerationByHost.set(hostId, generation)
return { generation, profileRevision: getHostProfilePublicationRevision(hostId) }
return {
generation,
profileRevision: getHostProfilePublicationRevision(hostId)
}
}
export function getHostEndpointPublicationLifecycle(
@@ -77,9 +84,9 @@ export function serializeHostProfilePublication<T>(
export function publishHostProfileTransaction(
host: HostProfile,
beforeHostSave: (() => Promise<void>) | null,
saveHost: (host: HostProfile) => Promise<void>
saveHost: (host: HostProfile) => Promise<void>,
publicationRevision = getHostProfilePublicationRevision(host.id)
): Promise<void> {
const publicationRevision = getHostProfilePublicationRevision(host.id)
return serializeHostProfilePublication(host.id, async () => {
requirePublicationRevision(host.id, publicationRevision)
await beforeHostSave?.()
mobile/src/transport/host-reconnect-profile-cache.test.ts
+44 -1
View File
@@ -18,7 +18,11 @@ describe('HostReconnectProfileCache', () => {
...HOST,
endpoints: [
{ id: 'direct-primary', kind: 'lan', url: HOST.endpoint },
{ id: 'relay-primary', kind: 'relay', url: 'wss://relay.invalid/v1/connect/host' }
{
id: 'relay-primary',
kind: 'relay',
url: 'wss://relay.invalid/v1/connect/host'
}
],
relayHostId: 'AbCdEf0123_-xyZ9',
relay: {
@@ -95,4 +99,43 @@ describe('HostReconnectProfileCache', () => {
expect(cache.get(HOST.id, 4)).toEqual(editedHost)
})
it('merges a fallback endpoint edit into the latest cached relay routing', () => {
const cache = new HostReconnectProfileCache()
const relayHost: HostProfile = {
...HOST,
endpoints: [
{ id: 'direct-primary', kind: 'lan', url: HOST.endpoint },
{
id: 'relay-primary',
kind: 'relay',
url: 'wss://relay.invalid/v1/connect/host'
}
],
relayHostId: 'AbCdEf0123_-xyZ9',
relay: {
v: 1,
directorUrl: 'https://relay.invalid',
cellUrl: 'https://relay.invalid',
assignmentEpoch: 1,
relayHostId: 'AbCdEf0123_-xyZ9',
e2eeFraming: 2
}
}
cache.prime(HOST, 4)
cache.prime(relayHost, 5)
const edited = cache.reconnectEditedProfile(HOST.id, 6, HOST, {
endpoint: 'ws://127.0.0.1:2'
})
expect(edited.host).toEqual({
...relayHost,
endpoint: 'ws://127.0.0.1:2',
endpoints: [
{ id: 'direct-primary', kind: 'lan', url: 'ws://127.0.0.1:2' },
relayHost.endpoints![1]
]
})
})
})
mobile/src/transport/host-reconnect-profile-cache.ts
+37 -2
View File
@@ -1,4 +1,5 @@
import type { HostProfile } from './types'
import { hostProfileAfterEdit, type HostProfileEdit } from './host-endpoint-edit'
type CachedHostProfile = {
host: HostProfile
@@ -31,7 +32,12 @@ export class HostReconnectProfileCache {
? previous.publicationVersion
: (this.publicationVersions.get(host.id) ?? 0) + 1
this.publicationVersions.set(host.id, publicationVersion)
this.profiles.set(host.id, { host, version, publicationVersion, sourceRevision })
this.profiles.set(host.id, {
host,
version,
publicationVersion,
sourceRevision
})
return version
}
@@ -53,6 +59,22 @@ export class HostReconnectProfileCache {
}
}
reconnectEditedProfile(
hostId: string,
currentRevision: number,
fallbackHost: HostProfile,
updates: HostProfileEdit
): HostOpenProfile {
const cached = this.profiles.get(hostId)?.host
const base = cached && reconnectIdentityMatches(cached, fallbackHost) ? cached : fallbackHost
const host = hostProfileAfterEdit(base, updates)
return {
host,
sourceRevision: currentRevision,
version: this.prime(host, currentRevision)
}
}
openProfile(
hostId: string,
currentRevision: number,
@@ -92,7 +114,12 @@ export class HostReconnectProfileCache {
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 })
this.profiles.set(host.id, {
host,
version,
publicationVersion,
sourceRevision
})
return version
}
@@ -154,3 +181,11 @@ function reconnectProfileMatches(left: HostProfile, right: HostProfile): boolean
JSON.stringify(left.relay ?? null) === JSON.stringify(right.relay ?? null)
)
}
function reconnectIdentityMatches(left: HostProfile, right: HostProfile): boolean {
return (
left.id === right.id &&
left.deviceToken === right.deviceToken &&
left.publicKeyB64 === right.publicKeyB64
)
}
mobile/src/transport/host-store.test.ts
+3 -7
View File
@@ -44,6 +44,7 @@ import {
updateLastConnected
} from './host-store'
import {
MobileRelayUpgradeLifecycleRetiredError,
MobileRelayUpgradeHostRemovedError,
MobileRelayUpgradeHostSupersededError,
saveExistingHostRelayRouting,
@@ -463,13 +464,8 @@ describe('host-store list mutations', () => {
})
releaseOldToken?.('token-1')
await stalePublication
const [overlay] = JSON.parse(storedOverlayRaw ?? '[]') as MobileRelayHostOverlay[]
expect(overlay?.endpoints).toContainEqual({
id: 'direct-primary',
kind: 'lan',
url: 'ws://127.0.0.1:9'
})
await expect(stalePublication).rejects.toBeInstanceOf(MobileRelayUpgradeLifecycleRetiredError)
expect(JSON.parse(storedOverlayRaw ?? '[]')).toEqual([])
})
it('lets same-host re-pair supersede a publication stalled on identity lookup', async () => {
mobile/src/transport/host-store.ts
+2
View File
@@ -22,6 +22,7 @@ import {
import { scheduleOrphanedMobileRelayCleanup } from './mobile-relay-orphan-cleanup'
import {
recordDurableHostIdentity,
recordHostProfileMutation,
resetHostProfilePublicationForTests,
retireHostProfilePublication,
serializeHostProfilePublication
@@ -303,6 +304,7 @@ export async function updateHostNameAndEndpoint(
? undefined
: () => updateMobileRelayHostOverlayDirectEndpoint(hostId, updates.endpoint!)
)
recordHostProfileMutation(hostId)
})
}
mobile/src/transport/mobile-relay-pairing-recovery.test.ts
+112 -4
View File
@@ -7,10 +7,17 @@ import {
} from './mobile-relay-pairing-recovery'
import type { PairingCandidateClient } from './mobile-relay-physical-client'
import type { PairingOffer, RpcResponse } from './types'
import {
recordHostProfileMutation,
resetHostProfilePublicationForTests,
retireHostProfilePublication
} from './host-profile-publication'
vi.mock('react-native', () => ({ Platform: { OS: 'ios' } }))
vi.mock('expo-crypto', () => ({ getRandomBytes: vi.fn() }))
vi.mock('expo-secure-store', () => ({ WHEN_UNLOCKED_THIS_DEVICE_ONLY: 'WHEN_UNLOCKED' }))
vi.mock('expo-secure-store', () => ({
WHEN_UNLOCKED_THIS_DEVICE_ONLY: 'WHEN_UNLOCKED'
}))
vi.mock('@react-native-async-storage/async-storage', () => ({ default: {} }))
const now = Date.UTC(2026, 6, 13)
@@ -62,13 +69,19 @@ function endpoints(
relayHostId: offer.relay.relayHostId,
e2eeFraming: 2 as const
},
installStatus: { v: 1 as const, reqId: journal.metadata.installReqId, ...state }
installStatus: {
v: 1 as const,
reqId: journal.metadata.installReqId,
...state
}
}
}
function journal(mode: 'authenticated-direct' | 'relay-basis' = 'authenticated-direct') {
const value = createMobileRelayPairingJournal({
offer: offer as PairingOffer & { relay: NonNullable<PairingOffer['relay']> },
offer: offer as PairingOffer & {
relay: NonNullable<PairingOffer['relay']>
},
hostId: 'host-1',
hostName: 'Blue Whale',
now,
@@ -85,7 +98,10 @@ function journal(mode: 'authenticated-direct' | 'relay-basis' = 'authenticated-d
}
function client(handler: (method: string, params: unknown) => Promise<RpcResponse>) {
return { sendRequest: vi.fn(handler), close: vi.fn() } satisfies PairingCandidateClient
return {
sendRequest: vi.fn(handler),
close: vi.fn()
} satisfies PairingCandidateClient
}
function dependencies(args: {
@@ -116,6 +132,7 @@ function dependencies(args: {
describe('mobile relay pairing recovery', () => {
beforeEach(() => {
resetMobileRelayPairingRecoveryForTests()
resetHostProfilePublicationForTests()
})
it('recovers a lost direct-install response with the pending credential first', async () => {
@@ -224,4 +241,95 @@ describe('mobile relay pairing recovery', () => {
expect(saved.metadata.authorizationMode).toBe('authenticated-direct')
expect(deps.updateJournal).toHaveBeenCalledTimes(2)
})
it('does not resurrect a host removed while startup recovery waits on Relay', async () => {
const saved = journal()
const committed = installed(saved, 'authenticated-direct')
let releaseStatus: () => void = () => {}
const statusPending = new Promise<void>((resolve) => {
releaseStatus = resolve
})
const statusStarted = vi.fn()
const pending = client(async () => {
statusStarted()
await statusPending
return response(endpoints(saved, { state: 'committed', result: committed }))
})
const deps = dependencies({
journal: saved,
connectRelay: vi.fn(() => pending)
})
const recovery = recoverMobileRelayPairing(deps)
await vi.waitFor(() => expect(statusStarted).toHaveBeenCalledOnce())
retireHostProfilePublication(saved.metadata.host.id)
releaseStatus()
await expect(recovery).resolves.toBe('deferred')
expect(deps.writeCredentialBundle).not.toHaveBeenCalled()
expect(deps.saveHost).not.toHaveBeenCalled()
expect(deps.onHostPublished).not.toHaveBeenCalled()
expect(deps.clearJournal).not.toHaveBeenCalled()
})
it('does not overwrite a host edited while startup recovery waits on Relay', async () => {
const saved = journal()
const committed = installed(saved, 'authenticated-direct')
let releaseStatus: () => void = () => {}
const statusPending = new Promise<void>((resolve) => {
releaseStatus = resolve
})
const statusStarted = vi.fn()
const pending = client(async () => {
statusStarted()
await statusPending
return response(endpoints(saved, { state: 'committed', result: committed }))
})
const deps = dependencies({
journal: saved,
connectRelay: vi.fn(() => pending)
})
const recovery = recoverMobileRelayPairing(deps)
await vi.waitFor(() => expect(statusStarted).toHaveBeenCalledOnce())
recordHostProfileMutation(saved.metadata.host.id)
releaseStatus()
await expect(recovery).resolves.toBe('deferred')
expect(deps.writeCredentialBundle).not.toHaveBeenCalled()
expect(deps.saveHost).not.toHaveBeenCalled()
expect(deps.onHostPublished).not.toHaveBeenCalled()
})
it('preserves current durable host fields when retrying a retained journal', async () => {
const saved = journal()
const committed = installed(saved, 'authenticated-direct')
const editedHost = {
...saved.metadata.host,
name: 'Edited name',
endpoint: 'ws://192.168.1.99:6768',
deviceToken: saved.secrets.deviceToken,
lastConnected: 42
}
const pending = client(async () =>
response(endpoints(saved, { state: 'committed', result: committed }))
)
const deps = dependencies({
journal: saved,
connectRelay: vi.fn(() => pending)
})
deps.loadHosts.mockResolvedValue([editedHost])
await expect(recoverMobileRelayPairing(deps)).resolves.toBe('recovered')
expect(deps.saveHost).toHaveBeenCalledWith(
expect.objectContaining({
name: editedHost.name,
endpoint: editedHost.endpoint,
lastConnected: editedHost.lastConnected,
endpoints: expect.arrayContaining([
{ id: 'direct-primary', kind: 'lan', url: editedHost.endpoint }
])
})
)
})
})
mobile/src/transport/mobile-relay-pairing-recovery.ts
+26 -29
View File
@@ -2,8 +2,7 @@ import { Platform } from 'react-native'
import {
DeviceCredentialInstalledSchema,
PairingGetEndpointsResultSchema,
type DeviceCredentialInstalled,
type MobileRelayEndpoint
type DeviceCredentialInstalled
} from '../../../src/shared/mobile-relay-credential-contract'
import type { PairingRelay } from '../../../src/shared/mobile-relay-pairing-offer'
import { loadHosts, saveHost } from './host-store'
@@ -26,7 +25,11 @@ import {
} from './mobile-relay-physical-client'
import { createRecoveringPairingRelayCandidate } from './pairing-relay-candidate'
import type { HostProfile, RpcResponse } from './types'
import { publishHostProfileTransaction } from './host-profile-publication'
import { createMobileRelayRecoveredHost } from './mobile-relay-recovered-host'
import {
getHostProfilePublicationRevision,
publishHostProfileTransaction
} from './host-profile-publication'
export type MobileRelayPairingRecoveryResult = 'none' | 'recovered' | 'deferred'
@@ -90,9 +93,11 @@ async function runRecovery(
if (!journal) {
return 'none'
}
const bundle = await dependencies.readCredentialBundle(journal.metadata.host.id).catch(() => null)
const hostId = journal.metadata.host.id
const publicationRevision = getHostProfilePublicationRevision(hostId)
const bundle = await dependencies.readCredentialBundle(hostId).catch(() => null)
const hosts = await dependencies.loadHosts().catch(() => [])
const existing = hosts.find(({ id }) => id === journal!.metadata.host.id)
const existing = hosts.find(({ id }) => id === hostId)
if (existing?.relayHostId === journal.metadata.relay.relayHostId && bundle) {
await dependencies.clearJournal(journal.metadata.journalId)
return 'recovered'
@@ -116,7 +121,7 @@ async function runRecovery(
})
const endpoints = await getRecoveryStatus(client, journal, credential.kind)
if (endpoints.installStatus?.state === 'committed') {
await publishCommitted(journal, endpoints, dependencies)
await publishCommitted(journal, endpoints, dependencies, publicationRevision, existing)
return 'recovered'
}
if (credential.kind === 'invite' && endpoints.installStatus?.state === 'not-found') {
@@ -131,7 +136,7 @@ async function runRecovery(
)
const reconciled = await getRecoveryStatus(client, journal, 'invite')
assertCommitted(reconciled, installed)
await publishCommitted(journal, reconciled, dependencies)
await publishCommitted(journal, reconciled, dependencies, publicationRevision, existing)
return 'recovered'
}
} catch {
@@ -229,7 +234,9 @@ async function transitionToInviteAuthorization(
async function publishCommitted(
journal: MobileRelayPairingJournal,
endpoints: ReturnType<typeof PairingGetEndpointsResultSchema.parse>,
dependencies: RecoveryDependencies
dependencies: RecoveryDependencies,
publicationRevision: number,
existing: HostProfile | undefined
): Promise<void> {
if (endpoints.installStatus?.state !== 'committed' || !endpoints.relay) {
throw new Error('relay pairing recovery was not committed')
@@ -246,38 +253,28 @@ async function publishCommitted(
if (journal.metadata.authorizationMode !== installed.authorizationMode) {
await dependencies.updateJournal(journal.metadata.journalId, () => reconciledJournal.metadata)
}
const host = relayHost(reconciledJournal, endpoints.relay)
const host = createMobileRelayRecoveredHost(reconciledJournal, endpoints.relay, existing)
await publishHostProfileTransaction(
host,
() =>
dependencies.writeCredentialBundle(
promotePairingJournalCredential({ journal: reconciledJournal, installed })
promotePairingJournalCredential({
journal: reconciledJournal,
installed
})
),
dependencies.saveHost
dependencies.saveHost,
publicationRevision
)
dependencies.onHostPublished(host)
await dependencies.clearJournal(journal.metadata.journalId)
}
function relayHost(journal: MobileRelayPairingJournal, relay: MobileRelayEndpoint): HostProfile {
const host = journal.metadata.host
const url = new URL(relay.cellUrl)
url.protocol = 'wss:'
url.pathname = `/v1/connect/${encodeURIComponent(relay.relayHostId)}`
return {
...host,
deviceToken: journal.secrets.deviceToken,
endpoints: [
{ id: 'direct-primary', kind: 'lan', url: host.endpoint },
{ id: 'relay-primary', kind: 'relay', url: url.toString() }
],
relayHostId: relay.relayHostId,
relay
}
}
function pairingRelay(journal: MobileRelayPairingJournal): PairingRelay {
return { ...journal.metadata.relay, inviteToken: journal.secrets.inviteToken }
return {
...journal.metadata.relay,
inviteToken: journal.secrets.inviteToken
}
}
function requireSuccess(response: RpcResponse): unknown {
mobile/src/transport/mobile-relay-recovered-host.ts
+27
View File
@@ -0,0 +1,27 @@
import type { MobileRelayEndpoint } from '../../../src/shared/mobile-relay-credential-contract'
import type { MobileRelayPairingJournal } from './mobile-relay-pairing-journal'
import { relayWebSocketUrl } from './mobile-endpoint-supervisor-support'
import type { HostProfile } from './types'
export function createMobileRelayRecoveredHost(
journal: MobileRelayPairingJournal,
relay: MobileRelayEndpoint,
existing: HostProfile | undefined
): HostProfile {
const journalHost = journal.metadata.host
const host =
existing?.publicKeyB64 === journalHost.publicKeyB64 &&
existing.deviceToken === journal.secrets.deviceToken
? existing
: journalHost
return {
...host,
deviceToken: journal.secrets.deviceToken,
endpoints: [
{ id: 'direct-primary', kind: 'lan', url: host.endpoint },
{ id: 'relay-primary', kind: 'relay', url: relayWebSocketUrl(relay) }
],
relayHostId: relay.relayHostId,
relay
}
}
mobile/src/transport/mobile-relay-rpc-session-application-responsiveness.test.ts
+130
View File
@@ -0,0 +1,130 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
const fakes = vi.hoisted(() => ({
linkOptions: null as null | {
onHello(value: unknown): void
onAuthenticated(): void
onText(value: string): void
},
sendText: vi.fn(() => true),
close: vi.fn()
}))
vi.mock('./mobile-relay-e2ee-link', () => ({
MobileRelayE2eeLink: class {
constructor(options: NonNullable<typeof fakes.linkOptions>) {
fakes.linkOptions = options
}
sendText = fakes.sendText
close = fakes.close
}
}))
import { connectMobileRelayRpcSession } from './mobile-relay-rpc-session'
const relay = {
v: 1 as const,
directorUrl: 'https://relay.onorca.dev',
cellUrl: 'https://relay-c1.onorca.dev',
assignmentEpoch: 7,
relayHostId: 'AbCdEf0123_-xyZ9',
e2eeFraming: 2 as const
}
async function authenticateSession() {
const session = connectMobileRelayRpcSession({
relay,
resumeToken: 'resume-secret',
resumeCredentialVersion: 3,
resumeConfirmReqId: 'confirm-1',
deviceToken: 'device-token',
desktopPublicKeyB64: 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=',
requestTimeoutMs: 1000
})
fakes.linkOptions!.onHello({
type: 'relay-hello',
ok: true,
credentialKind: 'resume',
leaseExpiresAt: Date.now() + 60_000,
acceptedCredentialVersion: 3,
acceptedAs: 'current',
resumeExpiresAt: Date.now() + 300_000
})
fakes.linkOptions!.onAuthenticated()
await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledOnce())
const request = JSON.parse(fakes.sendText.mock.calls[0]![0] as string) as { id: string }
fakes.linkOptions!.onText(
JSON.stringify({
id: request.id,
ok: true,
result: {
v: 1,
relay,
resumeConfirmation: {
v: 1,
reqId: 'confirm-1',
currentVersion: 3,
acceptedAs: 'current',
renewed: true,
resumeExpiresAt: Date.now() + 300_000
}
},
_meta: { runtimeId: 'runtime-1' }
})
)
await vi.waitFor(() => expect(session.getState()).toBe('connected'))
fakes.sendText.mockClear()
return session
}
describe('mobile relay subscription application responsiveness', () => {
beforeEach(() => {
vi.clearAllMocks()
fakes.linkOptions = null
fakes.sendText.mockReturnValue(true)
})
it('clears an application stall when a subscription answers', 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'
})
expect(session.getRpcUnresponsiveSince?.()).not.toBeNull()
session.subscribe('terminal.subscribe', { terminal: 'term-1' }, vi.fn())
await vi.advanceTimersByTimeAsync(0)
const subscribe = fakes.sendText.mock.calls
.map(([payload]) => JSON.parse(payload as string) as { id: string; method: string })
.find(({ method }) => method === 'terminal.subscribe')!
fakes.linkOptions!.onText(
JSON.stringify({
id: subscribe.id,
ok: true,
streaming: true,
result: { type: 'subscribed', streamId: 42 },
_meta: {}
})
)
expect(session.getRpcUnresponsiveSince?.()).toBeNull()
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('connected')
expect(fakes.close).not.toHaveBeenCalled()
} finally {
session.close()
vi.useRealTimers()
}
})
})
mobile/src/transport/mobile-relay-rpc-session.ts
+2 -1
View File
@@ -52,7 +52,8 @@ export function connectMobileRelayRpcSession(
const streams = new MobileRelayRpcStreams({
nextId,
sendFrame,
waitForConnected: () => waitForConnected()
waitForConnected: () => waitForConnected(),
onApplicationResponse: (method) => applicationResponseTracker.recordResponse(method)
})
const link = new MobileRelayE2eeLink({
mobile/src/transport/mobile-relay-rpc-streams.ts
+14 -2
View File
@@ -30,6 +30,7 @@ type StreamManagerOptions = {
nextId: () => string
sendFrame: (request: { id: string; method: string; params?: unknown }) => boolean
waitForConnected: () => Promise<void>
onApplicationResponse?: (method: string) => void
}
export class MobileRelayRpcStreams {
@@ -62,7 +63,11 @@ export class MobileRelayRpcStreams {
.waitForConnected()
.then(() => {
if (!stream.cancelled) {
stream.sent = this.options.sendFrame({ id, method, params: stream.params })
stream.sent = this.options.sendFrame({
id,
method,
params: stream.params
})
if (!stream.sent) {
this.fail(id, stream, 'Connection interrupted')
}
@@ -90,7 +95,11 @@ export class MobileRelayRpcStreams {
}
const result = (response as RpcSuccess).result
if (result && typeof result === 'object') {
const metadata = result as { subscriptionId?: unknown; streamId?: unknown; type?: unknown }
const metadata = result as {
subscriptionId?: unknown
streamId?: unknown
type?: unknown
}
if (typeof metadata.subscriptionId === 'string') {
stream.subscriptionId = metadata.subscriptionId
}
@@ -119,6 +128,9 @@ export class MobileRelayRpcStreams {
return false
}
stream.controlResponseReceived = true
if (response.ok) {
this.options.onApplicationResponse?.(stream.method)
}
return true
}
mobile/src/transport/rpc-client-application-responsiveness.test.ts
+41 -1
View File
@@ -93,7 +93,9 @@ describe('direct RPC application responsiveness', () => {
.catch((error: unknown) => error)
await vi.advanceTimersByTimeAsync(100)
await expect(first).resolves.toMatchObject({ message: 'Request timed out: browser.screenshot' })
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)
@@ -121,4 +123,42 @@ describe('direct RPC application responsiveness', () => {
expect(client.getState()).toBe('reconnecting')
client.close()
})
it('clears an application stall when a subscription answers', 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'
})
expect(client.getRpcUnresponsiveSince?.()).not.toBeNull()
client.subscribe('terminal.subscribe', { terminal: 'term-1' }, () => {})
const subscribe = sentRequest(socket, 'terminal.subscribe')
socket.receive(
`encrypted:${JSON.stringify({
id: subscribe.id,
ok: true,
streaming: true,
result: { type: 'subscribed', streamId: 42 }
})}`
)
expect(client.getRpcUnresponsiveSince?.()).toBeNull()
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).not.toHaveBeenCalled()
expect(client.getState()).toBe('connected')
client.close()
})
})
mobile/src/transport/rpc-client-context-contract.ts
+6
View File
@@ -1,11 +1,17 @@
import type { RpcClient } from './rpc-client'
import type { MobileConnectionPath } from './stable-logical-rpc-client'
import type { ConnectionState, HostProfile } from './types'
import type { HostProfileEdit } from './host-endpoint-edit'
export type RpcClientContextValue = {
acquire: (hostId: string, host?: HostProfile) => RpcClient | null
release: (hostId: string) => void
forceReconnect: (hostId: string, host?: HostProfile) => Promise<void>
forceReconnectAfterEdit: (
hostId: string,
fallbackHost: HostProfile,
updates: HostProfileEdit
) => Promise<void>
closeHost: (hostId: string) => void
getState: (hostId: string) => ConnectionState
getReconnectAttempt: (hostId: string) => number
mobile/src/transport/rpc-client-stream-frame-routing.ts
+47
View File
@@ -0,0 +1,47 @@
import {
decodeBrowserScreencastFrame,
type BrowserScreencastFrame
} from './browser-screencast-protocol'
import {
handleTerminalBinaryFrame,
type TerminalSnapshotState
} from './rpc-client-terminal-binary-frame'
export type RpcStreamingListener = (result: unknown) => void
export type RpcClientStreamRequest = {
method: string
params: unknown
listener: RpcStreamingListener
onBinaryFrame?: (frame: BrowserScreencastFrame) => void
subscriptionId?: string
cancelled?: boolean
sent?: 'awaiting' | 'received'
}
type StreamFrameRoutingOptions = {
activeBrowserRequestId: string | null
streams: Map<string, RpcClientStreamRequest>
terminalSnapshots: Map<number, TerminalSnapshotState>
terminalListeners: Map<number, RpcStreamingListener>
}
export function routeRpcClientStreamFrame(
bytes: Uint8Array,
options: StreamFrameRoutingOptions
): void {
const browserFrame = decodeBrowserScreencastFrame(bytes)
if (browserFrame) {
const stream = options.activeBrowserRequestId
? options.streams.get(options.activeBrowserRequestId)
: undefined
if (stream && !stream.cancelled && stream.method === 'browser.screencast') {
stream.onBinaryFrame?.(browserFrame)
}
return
}
handleTerminalBinaryFrame(bytes, {
terminalSnapshots: options.terminalSnapshots,
getListener: (streamId) => options.terminalListeners.get(streamId)
})
}
mobile/src/transport/rpc-client.ts
+53 -59
View File
@@ -8,14 +8,12 @@ import {
decrypt,
decryptBytes
} from './e2ee'
import type { TerminalSnapshotState } from './rpc-client-terminal-binary-frame'
import {
handleTerminalBinaryFrame,
type TerminalSnapshotState
} from './rpc-client-terminal-binary-frame'
import {
decodeBrowserScreencastFrame,
type BrowserScreencastFrame
} from './browser-screencast-protocol'
routeRpcClientStreamFrame,
type RpcClientStreamRequest as StreamRequest,
type RpcStreamingListener as StreamingListener
} from './rpc-client-stream-frame-routing'
import type {
ConnectOptions,
RpcClient,
@@ -59,18 +57,6 @@ type ConnectWaiter = {
timeout: ReturnType<typeof setTimeout> | null
}
type StreamingListener = (result: unknown) => void
type StreamRequest = {
method: string
params: unknown
listener: StreamingListener
onBinaryFrame?: (frame: BrowserScreencastFrame) => void
subscriptionId?: string
cancelled?: boolean
sent?: 'awaiting' | 'received'
}
// 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.
@@ -412,7 +398,12 @@ export function connect(
}
resetTerminalStreamRoutingForRequest(id)
if (
sendEncrypted({ id, deviceToken, method: stream.method, params: stream.params })
sendEncrypted({
id,
deviceToken,
method: stream.method,
params: stream.params
})
) {
stream.sent = 'awaiting'
} else {
@@ -421,7 +412,10 @@ export function connect(
}
}
} else if (msg.type === 'e2ee_error' || (!msg.ok && msg.error?.code === 'unauthorized')) {
console.log('[net] e2ee auth FAILED', { msgType: msg.type, error: msg.error })
console.log('[net] e2ee auth FAILED', {
msgType: msg.type,
error: msg.error
})
if (handshakeTimer) {
clearTimeout(handshakeTimer)
handshakeTimer = null
@@ -451,7 +445,12 @@ export function connect(
if (!plaintextBytes) {
return
}
handleBinaryFrame(plaintextBytes)
routeRpcClientStreamFrame(plaintextBytes, {
activeBrowserRequestId: activeBrowserScreencastRequestId,
streams: streamListeners,
terminalSnapshots,
terminalListeners: terminalStreamListeners
})
return
}
@@ -471,11 +470,13 @@ export function connect(
}
const request = pending.get(response.id)
const lateApplicationResponse = applicationResponseTracker.consumeLateResponse(response.id)
const responseStream = streamListeners.get(response.id)
const firstStreamResponse = consumeFirstStreamControlResponse(responseStream)
if (
request ||
timedOutControlRequestIds.consume(response.id) ||
lateApplicationResponse ||
consumeFirstStreamControlResponse(streamListeners.get(response.id))
firstStreamResponse
) {
controlResponseSequence++
}
@@ -485,20 +486,22 @@ export function connect(
handleAuthRejection('Unauthorized — pairing may be revoked')
return
}
if (response.ok && firstStreamResponse && responseStream) {
applicationResponseTracker.recordResponse(responseStream.method)
}
const isStreaming = response.ok && (response as RpcSuccess).streaming === true
if (isStreaming) {
const stream = streamListeners.get(response.id)
if (stream && response.ok) {
if (responseStream && response.ok) {
const result = (response as RpcSuccess).result
if (isStreamingSubscriptionReadyResult(result)) {
stream.subscriptionId = result.subscriptionId
if (stream.cancelled) {
sendServerSubscriptionUnsubscribe(stream)
responseStream.subscriptionId = result.subscriptionId
if (responseStream.cancelled) {
sendServerSubscriptionUnsubscribe(responseStream)
removeStreamListener(response.id)
return
}
if (stream.method === 'browser.screencast') {
if (responseStream.method === 'browser.screencast') {
if (
pendingBrowserScreencastRequestId !== response.id &&
activeBrowserScreencastRequestId !== response.id
@@ -518,10 +521,10 @@ export function connect(
terminalStreamIdsByRequest.set(response.id, ids)
}
ids.add(result.streamId)
terminalStreamListeners.set(result.streamId, stream.listener)
terminalStreamListeners.set(result.streamId, responseStream.listener)
}
if (!stream.cancelled) {
stream.listener(result)
if (!responseStream.cancelled) {
responseStream.listener(result)
}
}
return
@@ -762,7 +765,9 @@ export function connect(
if (controlResponded) {
return
}
console.log('[net] activity-probe TIMEOUT — forcing reconnect', { state })
console.log('[net] activity-probe TIMEOUT — forcing reconnect', {
state
})
forceSocketReconnect(probeWs)
}, 8_000)
pending.set(id, {
@@ -910,29 +915,6 @@ export function connect(
}
}
function handleBinaryFrame(bytes: Uint8Array): void {
const browserFrame = decodeBrowserScreencastFrame(bytes)
if (browserFrame) {
handleBrowserBinaryFrame(browserFrame)
return
}
handleTerminalBinaryFrame(bytes, {
terminalSnapshots,
getListener: (streamId) => terminalStreamListeners.get(streamId)
})
}
function handleBrowserBinaryFrame(frame: BrowserScreencastFrame) {
if (!activeBrowserScreencastRequestId) {
return
}
const stream = streamListeners.get(activeBrowserScreencastRequestId)
if (!stream || stream.cancelled || stream.method !== 'browser.screencast') {
return
}
stream.onBinaryFrame?.(frame)
}
function sendEncrypted(request: unknown): boolean {
if (ws && ws.readyState === WebSocket.OPEN && sharedKey) {
ws.send(encrypt(JSON.stringify(request), sharedKey))
@@ -1014,7 +996,11 @@ export function connect(
const id = nextId()
const timeout = setTimeout(() => {
pending.delete(id)
console.log('[net] sendRequest TIMEOUT', { method, timeoutMs, state })
console.log('[net] sendRequest TIMEOUT', {
method,
timeoutMs,
state
})
// Why: the frame was written 30s ago — the host may have processed it.
reject(markRpcDeliveryUnknown(new Error(`Request timed out: ${method}`)))
if (requestWs === ws) {
@@ -1084,7 +1070,10 @@ export function connect(
}
} else {
// Registered now; the outbound subscribe is (re-)sent once the channel reaches 'connected'.
console.log('[net] subscribe queued — waiting for connected', { method, state })
console.log('[net] subscribe queued — waiting for connected', {
method,
state
})
}
return () => {
@@ -1111,7 +1100,12 @@ export function connect(
} else {
const unsub = buildStreamUnsubscribe(stream?.method, stream?.params)
if (unsub) {
sendEncrypted({ id: nextId(), deviceToken, method: unsub.method, params: unsub.params })
sendEncrypted({
id: nextId(),
deviceToken,
method: unsub.method,
params: unsub.params
})
}
}
removeStreamListener(id)