diff --git a/mobile/src/transport/client-context.test.ts b/mobile/src/transport/client-context.test.ts index ffe65751857..b05bdcbeb48 100644 --- a/mobile/src/transport/client-context.test.ts +++ b/mobile/src/transport/client-context.test.ts @@ -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) diff --git a/mobile/src/transport/client-context.tsx b/mobile/src/transport/client-context.tsx index 9a113697d61..9f5a9e06fcb 100644 --- a/mobile/src/transport/client-context.tsx +++ b/mobile/src/transport/client-context.tsx @@ -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>(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 => + 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] ) diff --git a/mobile/src/transport/host-force-reconnect.ts b/mobile/src/transport/host-force-reconnect.ts new file mode 100644 index 00000000000..b72585e7fcb --- /dev/null +++ b/mobile/src/transport/host-force-reconnect.ts @@ -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 +} + +export class HostForceReconnectCoordinator { + private readonly pendingByHost = new Map>() + + run(operation: HostReconnectOperation): Promise { + 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 { + 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) + } +}