mirror of
https://github.com/stablyai/orca.git
synced 2026-10-21 16:03:26 +00:00
fix(mobile): coalesce force reconnect requests
This commit is contained in:
@@ -282,6 +282,38 @@ describe('useHostClient', () => {
|
||||
harness.unmount()
|
||||
})
|
||||
|
||||
it('coalesces overlapping Force Reconnect calls for one host', async () => {
|
||||
const stale = makeFakeClient('connected')
|
||||
const fresh = makeFakeClient('connecting')
|
||||
let resolveHealthCheck: (() => void) | null = null
|
||||
fresh.sendRequest = vi.fn(
|
||||
() =>
|
||||
new Promise((resolve) => {
|
||||
resolveHealthCheck = () => resolve({ id: 'health', ok: true, result: {} })
|
||||
})
|
||||
)
|
||||
connectMock.mockReturnValueOnce(stale).mockReturnValueOnce(fresh)
|
||||
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 second = harness.forceReconnect(HOST.id)
|
||||
|
||||
expect(second).toBe(first)
|
||||
expect(connectMock).toHaveBeenCalledTimes(2)
|
||||
expect(stale.closeMock).toHaveBeenCalledOnce()
|
||||
expect(fresh.closeMock).not.toHaveBeenCalled()
|
||||
|
||||
resolveHealthCheck?.()
|
||||
await act(async () => Promise.all([first, second]))
|
||||
|
||||
harness.unmount()
|
||||
})
|
||||
|
||||
it('rejects Force Reconnect when a replacement client cannot open', async () => {
|
||||
const stale = makeFakeClient('connected')
|
||||
connectMock.mockReturnValue(stale)
|
||||
|
||||
@@ -12,7 +12,7 @@ import {
|
||||
} from 'react'
|
||||
import type { RpcClient } from './rpc-client'
|
||||
import { connectionLogStore } from './connection-log-buffer'
|
||||
import { verifyForceReconnectRpcHealth } from './force-reconnect-rpc-health'
|
||||
import { HostForceReconnectCoordinator, type HostReconnectEntry } from './host-force-reconnect'
|
||||
import { subscribeConnectionRevivalTriggers } from './connection-revival-triggers'
|
||||
import { HostClientOpenRegistry } from './host-client-open-registry'
|
||||
import { loadHosts } from './host-store'
|
||||
@@ -20,11 +20,8 @@ import { openHostLogicalClient } from './host-logical-client'
|
||||
import type { MobileConnectionPath, StableLogicalRpcClient } from './stable-logical-rpc-client'
|
||||
import type { ConnectionState, HostProfile } from './types'
|
||||
|
||||
type StoreEntry = {
|
||||
client: RpcClient
|
||||
type StoreEntry = HostReconnectEntry & {
|
||||
state: ConnectionState
|
||||
refCount: number
|
||||
unsubState: () => void
|
||||
}
|
||||
|
||||
export type RpcClientContextValue = {
|
||||
@@ -54,6 +51,7 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
|
||||
|
||||
// Pending opens keyed by hostId so two acquire() callers in the same render don't race the host lookup.
|
||||
const pendingOpensRef = useRef(new HostClientOpenRegistry())
|
||||
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())
|
||||
@@ -200,23 +198,14 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
|
||||
}, [])
|
||||
|
||||
const forceReconnect = useCallback(
|
||||
async (hostId: string) => {
|
||||
const entry = storeRef.current.get(hostId)
|
||||
// Why: preserve refcount across the swap; via Disconnect→Reconnect the entry is already gone, so fall back to listener count.
|
||||
const listenerCount = stateListenersRef.current.get(hostId)?.size ?? 0
|
||||
const savedRefCount = entry?.refCount ?? Math.max(1, listenerCount)
|
||||
if (entry) {
|
||||
entry.unsubState()
|
||||
entry.client.close()
|
||||
storeRef.current.delete(hostId)
|
||||
}
|
||||
const fresh = await openEntry(hostId)
|
||||
if (!fresh) {
|
||||
throw new Error('Unable to open a replacement connection')
|
||||
}
|
||||
fresh.refCount = savedRefCount
|
||||
await verifyForceReconnectRpcHealth(fresh.client)
|
||||
},
|
||||
(hostId: string): Promise<void> =>
|
||||
forceReconnectCoordinatorRef.current.run({
|
||||
hostId,
|
||||
entry: storeRef.current.get(hostId),
|
||||
listenerCount: stateListenersRef.current.get(hostId)?.size ?? 0,
|
||||
removeEntry: () => storeRef.current.delete(hostId),
|
||||
openReplacement: () => openEntry(hostId)
|
||||
}),
|
||||
[openEntry]
|
||||
)
|
||||
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
import type { RpcClient } from './rpc-client'
|
||||
import { verifyForceReconnectRpcHealth } from './force-reconnect-rpc-health'
|
||||
|
||||
export type HostReconnectEntry = {
|
||||
client: RpcClient
|
||||
refCount: number
|
||||
unsubState: () => void
|
||||
}
|
||||
|
||||
type HostReconnectOperation = {
|
||||
hostId: string
|
||||
entry: HostReconnectEntry | undefined
|
||||
listenerCount: number
|
||||
removeEntry: () => void
|
||||
openReplacement: () => Promise<HostReconnectEntry | null>
|
||||
}
|
||||
|
||||
export class HostForceReconnectCoordinator {
|
||||
private readonly pendingByHost = new Map<string, Promise<void>>()
|
||||
|
||||
run(operation: HostReconnectOperation): Promise<void> {
|
||||
const pending = this.pendingByHost.get(operation.hostId)
|
||||
if (pending) {
|
||||
return pending
|
||||
}
|
||||
const reconnect = this.replaceAndVerify(operation)
|
||||
this.pendingByHost.set(operation.hostId, reconnect)
|
||||
const clearPending = () => {
|
||||
if (this.pendingByHost.get(operation.hostId) === reconnect) {
|
||||
this.pendingByHost.delete(operation.hostId)
|
||||
}
|
||||
}
|
||||
void reconnect.then(clearPending, clearPending)
|
||||
return reconnect
|
||||
}
|
||||
|
||||
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()
|
||||
operation.removeEntry()
|
||||
}
|
||||
const fresh = await operation.openReplacement()
|
||||
if (!fresh) {
|
||||
throw new Error('Unable to open a replacement connection')
|
||||
}
|
||||
fresh.refCount = savedRefCount
|
||||
await verifyForceReconnectRpcHealth(fresh.client)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user