mirror of
https://github.com/stablyai/orca.git
synced 2026-10-21 16:03:26 +00:00
fix(mobile): honor reconnect profile updates
This commit is contained in:
@@ -1,7 +1,7 @@
|
||||
import { createElement } from 'react'
|
||||
import { act, create, type ReactTestRenderer } from 'react-test-renderer'
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import type { ConnectionState } from './types'
|
||||
import type { ConnectionState, HostProfile } from './types'
|
||||
import type { RpcClient } from './rpc-client'
|
||||
|
||||
const connectMock = vi.fn()
|
||||
@@ -20,7 +20,13 @@ vi.mock('./connection-revival-triggers', () => ({
|
||||
subscribeConnectionRevivalTriggers: () => () => {}
|
||||
}))
|
||||
|
||||
import { RpcClientProvider, useCloseHost, useForceReconnect, useHostClient } from './client-context'
|
||||
import {
|
||||
RpcClientProvider,
|
||||
useCloseHost,
|
||||
useForceReconnect,
|
||||
useHostClient,
|
||||
usePrimeHosts
|
||||
} from './client-context'
|
||||
|
||||
type FakeClient = RpcClient & {
|
||||
emitState: (state: ConnectionState) => void
|
||||
@@ -54,7 +60,7 @@ function makeFakeClient(initialState: ConnectionState): FakeClient {
|
||||
} as FakeClient
|
||||
}
|
||||
|
||||
const HOST = {
|
||||
const HOST: HostProfile = {
|
||||
id: 'host-1',
|
||||
name: 'Host 1',
|
||||
endpoint: 'ws://127.0.0.1:1',
|
||||
@@ -67,6 +73,7 @@ type Harness = {
|
||||
readonly hook: ReturnType<typeof useHostClient>
|
||||
readonly closeHost: (hostId: string) => void
|
||||
readonly forceReconnect: (hostId: string) => Promise<void>
|
||||
readonly primeHosts: (hosts: HostProfile[]) => void
|
||||
readonly unmount: () => void
|
||||
}
|
||||
|
||||
@@ -86,12 +93,14 @@ async function renderHarness(hostId: string): Promise<Harness> {
|
||||
let hook: ReturnType<typeof useHostClient> | null = null
|
||||
let closeHost: ((hostId: string) => void) | null = null
|
||||
let forceReconnect: ((hostId: string) => Promise<void>) | null = null
|
||||
let primeHosts: ((hosts: HostProfile[]) => void) | null = null
|
||||
let renderer: ReactTestRenderer | null = null
|
||||
|
||||
function Probe(): null {
|
||||
hook = useHostClient(hostId)
|
||||
closeHost = useCloseHost()
|
||||
forceReconnect = useForceReconnect()
|
||||
primeHosts = usePrimeHosts()
|
||||
return null
|
||||
}
|
||||
|
||||
@@ -103,7 +112,7 @@ async function renderHarness(hostId: string): Promise<Harness> {
|
||||
} finally {
|
||||
restore()
|
||||
}
|
||||
if (!hook || !closeHost || !forceReconnect || !renderer) {
|
||||
if (!hook || !closeHost || !forceReconnect || !primeHosts || !renderer) {
|
||||
throw new Error('harness did not render')
|
||||
}
|
||||
const mounted = renderer as ReactTestRenderer
|
||||
@@ -126,6 +135,12 @@ async function renderHarness(hostId: string): Promise<Harness> {
|
||||
}
|
||||
return forceReconnect(id)
|
||||
},
|
||||
primeHosts: (hosts) => {
|
||||
if (!primeHosts) {
|
||||
throw new Error('primeHosts hook not rendered')
|
||||
}
|
||||
primeHosts(hosts)
|
||||
},
|
||||
unmount: () => mounted.unmount()
|
||||
}
|
||||
}
|
||||
@@ -314,6 +329,54 @@ describe('useHostClient', () => {
|
||||
harness.unmount()
|
||||
})
|
||||
|
||||
it('queues a changed endpoint behind an older reconnect health probe', async () => {
|
||||
const stale = makeFakeClient('connected')
|
||||
const oldReplacement = makeFakeClient('connecting')
|
||||
const newReplacement = makeFakeClient('connecting')
|
||||
let resolveOldHealthCheck: (() => void) | null = null
|
||||
oldReplacement.sendRequest = vi.fn(
|
||||
() =>
|
||||
new Promise((resolve) => {
|
||||
resolveOldHealthCheck = () => resolve({ id: 'old-health', ok: true, result: {} })
|
||||
})
|
||||
)
|
||||
newReplacement.sendRequest = vi.fn(async () => ({ id: 'new-health', ok: true, result: {} }))
|
||||
connectMock
|
||||
.mockReturnValueOnce(stale)
|
||||
.mockReturnValueOnce(oldReplacement)
|
||||
.mockReturnValueOnce(newReplacement)
|
||||
loadHostsMock.mockResolvedValue([HOST])
|
||||
|
||||
const harness = await renderHarness(HOST.id)
|
||||
const first = harness.forceReconnect(HOST.id)
|
||||
await act(async () => {
|
||||
await Promise.resolve()
|
||||
await Promise.resolve()
|
||||
})
|
||||
|
||||
const updatedHost = { ...HOST, endpoint: 'ws://127.0.0.1:2' }
|
||||
harness.primeHosts([updatedHost])
|
||||
const second = harness.forceReconnect(HOST.id)
|
||||
expect(second).not.toBe(first)
|
||||
expect(connectMock).toHaveBeenCalledTimes(2)
|
||||
|
||||
resolveOldHealthCheck?.()
|
||||
await act(async () => {
|
||||
await Promise.all([first, second])
|
||||
})
|
||||
|
||||
expect(connectMock).toHaveBeenCalledTimes(3)
|
||||
expect(connectMock.mock.calls[2]?.[0]).toEqual(updatedHost)
|
||||
expect(oldReplacement.closeMock).toHaveBeenCalledOnce()
|
||||
expect(newReplacement.sendRequest).toHaveBeenCalledWith('status.get', undefined, {
|
||||
timeoutMs: 15_000,
|
||||
budgetSpansConnect: true,
|
||||
strictDeadline: true
|
||||
})
|
||||
|
||||
harness.unmount()
|
||||
})
|
||||
|
||||
it('rejects Force Reconnect when a replacement client cannot open', async () => {
|
||||
const stale = makeFakeClient('connected')
|
||||
connectMock.mockReturnValue(stale)
|
||||
|
||||
@@ -13,6 +13,7 @@ import {
|
||||
import type { RpcClient } from './rpc-client'
|
||||
import { connectionLogStore } from './connection-log-buffer'
|
||||
import { HostForceReconnectCoordinator, type HostReconnectEntry } from './host-force-reconnect'
|
||||
import { HostReconnectProfileCache } from './host-reconnect-profile-cache'
|
||||
import { subscribeConnectionRevivalTriggers } from './connection-revival-triggers'
|
||||
import { HostClientOpenRegistry } from './host-client-open-registry'
|
||||
import { loadHosts } from './host-store'
|
||||
@@ -54,7 +55,7 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
|
||||
const forceReconnectCoordinatorRef = useRef(new HostForceReconnectCoordinator())
|
||||
|
||||
// Why: cache of already-loaded HostProfiles so openEntry can skip a second loadHosts()/Keychain pass on cold start.
|
||||
const primedHostsRef = useRef<Map<string, HostProfile>>(new Map())
|
||||
const primedHostsRef = useRef(new HostReconnectProfileCache())
|
||||
|
||||
function notifyHostState(hostId: string, state: ConnectionState) {
|
||||
const set = stateListenersRef.current.get(hostId)
|
||||
@@ -163,7 +164,7 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
|
||||
const acquire = useCallback(
|
||||
(hostId: string, host?: HostProfile): RpcClient | null => {
|
||||
if (host) {
|
||||
primedHostsRef.current.set(hostId, host)
|
||||
primedHostsRef.current.prime(host)
|
||||
}
|
||||
const existing = storeRef.current.get(hostId)
|
||||
if (existing) {
|
||||
@@ -184,7 +185,7 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
|
||||
|
||||
const primeHosts = useCallback((hosts: HostProfile[]) => {
|
||||
for (const host of hosts) {
|
||||
primedHostsRef.current.set(host.id, host)
|
||||
primedHostsRef.current.prime(host)
|
||||
}
|
||||
}, [])
|
||||
|
||||
@@ -201,8 +202,9 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
|
||||
(hostId: string): Promise<void> =>
|
||||
forceReconnectCoordinatorRef.current.run({
|
||||
hostId,
|
||||
entry: storeRef.current.get(hostId),
|
||||
listenerCount: stateListenersRef.current.get(hostId)?.size ?? 0,
|
||||
profileVersion: primedHostsRef.current.version(hostId),
|
||||
getEntry: () => storeRef.current.get(hostId),
|
||||
getListenerCount: () => stateListenersRef.current.get(hostId)?.size ?? 0,
|
||||
removeEntry: () => storeRef.current.delete(hostId),
|
||||
openReplacement: () => openEntry(hostId)
|
||||
}),
|
||||
|
||||
@@ -9,24 +9,36 @@ export type HostReconnectEntry = {
|
||||
|
||||
type HostReconnectOperation = {
|
||||
hostId: string
|
||||
entry: HostReconnectEntry | undefined
|
||||
listenerCount: number
|
||||
profileVersion: number
|
||||
getEntry: () => HostReconnectEntry | undefined
|
||||
getListenerCount: () => number
|
||||
removeEntry: () => void
|
||||
openReplacement: () => Promise<HostReconnectEntry | null>
|
||||
}
|
||||
|
||||
type PendingReconnect = {
|
||||
profileVersion: number
|
||||
promise: Promise<void>
|
||||
}
|
||||
|
||||
export class HostForceReconnectCoordinator {
|
||||
private readonly pendingByHost = new Map<string, Promise<void>>()
|
||||
private readonly pendingByHost = new Map<string, PendingReconnect>()
|
||||
|
||||
run(operation: HostReconnectOperation): Promise<void> {
|
||||
const pending = this.pendingByHost.get(operation.hostId)
|
||||
if (pending) {
|
||||
return pending
|
||||
if (pending?.profileVersion === operation.profileVersion) {
|
||||
return pending.promise
|
||||
}
|
||||
const reconnect = this.replaceAndVerify(operation)
|
||||
this.pendingByHost.set(operation.hostId, reconnect)
|
||||
// Why: a changed endpoint must supersede the in-flight profile without racing its health probe.
|
||||
const reconnect = pending
|
||||
? pending.promise.catch(() => undefined).then(() => this.replaceAndVerify(operation))
|
||||
: this.replaceAndVerify(operation)
|
||||
this.pendingByHost.set(operation.hostId, {
|
||||
profileVersion: operation.profileVersion,
|
||||
promise: reconnect
|
||||
})
|
||||
const clearPending = () => {
|
||||
if (this.pendingByHost.get(operation.hostId) === reconnect) {
|
||||
if (this.pendingByHost.get(operation.hostId)?.promise === reconnect) {
|
||||
this.pendingByHost.delete(operation.hostId)
|
||||
}
|
||||
}
|
||||
@@ -35,10 +47,11 @@ export class HostForceReconnectCoordinator {
|
||||
}
|
||||
|
||||
private async replaceAndVerify(operation: HostReconnectOperation): Promise<void> {
|
||||
const savedRefCount = operation.entry?.refCount ?? Math.max(1, operation.listenerCount)
|
||||
if (operation.entry) {
|
||||
operation.entry.unsubState()
|
||||
operation.entry.client.close()
|
||||
const entry = operation.getEntry()
|
||||
const savedRefCount = entry?.refCount ?? Math.max(1, operation.getListenerCount())
|
||||
if (entry) {
|
||||
entry.unsubState()
|
||||
entry.client.close()
|
||||
operation.removeEntry()
|
||||
}
|
||||
const fresh = await operation.openReplacement()
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
import type { HostProfile } from './types'
|
||||
|
||||
type CachedHostProfile = {
|
||||
host: HostProfile
|
||||
version: number
|
||||
}
|
||||
|
||||
export class HostReconnectProfileCache {
|
||||
private readonly profiles = new Map<string, CachedHostProfile>()
|
||||
private readonly latestVersions = new Map<string, number>()
|
||||
|
||||
prime(host: HostProfile): void {
|
||||
const current = this.profiles.get(host.id)
|
||||
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 })
|
||||
}
|
||||
|
||||
get(hostId: string): HostProfile | undefined {
|
||||
return this.profiles.get(hostId)?.host
|
||||
}
|
||||
|
||||
version(hostId: string): number {
|
||||
return this.latestVersions.get(hostId) ?? 0
|
||||
}
|
||||
|
||||
delete(hostId: string): void {
|
||||
this.profiles.delete(hostId)
|
||||
}
|
||||
}
|
||||
|
||||
function reconnectProfileMatches(left: HostProfile, right: HostProfile): boolean {
|
||||
return (
|
||||
left.endpoint === right.endpoint &&
|
||||
left.deviceToken === right.deviceToken &&
|
||||
left.publicKeyB64 === right.publicKeyB64 &&
|
||||
left.relayHostId === right.relayHostId &&
|
||||
JSON.stringify(left.endpoints ?? null) === JSON.stringify(right.endpoints ?? null) &&
|
||||
JSON.stringify(left.relay ?? null) === JSON.stringify(right.relay ?? null)
|
||||
)
|
||||
}
|
||||
Reference in New Issue
Block a user