mirror of
https://github.com/stablyai/orca.git
synced 2026-10-20 16:03:20 +00:00
fix(mobile): keep reconnect profiles current
This commit is contained in:
@@ -297,6 +297,100 @@ describe('useHostClient', () => {
|
||||
harness.unmount()
|
||||
})
|
||||
|
||||
it('drops the stale client while Force Reconnect reloads an uncached profile', async () => {
|
||||
const stale = makeFakeClient('connected')
|
||||
const fresh = makeFakeClient('connecting')
|
||||
fresh.sendRequest = vi.fn(async () => ({ id: 'health', ok: true, result: {} }))
|
||||
let resolveHosts: ((hosts: HostProfile[]) => void) | null = null
|
||||
const profileLookup = new Promise<HostProfile[]>((resolve) => {
|
||||
resolveHosts = resolve
|
||||
})
|
||||
connectMock.mockReturnValueOnce(stale).mockReturnValueOnce(fresh)
|
||||
loadHostsMock.mockResolvedValueOnce([HOST]).mockReturnValueOnce(profileLookup)
|
||||
|
||||
const harness = await renderHarness(HOST.id)
|
||||
let reconnect: Promise<void> | null = null
|
||||
await act(async () => {
|
||||
reconnect = harness.forceReconnect(HOST.id)
|
||||
await Promise.resolve()
|
||||
})
|
||||
|
||||
expect(stale.closeMock).toHaveBeenCalledOnce()
|
||||
expect(harness.hook.client).toBeNull()
|
||||
expect(harness.hook.state).toBe('disconnected')
|
||||
expect(connectMock).toHaveBeenCalledOnce()
|
||||
|
||||
await act(async () => {
|
||||
resolveHosts?.([HOST])
|
||||
await reconnect!
|
||||
})
|
||||
expect(harness.hook.client).toBe(fresh)
|
||||
harness.unmount()
|
||||
})
|
||||
|
||||
it('reconnects with relay metadata published by the endpoint lifecycle', async () => {
|
||||
const stale = makeFakeClient('connected')
|
||||
const fresh = makeFakeClient('connecting')
|
||||
fresh.sendRequest = vi.fn(async () => ({ id: 'health', ok: true, result: {} }))
|
||||
connectMock.mockReturnValueOnce(stale).mockReturnValueOnce(fresh)
|
||||
loadHostsMock.mockResolvedValue([HOST])
|
||||
|
||||
const harness = await renderHarness(HOST.id)
|
||||
const relayHostId = 'AbCdEf0123_-xyZ9'
|
||||
const upgradedHost: HostProfile = {
|
||||
...HOST,
|
||||
endpoints: [
|
||||
{ id: 'direct-primary', kind: 'lan', url: HOST.endpoint },
|
||||
{ id: 'relay-primary', kind: 'relay', url: 'wss://relay.invalid/v1/connect/host' }
|
||||
],
|
||||
relayHostId,
|
||||
relay: {
|
||||
v: 1,
|
||||
directorUrl: 'https://relay.invalid',
|
||||
cellUrl: 'https://relay.invalid',
|
||||
assignmentEpoch: 1,
|
||||
relayHostId,
|
||||
e2eeFraming: 2
|
||||
}
|
||||
}
|
||||
const publishHostUpdate = connectMock.mock.calls[0]?.[2] as
|
||||
| ((host: HostProfile) => void)
|
||||
| undefined
|
||||
publishHostUpdate?.(upgradedHost)
|
||||
|
||||
await act(async () => harness.forceReconnect(HOST.id))
|
||||
|
||||
expect(connectMock.mock.calls[1]?.[0]).toEqual(upgradedHost)
|
||||
expect(fresh.sendRequest).toHaveBeenCalledOnce()
|
||||
harness.unmount()
|
||||
})
|
||||
|
||||
it('supersedes a pending old-profile open when Force Reconnect uses a saved endpoint', async () => {
|
||||
let resolveOldLookup: ((hosts: HostProfile[]) => void) | null = null
|
||||
const oldLookup = new Promise<HostProfile[]>((resolve) => {
|
||||
resolveOldLookup = resolve
|
||||
})
|
||||
const fresh = makeFakeClient('connecting')
|
||||
fresh.sendRequest = vi.fn(async () => ({ id: 'health', ok: true, result: {} }))
|
||||
loadHostsMock.mockReturnValue(oldLookup)
|
||||
connectMock.mockReturnValue(fresh)
|
||||
|
||||
const harness = await renderHarness(HOST.id)
|
||||
const updatedHost = { ...HOST, endpoint: 'ws://127.0.0.1:2' }
|
||||
harness.primeHosts([updatedHost])
|
||||
await act(async () => harness.forceReconnect(HOST.id))
|
||||
|
||||
expect(connectMock).toHaveBeenCalledOnce()
|
||||
expect(connectMock.mock.calls[0]?.[0]).toEqual(updatedHost)
|
||||
|
||||
await act(async () => {
|
||||
resolveOldLookup?.([HOST])
|
||||
await oldLookup
|
||||
})
|
||||
expect(connectMock).toHaveBeenCalledOnce()
|
||||
harness.unmount()
|
||||
})
|
||||
|
||||
it('coalesces overlapping Force Reconnect calls for one host', async () => {
|
||||
const stale = makeFakeClient('connected')
|
||||
const fresh = makeFakeClient('connecting')
|
||||
|
||||
@@ -18,8 +18,9 @@ import { subscribeConnectionRevivalTriggers } from './connection-revival-trigger
|
||||
import { HostClientOpenRegistry } from './host-client-open-registry'
|
||||
import { loadHosts } from './host-store'
|
||||
import { openHostLogicalClient } from './host-logical-client'
|
||||
import type { MobileConnectionPath, StableLogicalRpcClient } from './stable-logical-rpc-client'
|
||||
import type { ConnectionState, HostProfile } from './types'
|
||||
import { clientActivePath } from './rpc-client-active-path'
|
||||
import type { MobileConnectionPath } from './stable-logical-rpc-client'
|
||||
import type { ConnectionLogSink, ConnectionState, HostProfile } from './types'
|
||||
|
||||
type StoreEntry = HostReconnectEntry & {
|
||||
state: ConnectionState
|
||||
@@ -86,7 +87,8 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
|
||||
}, [])
|
||||
|
||||
const openEntry = useCallback(async (hostId: string): Promise<StoreEntry | null> => {
|
||||
const existing = pendingOpensRef.current.getActivePromise(hostId)
|
||||
const profileVersion = primedHostsRef.current.version(hostId)
|
||||
const existing = pendingOpensRef.current.getActivePromise(hostId, profileVersion)
|
||||
if (existing) {
|
||||
await existing
|
||||
return storeRef.current.get(hostId) ?? null
|
||||
@@ -95,7 +97,7 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
|
||||
const promise = new Promise<void>((res) => {
|
||||
resolve = res
|
||||
})
|
||||
const pendingOpen = pendingOpensRef.current.register(hostId, promise)
|
||||
const pendingOpen = pendingOpensRef.current.register(hostId, profileVersion, promise)
|
||||
|
||||
try {
|
||||
// Why: prefer the primed cache so we don't serialize a second Keychain pass on cold start.
|
||||
@@ -129,8 +131,11 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
|
||||
}
|
||||
|
||||
let client: RpcClient
|
||||
const onLog: ConnectionLogSink = (entry) => connectionLogStore.append(hostId, entry)
|
||||
try {
|
||||
client = openHostLogicalClient(host, (entry) => connectionLogStore.append(hostId, entry))
|
||||
client = openHostLogicalClient(host, onLog, (updatedHost) =>
|
||||
primedHostsRef.current.prime(updatedHost)
|
||||
)
|
||||
} catch {
|
||||
// Why: openHostLogicalClient can throw synchronously (bad public key / invalid URL); notify so the UI leaves 'connecting'.
|
||||
notifyHostState(hostId, 'disconnected')
|
||||
@@ -206,7 +211,11 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
|
||||
profileVersion: primedHostsRef.current.version(hostId),
|
||||
getEntry: () => storeRef.current.get(hostId),
|
||||
getListenerCount: () => stateListenersRef.current.get(hostId)?.size ?? 0,
|
||||
removeEntry: () => storeRef.current.delete(hostId),
|
||||
removeEntry: () => {
|
||||
storeRef.current.delete(hostId)
|
||||
notifyHostState(hostId, 'disconnected')
|
||||
notifyAllHosts()
|
||||
},
|
||||
openReplacement: () => openEntry(hostId)
|
||||
}),
|
||||
[openEntry]
|
||||
@@ -458,8 +467,3 @@ export function usePrimeHosts(): (hosts: HostProfile[]) => void {
|
||||
const ctx = useRpcClientContext()
|
||||
return ctx.primeHosts
|
||||
}
|
||||
|
||||
function clientActivePath(client: RpcClient | undefined): MobileConnectionPath {
|
||||
const logical = client as Partial<StableLogicalRpcClient> | undefined
|
||||
return typeof logical?.getActivePath === 'function' ? logical.getActivePath() : 'lan'
|
||||
}
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { HostClientOpenRegistry } from './host-client-open-registry'
|
||||
|
||||
describe('HostClientOpenRegistry', () => {
|
||||
it('coalesces only opens for the same profile version', () => {
|
||||
const registry = new HostClientOpenRegistry()
|
||||
const first = Promise.resolve()
|
||||
registry.register('host-1', 1, first)
|
||||
|
||||
expect(registry.getActivePromise('host-1', 1)).toBe(first)
|
||||
expect(registry.getActivePromise('host-1', 2)).toBeNull()
|
||||
})
|
||||
|
||||
it('cancels an older-profile ticket when its replacement registers', () => {
|
||||
const registry = new HostClientOpenRegistry()
|
||||
const first = registry.register('host-1', 1, Promise.resolve())
|
||||
const secondPromise = Promise.resolve()
|
||||
registry.register('host-1', 2, secondPromise)
|
||||
|
||||
expect(first.cancelled).toBe(true)
|
||||
expect(registry.getActivePromise('host-1', 2)).toBe(secondPromise)
|
||||
})
|
||||
})
|
||||
@@ -1,18 +1,25 @@
|
||||
export type HostClientOpenTicket = {
|
||||
cancelled: boolean
|
||||
profileVersion: number
|
||||
promise: Promise<void>
|
||||
}
|
||||
|
||||
export class HostClientOpenRegistry {
|
||||
private readonly pending = new Map<string, HostClientOpenTicket>()
|
||||
|
||||
getActivePromise(hostId: string): Promise<void> | null {
|
||||
getActivePromise(hostId: string, profileVersion: number): Promise<void> | null {
|
||||
const ticket = this.pending.get(hostId)
|
||||
return ticket && !ticket.cancelled ? ticket.promise : null
|
||||
return ticket && !ticket.cancelled && ticket.profileVersion === profileVersion
|
||||
? ticket.promise
|
||||
: null
|
||||
}
|
||||
|
||||
register(hostId: string, promise: Promise<void>): HostClientOpenTicket {
|
||||
const ticket = { cancelled: false, promise }
|
||||
register(hostId: string, profileVersion: number, promise: Promise<void>): HostClientOpenTicket {
|
||||
const previous = this.pending.get(hostId)
|
||||
if (previous) {
|
||||
previous.cancelled = true
|
||||
}
|
||||
const ticket = { cancelled: false, profileVersion, promise }
|
||||
this.pending.set(hostId, ticket)
|
||||
return ticket
|
||||
}
|
||||
|
||||
@@ -5,7 +5,11 @@ import type { ConnectionLogSink, HostProfile } from './types'
|
||||
import { directPathForEndpoint } from './mobile-direct-endpoint-probe'
|
||||
import { startMobileEndpointLifecycle } from './mobile-endpoint-lifecycle'
|
||||
|
||||
export function openHostLogicalClient(host: HostProfile, onLog: ConnectionLogSink): RpcClient {
|
||||
export function openHostLogicalClient(
|
||||
host: HostProfile,
|
||||
onLog: ConnectionLogSink,
|
||||
onHostUpdated: (host: HostProfile) => void = () => {}
|
||||
): RpcClient {
|
||||
// Why: the stable facade owns app-visible RPC/subscription state while the
|
||||
// direct socket remains a replaceable first physical generation.
|
||||
const logical = createStableLogicalRpcClient(
|
||||
@@ -16,7 +20,7 @@ export function openHostLogicalClient(host: HostProfile, onLog: ConnectionLogSin
|
||||
return logical
|
||||
}
|
||||
|
||||
const endpointLifecycle = startMobileEndpointLifecycle(logical, host, onLog)
|
||||
const endpointLifecycle = startMobileEndpointLifecycle(logical, host, onLog, onHostUpdated)
|
||||
endpointLifecycle.setForeground(AppState.currentState === 'active')
|
||||
const appStateSubscription = AppState.addEventListener('change', (state) => {
|
||||
endpointLifecycle.setForeground(state === 'active')
|
||||
|
||||
@@ -25,7 +25,8 @@ type EndpointOwner = EndpointLifecycle & {
|
||||
export function startMobileEndpointLifecycle(
|
||||
logical: StableLogicalRpcClient,
|
||||
initialHost: HostProfile,
|
||||
onLog: ConnectionLogSink
|
||||
onLog: ConnectionLogSink,
|
||||
onHostUpdated: (host: HostProfile) => void = () => {}
|
||||
): EndpointLifecycle {
|
||||
let stopped = false
|
||||
let foreground = true
|
||||
@@ -35,7 +36,7 @@ export function startMobileEndpointLifecycle(
|
||||
if (stopped) {
|
||||
return
|
||||
}
|
||||
const supervisor = createSupervisor(logical, host, onLog)
|
||||
const supervisor = createSupervisor(logical, host, onLog, onHostUpdated)
|
||||
owner.stop()
|
||||
owner = supervisor
|
||||
supervisor.setForeground(foreground)
|
||||
@@ -43,7 +44,7 @@ export function startMobileEndpointLifecycle(
|
||||
}
|
||||
|
||||
if (initialHost.relay) {
|
||||
owner = createSupervisor(logical, initialHost, onLog)
|
||||
owner = createSupervisor(logical, initialHost, onLog, onHostUpdated)
|
||||
void owner.start()
|
||||
} else {
|
||||
owner = new MobileRelayDirectUpgradeController(logical, initialHost, {
|
||||
@@ -53,7 +54,10 @@ export function startMobileEndpointLifecycle(
|
||||
host,
|
||||
dependencies: { randomBytes: ExpoCrypto.getRandomBytes }
|
||||
}),
|
||||
onUpgraded: ({ host }) => startSupervisor(host)
|
||||
onUpgraded: async ({ host }) => {
|
||||
onHostUpdated(host)
|
||||
await startSupervisor(host)
|
||||
}
|
||||
})
|
||||
void owner.start()
|
||||
}
|
||||
@@ -73,7 +77,8 @@ export function startMobileEndpointLifecycle(
|
||||
function createSupervisor(
|
||||
logical: StableLogicalRpcClient,
|
||||
host: HostProfile,
|
||||
onLog: ConnectionLogSink
|
||||
onLog: ConnectionLogSink,
|
||||
onHostUpdated: (host: HostProfile) => void
|
||||
): MobileEndpointSupervisor {
|
||||
return new MobileEndpointSupervisor(logical, host, {
|
||||
openDirect: (endpoint) => connect(endpoint, host.deviceToken, host.publicKeyB64, { onLog }),
|
||||
@@ -90,6 +95,7 @@ function createSupervisor(
|
||||
readBundle: readMobileRelayCredentialBundle,
|
||||
writeBundle: writeMobileRelayCredentialBundle,
|
||||
saveHost,
|
||||
onHostUpdated,
|
||||
now: Date.now,
|
||||
randomBytes: ExpoCrypto.getRandomBytes,
|
||||
setTimer: setTimeout,
|
||||
|
||||
@@ -16,6 +16,7 @@ export type MobileEndpointSupervisorDependencies = {
|
||||
readBundle: (hostId: string) => Promise<MobileRelayCredentialBundle | null>
|
||||
writeBundle: (bundle: MobileRelayCredentialBundle) => Promise<void>
|
||||
saveHost: (host: HostProfile) => Promise<void>
|
||||
onHostUpdated?: (host: HostProfile) => void
|
||||
now: () => number
|
||||
randomBytes: (length: number) => Uint8Array
|
||||
setTimer: typeof setTimeout
|
||||
|
||||
@@ -259,9 +259,11 @@ describe('mobile endpoint supervisor', () => {
|
||||
.mockReturnValueOnce(new FakeRelaySession('disconnected', new RelayOuterError(4409)))
|
||||
.mockReturnValueOnce(new FakeRelaySession('connected'))
|
||||
const resolved = { ...relay, cellUrl: 'https://relay-c2.onorca.dev', assignmentEpoch: 8 }
|
||||
const onHostUpdated = vi.fn()
|
||||
const deps = dependencies({
|
||||
openRelay,
|
||||
resolveRelay: vi.fn(async () => resolved)
|
||||
resolveRelay: vi.fn(async () => resolved),
|
||||
onHostUpdated
|
||||
})
|
||||
const supervisor = new MobileEndpointSupervisor(logical, host, deps)
|
||||
|
||||
@@ -272,6 +274,9 @@ describe('mobile endpoint supervisor', () => {
|
||||
expect(deps.saveHost).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ relay: resolved, endpoint: host.endpoint })
|
||||
)
|
||||
expect(onHostUpdated).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ relay: resolved, endpoint: host.endpoint })
|
||||
)
|
||||
supervisor.stop()
|
||||
})
|
||||
|
||||
|
||||
@@ -180,6 +180,7 @@ export class MobileEndpointSupervisor {
|
||||
resumeToken: credential.token
|
||||
})
|
||||
this.host = await persistRelayHost(this.host, resolved, this.dependencies.saveHost)
|
||||
this.dependencies.onHostUpdated?.(this.host)
|
||||
return await this.openAndMigrateRelay(credential)
|
||||
} catch (error) {
|
||||
return { ok: false, error: toError(error) }
|
||||
@@ -303,6 +304,7 @@ export class MobileEndpointSupervisor {
|
||||
// Why: a scheduled rotation can finish after the old credential enters the rejection gate.
|
||||
credentialRefreshed = true
|
||||
this.host = await persistRelayHost(this.host, result.relay, this.dependencies.saveHost)
|
||||
this.dependencies.onHostUpdated?.(this.host)
|
||||
} catch {
|
||||
// Why: pending material remains durable; the next authenticated direct
|
||||
// opportunity must reconcile it before creating another install key.
|
||||
|
||||
@@ -0,0 +1,7 @@
|
||||
import type { RpcClient } from './rpc-client'
|
||||
import type { MobileConnectionPath, StableLogicalRpcClient } from './stable-logical-rpc-client'
|
||||
|
||||
export function clientActivePath(client: RpcClient | undefined): MobileConnectionPath {
|
||||
const logical = client as Partial<StableLogicalRpcClient> | undefined
|
||||
return typeof logical?.getActivePath === 'function' ? logical.getActivePath() : 'lan'
|
||||
}
|
||||
Reference in New Issue
Block a user