fix(mobile): fence reconnect lifecycle races

This commit is contained in:
Brennan Benson
2026-08-01 07:42:09 -07:00
parent 314f1f5f6b
commit ebbb6c15c5
19 changed files with 451 additions and 61 deletions
+3 -1
View File
@@ -15,6 +15,7 @@ import { useLocalSearchParams, useRouter } from 'expo-router'
import { ChevronLeft } from 'lucide-react-native'
import { colors, radii, spacing, typography } from '../../../src/theme/mobile-theme'
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 { useForceReconnect, usePrimeHosts } from '../../../src/transport/client-context'
@@ -124,8 +125,9 @@ export default function EditHostScreen() {
// Why: the write already committed above; a re-prime failure here
// must not be reported as a save failure — the next loadHosts() call
// elsewhere in the app picks up the fresh state regardless.
const hostLoadRevision = getHostListLoadRevision()
const hosts = await loadHosts()
primeHosts(hosts)
primeHosts(hosts, hostLoadRevision)
} catch {
// best-effort re-prime; persisted data is unaffected
}
+8 -8
View File
@@ -16,6 +16,7 @@ import {
} from '../src/components/AccountUsage'
import AsyncStorage from '@react-native-async-storage/async-storage'
import { loadHosts } from '../src/transport/host-store'
import { getHostListLoadRevision } from '../src/transport/host-list-load-sharing'
import { navigateToMobileHostEdit } from '../src/transport/host-edit-navigation'
import { removeHostAndCloseClient } from '../src/transport/host-removal-lifecycle'
import { pickResumeWorktree } from '../src/worktree/resume-worktree'
@@ -319,12 +320,6 @@ export default function HomeScreen() {
const closeHostClient = useCloseHost()
const forceReconnectHost = useForceReconnect()
const primeHosts = usePrimeHosts()
// Why: prime the cache with loaded HostProfiles to avoid a second serialized Keychain pass (multi-second connect latency) on cold start.
useEffect(() => {
if (hosts.length > 0) {
primeHosts(hosts)
}
}, [hosts, primeHosts])
const allClientsRef = useRef<Array<{ hostId: string; client: RpcClient }>>([])
// Why: keep the focus callback stable (no refetch per render) while still exposing the latest host clients.
allClientsRef.current = allClients.map((entry) => ({
@@ -374,10 +369,12 @@ export default function HomeScreen() {
useFocusEffect(
useCallback(() => {
let stale = false
const hostLoadRevision = getHostListLoadRevision()
void loadHosts().then(async (h) => {
if (stale) {
return
}
primeHosts(h, hostLoadRevision)
setHosts(h)
if (h.length === 0 || onboardingOptInCheckedRef.current) {
return
@@ -410,7 +407,7 @@ export default function HomeScreen() {
return () => {
stale = true
}
}, [router])
}, [primeHosts, router])
)
const sortedHosts = useMemo(
@@ -670,7 +667,10 @@ export default function HomeScreen() {
try {
await removeHostAndCloseClient(hostToRemove.id, closeHostClient)
setConfirmRemove(null)
setHosts(await loadHosts())
const hostLoadRevision = getHostListLoadRevision()
const remainingHosts = await loadHosts()
primeHosts(remainingHosts, hostLoadRevision)
setHosts(remainingHosts)
} catch {
// Why: ConfirmModal closes on confirm; re-open for retry so the failure isn't silent.
setConfirmRemove(hostToRemove)
+8 -4
View File
@@ -3,7 +3,11 @@ import { act, create, type ReactTestRenderer } from 'react-test-renderer'
import { beforeEach, describe, expect, it, vi } from 'vitest'
import type { ConnectionState, HostProfile } from './types'
import type { RpcClient } from './rpc-client'
import { dropSharedHostListLoad, shareHostListLoad } from './host-list-load-sharing'
import {
dropSharedHostListLoad,
getHostListLoadRevision,
shareHostListLoad
} from './host-list-load-sharing'
const connectMock = vi.fn()
const loadHostsMock = vi.fn()
@@ -74,7 +78,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 primeHosts: (hosts: HostProfile[], sourceRevision?: number) => void
readonly unmount: () => void
}
@@ -136,11 +140,11 @@ async function renderHarness(hostId: string): Promise<Harness> {
}
return forceReconnect(id)
},
primeHosts: (hosts) => {
primeHosts: (hosts, sourceRevision = getHostListLoadRevision()) => {
if (!primeHosts) {
throw new Error('primeHosts hook not rendered')
}
primeHosts(hosts)
primeHosts(hosts, sourceRevision)
},
unmount: () => mounted.unmount()
}
+8 -6
View File
@@ -17,6 +17,7 @@ import { HostReconnectProfileCache, type HostOpenProfile } from './host-reconnec
import { subscribeConnectionRevivalTriggers } from './connection-revival-triggers'
import { HostClientOpenRegistry } from './host-client-open-registry'
import { cancelHostClientOpenProfile, loadHostClientOpenProfile } from './host-client-open-profile'
import { getHostListLoadRevision } from './host-list-load-sharing'
import { loadHosts } from './host-store'
import { openHostLogicalClient } from './host-logical-client'
import { clientActivePath } from './rpc-client-active-path'
@@ -176,10 +177,8 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
[openEntry]
)
const primeHosts = useCallback((hosts: HostProfile[]) => {
for (const host of hosts) {
primedHostsRef.current.prime(host)
}
const primeHosts = useCallback((hosts: HostProfile[], sourceRevision: number) => {
primedHostsRef.current.primeLoadedHosts(hosts, sourceRevision, getHostListLoadRevision())
}, [])
// Why: no idle-close on refcount→0 — transient nav gaps flashed false 'disconnected', so keep sockets alive while foregrounded.
@@ -202,7 +201,10 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
profileVersion: profile.version,
getEntry: () => storeRef.current.get(hostId),
getListenerCount: () => stateListenersRef.current.get(hostId)?.size ?? 0,
removeEntry: () => {
removeEntry: (expected) => {
if (storeRef.current.get(hostId) !== expected) {
return
}
storeRef.current.delete(hostId)
notifyHostState(hostId, 'disconnected')
notifyAllHosts()
@@ -456,6 +458,6 @@ export function useForceReconnect(): (hostId: string) => Promise<void> {
}
// Why: primes already-loaded HostProfiles so the provider can skip a second loadHosts()/Keychain pass on cold start.
export function usePrimeHosts(): (hosts: HostProfile[]) => void {
export function usePrimeHosts(): (hosts: HostProfile[], sourceRevision: number) => void {
return useRpcClientContext().primeHosts
}
@@ -0,0 +1,136 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
import type { MobileRelayCredentialBundle } from './mobile-relay-credential-bundle'
import type { HostProfile } from './types'
const hostStoreMock = vi.hoisted(() => ({ loadStoredHostIdentity: vi.fn() }))
const tokenStoreMock = vi.hoisted(() => ({ readHostDeviceToken: vi.fn() }))
const overlayStoreMock = vi.hoisted(() => ({ saveMobileRelayHostOverlay: vi.fn() }))
vi.mock('./host-store', () => hostStoreMock)
vi.mock('./host-device-token-store', () => tokenStoreMock)
vi.mock('./mobile-relay-host-overlay-store', () => overlayStoreMock)
vi.mock('./host-list-load-sharing', () => ({ dropSharedHostListLoad: vi.fn() }))
import {
MobileRelayUpgradeHostSupersededError,
saveExistingHostRelayRouting,
writeExistingHostRelayCredentialBundle
} from './existing-host-relay-routing'
import { beginHostEndpointPublicationLifecycle } from './host-profile-publication'
const HOST: HostProfile = {
id: 'host-1',
name: 'Host 1',
endpoint: 'ws://127.0.0.1:1',
deviceToken: 'token-1',
publicKeyB64: 'A'.repeat(44),
lastConnected: 0
}
describe('existing host relay endpoint lifecycle publication', () => {
beforeEach(() => {
vi.clearAllMocks()
hostStoreMock.loadStoredHostIdentity.mockResolvedValue(HOST)
tokenStoreMock.readHostDeviceToken.mockResolvedValue(HOST.deviceToken)
overlayStoreMock.saveMobileRelayHostOverlay.mockResolvedValue(undefined)
})
it('rejects a credential write from a retired endpoint lifecycle', async () => {
let releaseOldToken: ((token: string | null) => void) | null = null
tokenStoreMock.readHostDeviceToken.mockImplementationOnce(
() =>
new Promise<string | null>((resolve) => {
releaseOldToken = resolve
})
)
const oldGeneration = beginHostEndpointPublicationLifecycle(HOST.id)
const writes: string[] = []
const oldWrite = writeExistingHostRelayCredentialBundle(
HOST,
credentialBundle(1),
async () => {
writes.push('old')
},
oldGeneration
)
await vi.waitFor(() => expect(tokenStoreMock.readHostDeviceToken).toHaveBeenCalledOnce())
const replacementGeneration = beginHostEndpointPublicationLifecycle(HOST.id)
await writeExistingHostRelayCredentialBundle(
HOST,
credentialBundle(2),
async () => {
writes.push('replacement')
},
replacementGeneration
)
releaseOldToken?.(HOST.deviceToken)
await expect(oldWrite).rejects.toBeInstanceOf(MobileRelayUpgradeHostSupersededError)
expect(writes).toEqual(['replacement'])
})
it('rejects relay routing from a retired endpoint lifecycle', async () => {
let releaseOldToken: ((token: string | null) => void) | null = null
tokenStoreMock.readHostDeviceToken.mockImplementationOnce(
() =>
new Promise<string | null>((resolve) => {
releaseOldToken = resolve
})
)
const oldGeneration = beginHostEndpointPublicationLifecycle(HOST.id)
const stalePublication = saveExistingHostRelayRouting(
relayHostProfile('AbCdEf0123_-old9'),
undefined,
oldGeneration
)
await vi.waitFor(() => expect(tokenStoreMock.readHostDeviceToken).toHaveBeenCalledOnce())
const replacementGeneration = beginHostEndpointPublicationLifecycle(HOST.id)
await saveExistingHostRelayRouting(
relayHostProfile('AbCdEf0123_-new9'),
undefined,
replacementGeneration
)
releaseOldToken?.(HOST.deviceToken)
await expect(stalePublication).rejects.toBeInstanceOf(MobileRelayUpgradeHostSupersededError)
expect(overlayStoreMock.saveMobileRelayHostOverlay).toHaveBeenCalledOnce()
expect(overlayStoreMock.saveMobileRelayHostOverlay).toHaveBeenCalledWith(
expect.objectContaining({ relayHostId: 'AbCdEf0123_-new9' })
)
})
})
function credentialBundle(version: number): MobileRelayCredentialBundle {
return {
v: 1,
hostId: HOST.id,
deviceToken: HOST.deviceToken,
current: {
token: String.fromCharCode(64 + version).repeat(43),
hash: String.fromCharCode(66 + version).repeat(43),
version,
expiresAt: version
}
}
}
function relayHostProfile(relayHostId: string): HostProfile {
return {
...HOST,
endpoints: [
{ id: 'direct-primary', kind: 'lan', url: HOST.endpoint },
{ id: 'relay-primary', kind: 'relay', url: `wss://relay.invalid/${relayHostId}` }
],
relayHostId,
relay: {
v: 1,
directorUrl: 'https://relay.invalid',
cellUrl: 'https://relay.invalid',
assignmentEpoch: 1,
relayHostId,
e2eeFraming: 2
}
}
}
@@ -2,6 +2,7 @@ import { HostProfileSchema, type HostProfile } from './types'
import { loadStoredHostIdentity } from './host-store'
import { readHostDeviceToken } from './host-device-token-store'
import {
getHostEndpointPublicationGeneration,
getHostProfilePublicationRevision,
serializeHostProfilePublication
} from './host-profile-publication'
@@ -14,13 +15,14 @@ export class MobileRelayUpgradeHostSupersededError extends Error {}
export async function saveExistingHostRelayRouting(
host: HostProfile,
beforePublish?: () => Promise<void>
beforePublish?: () => Promise<void>,
endpointGeneration = getHostEndpointPublicationGeneration(host.id)
): Promise<void> {
const validated = HostProfileSchema.parse(host)
const revision = getHostProfilePublicationRevision(validated.id)
await requireCurrentHostCredential(validated)
await serializeHostProfilePublication(validated.id, async () => {
requireUnchangedPublicationRevision(validated.id, revision)
requireCurrentPublication(validated.id, revision, endpointGeneration)
const existing = await requireCurrentHostMetadata(validated)
const { endpoints, relayHostId, relay } = validated
if (!endpoints || !relayHostId || !relay) {
@@ -46,13 +48,14 @@ export async function saveExistingHostRelayRouting(
export async function writeExistingHostRelayCredentialBundle(
host: HostProfile,
bundle: MobileRelayCredentialBundle,
writeBundle: (bundle: MobileRelayCredentialBundle) => Promise<void>
writeBundle: (bundle: MobileRelayCredentialBundle) => Promise<void>,
endpointGeneration = getHostEndpointPublicationGeneration(host.id)
): Promise<void> {
const validated = HostProfileSchema.parse(host)
const revision = getHostProfilePublicationRevision(validated.id)
await requireCurrentHostCredential(validated)
await serializeHostProfilePublication(validated.id, async () => {
requireUnchangedPublicationRevision(validated.id, revision)
requireCurrentPublication(validated.id, revision, endpointGeneration)
await requireCurrentHostMetadata(validated)
if (bundle.hostId !== validated.id || bundle.deviceToken !== validated.deviceToken) {
throw new MobileRelayUpgradeHostSupersededError('mobile relay credential identity mismatch')
@@ -90,8 +93,15 @@ function requireMatchingHostMetadata(
}
}
function requireUnchangedPublicationRevision(hostId: string, revision: number): void {
if (getHostProfilePublicationRevision(hostId) !== revision) {
function requireCurrentPublication(
hostId: string,
revision: number,
endpointGeneration: number
): void {
if (
getHostProfilePublicationRevision(hostId) !== revision ||
getHostEndpointPublicationGeneration(hostId) !== endpointGeneration
) {
throw new MobileRelayUpgradeHostSupersededError('mobile relay upgrade host was re-paired')
}
}
@@ -1,5 +1,6 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import { HostForceReconnectCoordinator } from './host-force-reconnect'
import { HostForceReconnectCoordinator, type HostReconnectEntry } from './host-force-reconnect'
import type { RpcClient } from './rpc-client'
describe('HostForceReconnectCoordinator', () => {
afterEach(() => vi.useRealTimers())
@@ -49,4 +50,44 @@ describe('HostForceReconnectCoordinator', () => {
await vi.advanceTimersByTimeAsync(15_000)
await expect(reconnect).resolves.toBeUndefined()
})
it('retires a replacement that fails RPC health verification', async () => {
const close = vi.fn()
const unsubState = vi.fn()
const client = {
sendRequest: vi.fn(async () => {
throw new Error('health failed')
}),
getState: () => 'connected',
close
} as unknown as RpcClient
const fresh: HostReconnectEntry = { client, refCount: 0, unsubState }
let current: HostReconnectEntry | undefined
const removeEntry = vi.fn((expected: HostReconnectEntry) => {
if (current === expected) {
current = undefined
}
})
const coordinator = new HostForceReconnectCoordinator()
await expect(
coordinator.run({
hostId: 'host-1',
profileVersion: 0,
getEntry: () => current,
getListenerCount: () => 1,
removeEntry,
cancelPendingOpen: vi.fn(),
openReplacement: async () => {
current = fresh
return fresh
}
})
).rejects.toThrow('health failed')
expect(unsubState).toHaveBeenCalledOnce()
expect(close).toHaveBeenCalledOnce()
expect(removeEntry).toHaveBeenCalledWith(fresh)
expect(current).toBeUndefined()
})
})
+15 -4
View File
@@ -15,7 +15,7 @@ type HostReconnectOperation = {
profileVersion: number
getEntry: () => HostReconnectEntry | undefined
getListenerCount: () => number
removeEntry: () => void
removeEntry: (expected: HostReconnectEntry) => void
cancelPendingOpen: () => void
openReplacement: () => Promise<HostReconnectEntry | null>
}
@@ -96,9 +96,7 @@ export class HostForceReconnectCoordinator {
const entry = operation.getEntry()
const savedRefCount = entry?.refCount ?? Math.max(1, operation.getListenerCount())
if (entry) {
entry.unsubState()
entry.client.close()
operation.removeEntry()
this.retireCurrentEntry(operation, entry)
}
const fresh = await this.openBeforeDeadline(operation, deadline, cancelled)
if (this.wasCancelled(operation.hostId, generation)) {
@@ -112,11 +110,24 @@ export class HostForceReconnectCoordinator {
await verifyForceReconnectRpcHealth(fresh.client, deadline)
} catch (error) {
if (!this.wasCancelled(operation.hostId, generation)) {
this.retireCurrentEntry(operation, fresh)
throw error
}
}
}
private retireCurrentEntry(
operation: HostReconnectOperation,
expected: HostReconnectEntry
): void {
if (operation.getEntry() !== expected) {
return
}
expected.unsubState()
expected.client.close()
operation.removeEntry(expected)
}
private async openBeforeDeadline(
operation: HostReconnectOperation,
deadline: number,
@@ -2,11 +2,22 @@ import type { HostProfile } from './types'
const pendingByHost = new Map<string, Promise<void>>()
const revisionByHost = new Map<string, number>()
const endpointGenerationByHost = new Map<string, number>()
export function getHostProfilePublicationRevision(hostId: string): number {
return revisionByHost.get(hostId) ?? 0
}
export function beginHostEndpointPublicationLifecycle(hostId: string): number {
const generation = (endpointGenerationByHost.get(hostId) ?? 0) + 1
endpointGenerationByHost.set(hostId, generation)
return generation
}
export function getHostEndpointPublicationGeneration(hostId: string): number {
return endpointGenerationByHost.get(hostId) ?? 0
}
export function serializeHostProfilePublication<T>(
hostId: string,
publish: () => Promise<T>
@@ -0,0 +1,40 @@
import { describe, expect, it } from 'vitest'
import { HostReconnectProfileCache } from './host-reconnect-profile-cache'
import type { HostProfile } from './types'
const HOST: HostProfile = {
id: 'host-1',
name: 'Host 1',
endpoint: 'ws://127.0.0.1:1',
deviceToken: 'token',
publicKeyB64: 'key',
lastConnected: 0
}
describe('HostReconnectProfileCache', () => {
it('does not let an invalidated host-list snapshot replace 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)
const relayVersion = cache.prime(relayHost)
expect(cache.primeLoaded(HOST, 4, 5)).toBeNull()
expect(cache.get(HOST.id)).toEqual(relayHost)
expect(cache.version(HOST.id)).toBe(relayVersion)
})
})
@@ -25,6 +25,19 @@ export class HostReconnectProfileCache {
return version
}
primeLoaded(host: HostProfile, sourceRevision: number, currentRevision: number): number | null {
if (sourceRevision !== currentRevision) {
return null
}
return this.prime(host)
}
primeLoadedHosts(hosts: HostProfile[], sourceRevision: number, currentRevision: number): void {
for (const host of hosts) {
this.primeLoaded(host, sourceRevision, currentRevision)
}
}
primeFromVersion(host: HostProfile, sourceVersion: number): number | null {
if (this.version(host.id) !== sourceVersion) {
return null
@@ -12,6 +12,7 @@ import {
saveExistingHostRelayRouting,
writeExistingHostRelayCredentialBundle
} from './existing-host-relay-routing'
import { beginHostEndpointPublicationLifecycle } from './host-profile-publication'
import { upgradeDirectMobileRelay } from './mobile-relay-direct-upgrade'
import { MobileRelayDirectUpgradeController } from './mobile-relay-direct-upgrade-controller'
import type { StableLogicalRpcClient } from './stable-logical-rpc-client'
@@ -34,6 +35,7 @@ export function startMobileEndpointLifecycle(
let stopped = false
let foreground = true
let owner: EndpointOwner
const endpointGeneration = beginHostEndpointPublicationLifecycle(initialHost.id)
const publishHostUpdate = (host: HostProfile): void => {
if (!stopped) {
onHostUpdated(host)
@@ -44,7 +46,7 @@ export function startMobileEndpointLifecycle(
if (stopped) {
return
}
const supervisor = createSupervisor(logical, host, onLog, publishHostUpdate)
const supervisor = createSupervisor(logical, host, onLog, publishHostUpdate, endpointGeneration)
owner.stop()
owner = supervisor
supervisor.setForeground(foreground)
@@ -52,7 +54,7 @@ export function startMobileEndpointLifecycle(
}
if (initialHost.relay) {
owner = createSupervisor(logical, initialHost, onLog, publishHostUpdate)
owner = createSupervisor(logical, initialHost, onLog, publishHostUpdate, endpointGeneration)
void owner.start()
} else {
owner = new MobileRelayDirectUpgradeController(logical, initialHost, {
@@ -60,7 +62,11 @@ export function startMobileEndpointLifecycle(
upgradeDirectMobileRelay({
client,
host,
dependencies: { randomBytes: ExpoCrypto.getRandomBytes }
dependencies: {
randomBytes: ExpoCrypto.getRandomBytes,
saveHost: (updatedHost, beforePublish) =>
saveExistingHostRelayRouting(updatedHost, beforePublish, endpointGeneration)
}
}),
onUpgraded: async ({ host }) => {
publishHostUpdate(host)
@@ -86,7 +92,8 @@ function createSupervisor(
logical: StableLogicalRpcClient,
host: HostProfile,
onLog: ConnectionLogSink,
onHostUpdated: (host: HostProfile) => void
onHostUpdated: (host: HostProfile) => void,
endpointGeneration: number
): MobileEndpointSupervisor {
return new MobileEndpointSupervisor(logical, host, {
openDirect: (endpoint) => connect(endpoint, host.deviceToken, host.publicKeyB64, { onLog }),
@@ -102,8 +109,14 @@ function createSupervisor(
resolveRelay: resolveMobileRelayEndpoint,
readBundle: readMobileRelayCredentialBundle,
writeBundle: (bundle) =>
writeExistingHostRelayCredentialBundle(host, bundle, writeMobileRelayCredentialBundle),
saveHost: saveExistingHostRelayRouting,
writeExistingHostRelayCredentialBundle(
host,
bundle,
writeMobileRelayCredentialBundle,
endpointGeneration
),
saveHost: (updatedHost, beforePublish) =>
saveExistingHostRelayRouting(updatedHost, beforePublish, endpointGeneration),
onHostUpdated,
now: Date.now,
randomBytes: ExpoCrypto.getRandomBytes,
@@ -15,7 +15,7 @@ export type MobileEndpointSupervisorDependencies = {
resolveRelay: typeof resolveMobileRelayEndpoint
readBundle: (hostId: string) => Promise<MobileRelayCredentialBundle | null>
writeBundle: (bundle: MobileRelayCredentialBundle) => Promise<void>
saveHost: (host: HostProfile) => Promise<void>
saveHost: (host: HostProfile, beforePublish?: () => Promise<void>) => Promise<void>
onHostUpdated?: (host: HostProfile) => void
now: () => number
randomBytes: (length: number) => Uint8Array
@@ -539,4 +539,40 @@ describe('mobile relay RPC session', () => {
vi.useRealTimers()
}
})
it('queues a fresh Relay timeout probe behind an older in-flight probe', async () => {
const { session } = await authenticateSession()
vi.useFakeTimers()
try {
session.notifyForeground()
const stalled = session.sendRequest('browser.screenshot', {}, { timeoutMs: 100 })
const stalledOutcome = stalled.catch((error: unknown) => error)
const healthy = session.sendRequest('speech.models.list', {}, { timeoutMs: 100 })
await vi.advanceTimersByTimeAsync(0)
const healthyRequest = fakes.sendText.mock.calls
.map(([payload]) => JSON.parse(payload as string) as { id: string; method: string })
.find(({ method }) => method === 'speech.models.list')!
fakes.linkOptions!.onText(
JSON.stringify({ id: healthyRequest.id, ok: true, result: {}, _meta: {} })
)
await expect(healthy).resolves.toMatchObject({ ok: true })
await vi.advanceTimersByTimeAsync(100)
await expect(stalledOutcome).resolves.toMatchObject({
message: 'relay RPC timed out: browser.screenshot'
})
await vi.advanceTimersByTimeAsync(7_900)
const probes = fakes.sendText.mock.calls
.map(([payload]) => JSON.parse(payload as string) as { method: string })
.filter(({ method }) => method === 'status.get')
expect(probes).toHaveLength(2)
expect(session.getState()).toBe('connected')
await vi.advanceTimersByTimeAsync(8_000)
expect(session.getState()).toBe('disconnected')
expect(fakes.close).toHaveBeenCalledOnce()
} finally {
vi.useRealTimers()
}
})
})
@@ -8,6 +8,7 @@ import { MobileRelayRpcStreams } from './mobile-relay-rpc-streams'
import { waitForMobileRelayRpcConnected } from './mobile-relay-rpc-connect-wait'
import { MobileE2EEAuthenticationError } from './mobile-e2ee-v2-physical-channel'
import { isRpcDeliveryUnknown, markRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
import { RpcControlProbeFollowUp } from './rpc-control-probe-follow-up'
import { openRpcRequestBudget, resolvePostConnectRequestTimeout } from './rpc-request-budget'
import { isRpcResponse } from './rpc-response-shape'
import type { RpcClient } from './rpc-client'
@@ -45,7 +46,10 @@ export function connectMobileRelayRpcSession(args: {
let state: ConnectionState = 'connecting'
let requestCounter = 0
let controlResponseSequence = 0
let controlProbeInFlight = false
const controlProbeFollowUp = new RpcControlProbeFollowUp<boolean>(
() => (!closed && state === 'connected' ? true : null),
probeControlPlane
)
let controlProbeTimer: ReturnType<typeof setInterval> | null = null
let lastConnectedAt: number | null = null
let leaseExpiresAt: number | null = null
@@ -168,7 +172,7 @@ export function connectMobileRelayRpcSession(args: {
// Why: the frame was written long ago — the desktop may have processed it.
const error = markRpcDeliveryUnknown(new Error(`relay RPC timed out: ${method}`))
reject(error)
probeControlPlane()
probeControlPlane(true)
}, timeoutMs)
pending.set(id, { resolve, reject, timer })
if (!sendFrame({ id, method, params })) {
@@ -221,11 +225,13 @@ export function connectMobileRelayRpcSession(args: {
streams.handleBinary(bytes)
}
function probeControlPlane(): void {
if (closed || state !== 'connected' || controlProbeInFlight) {
function probeControlPlane(queueAfterCurrent = false): void {
if (closed || state !== 'connected') {
return
}
if (!controlProbeFollowUp.begin(true, queueAfterCurrent)) {
return
}
controlProbeInFlight = true
const probeControlResponseSequence = controlResponseSequence
void sendRpc('status.get', undefined, CONTROL_PROBE_TIMEOUT_MS).then(
() => {
@@ -256,12 +262,12 @@ export function connectMobileRelayRpcSession(args: {
clearInterval(controlProbeTimer)
controlProbeTimer = null
}
finishControlProbe()
controlProbeFollowUp.finish()
}
function finishControlProbe(): void {
controlProbeInFlight = false
timedOutControlRequestIds.clear()
controlProbeFollowUp.finish(true)
}
function waitForConnected(timeoutMs = requestTimeoutMs): Promise<void> {
@@ -14,5 +14,5 @@ export type RpcClientContextValue = {
subscribeHostState: (hostId: string, listener: (state: ConnectionState) => void) => () => void
getAllClients: () => { hostId: string; client: RpcClient }[]
subscribeAllHosts: (listener: () => void) => () => void
primeHosts: (hosts: HostProfile[]) => void
primeHosts: (hosts: HostProfile[], sourceRevision: number) => void
}
@@ -301,6 +301,33 @@ describe('mobile rpc-client request deadline', () => {
client.close()
})
it('queues a fresh timeout probe behind an older in-flight probe', async () => {
const client = connect('ws://desktop.invalid', 'token', 'server-key')
const socket = mockSockets[0]!
socket.open()
client.notifyForeground()
const stalled = client.sendRequest('browser.screenshot', {}, { timeoutMs: 100 })
const stalledOutcome = stalled.catch((error: unknown) => error)
const healthy = client.sendRequest('speech.models.list', {}, { timeoutMs: 100 })
await vi.advanceTimersByTimeAsync(0)
const healthyRequest = sentRequest(socket, 'speech.models.list')
socket.receive(`encrypted:${JSON.stringify({ id: healthyRequest.id, ok: true, result: {} })}`)
await expect(healthy).resolves.toMatchObject({ ok: true })
await vi.advanceTimersByTimeAsync(100)
await expect(stalledOutcome).resolves.toMatchObject({
message: 'Request timed out: browser.screenshot'
})
await vi.advanceTimersByTimeAsync(7_900)
expect(sentRequests(socket, 'status.get')).toHaveLength(2)
expect(client.getState()).toBe('connected')
await vi.advanceTimersByTimeAsync(8_000)
expect(client.getState()).toBe('reconnecting')
expect(socket.close).toHaveBeenCalled()
client.close()
})
it('keeps Force Reconnect inside its deadline after a late connection', async () => {
const client = connect('ws://desktop.invalid', 'token', 'server-key')
const verification = verifyForceReconnectRpcHealth(client)
+18 -16
View File
@@ -34,6 +34,7 @@ import {
RpcSynthesizedCloseIndex
} from './rpc-socket-close-evidence'
import { markRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
import { RpcControlProbeFollowUp } from './rpc-control-probe-follow-up'
import { openRpcRequestBudget, resolvePostConnectRequestTimeout } from './rpc-request-budget'
import { isRpcResponse } from './rpc-response-shape'
import {
@@ -174,7 +175,11 @@ export function connect(
let connectTimer: ReturnType<typeof setTimeout> | null = null
let handshakeTimer: ReturnType<typeof setTimeout> | null = null
let activityProbeTimer: ReturnType<typeof setInterval> | null = null
let activityProbeInFlight = false
const activityProbeFollowUp = new RpcControlProbeFollowUp<WebSocket>(
() => (state === 'connected' ? ws : null),
runActivityProbe,
() => timedOutControlRequestIds.clear()
)
let intentionallyClosed = false
// Consecutive auth rejections; tolerate up to AUTH_RETRY_BUDGET (issue #5200) before latching to avoid a needless re-pair.
let authRejectionCount = 0
@@ -668,7 +673,7 @@ export function connect(
handshakeTimer = null
}
stopActivityProbe()
finishActivityProbe()
activityProbeFollowUp.finish()
if (intentionallyClosed) {
console.log('[net] handleSocketClosed — intentional close')
setState('disconnected')
@@ -779,11 +784,13 @@ export function connect(
}
// Why: stream frames can flow while control RPC is wedged; only a control response satisfies this probe.
function runActivityProbe(expectedWs: WebSocket | null = ws) {
if (state !== 'connected' || !ws || ws !== expectedWs || activityProbeInFlight) {
function runActivityProbe(expectedWs: WebSocket | null = ws, queueAfterCurrent = false): void {
if (state !== 'connected' || !ws || ws !== expectedWs) {
return
}
if (!activityProbeFollowUp.begin(expectedWs, queueAfterCurrent)) {
return
}
activityProbeInFlight = true
const probeWs = ws
const id = nextId()
const probeControlResponseSequence = controlResponseSequence
@@ -792,7 +799,7 @@ export function connect(
timedOut = true
pending.delete(id)
const controlResponded = controlResponseSequence > probeControlResponseSequence
finishActivityProbe()
activityProbeFollowUp.finish(probeWs)
if (controlResponded) {
return
}
@@ -805,28 +812,23 @@ export function connect(
return
}
clearTimeout(timeout)
finishActivityProbe()
activityProbeFollowUp.finish(probeWs)
},
reject: () => {
if (timedOut) {
return
}
clearTimeout(timeout)
finishActivityProbe()
activityProbeFollowUp.finish(probeWs)
}
})
if (!sendEncrypted({ id, deviceToken, method: 'status.get' })) {
clearTimeout(timeout)
pending.delete(id)
finishActivityProbe()
activityProbeFollowUp.finish(probeWs)
}
}
function finishActivityProbe(): void {
activityProbeInFlight = false
timedOutControlRequestIds.clear()
}
function forceSocketReconnect(socket: WebSocket | null): void {
if (!socket || socket !== ws) {
return
@@ -1055,7 +1057,7 @@ export function connect(
if (requestWs === ws) {
timedOutControlRequestIds.add(id)
}
runActivityProbe(requestWs)
runActivityProbe(requestWs, true)
}, timeoutMs)
pending.set(id, {
@@ -1208,7 +1210,7 @@ export function connect(
handshakeTimer = null
}
stopActivityProbe()
finishActivityProbe()
activityProbeFollowUp.finish()
if (ws) {
ws.close()
ws = null
@@ -0,0 +1,36 @@
export class RpcControlProbeFollowUp<T> {
private active = false
private queued: T | null = null
constructor(
private readonly getCurrentTarget: () => T | null,
private readonly launch: (target: T) => void,
private readonly onFinish: () => void = () => {}
) {}
begin(target: T, queueAfterCurrent: boolean): boolean {
if (!this.active) {
this.active = true
return true
}
if (queueAfterCurrent) {
this.queued = target
}
return false
}
finish(target?: T): void {
this.active = false
this.onFinish()
const queued = this.queued
this.queued = null
if (
target !== undefined &&
queued !== null &&
queued === target &&
this.getCurrentTarget() === queued
) {
queueMicrotask(() => this.launch(queued))
}
}
}