diff --git a/mobile/app/h/[hostId]/edit.tsx b/mobile/app/h/[hostId]/edit.tsx index 5983cd4fc74..2982215c6ea 100644 --- a/mobile/app/h/[hostId]/edit.tsx +++ b/mobile/app/h/[hostId]/edit.tsx @@ -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 } diff --git a/mobile/app/index.tsx b/mobile/app/index.tsx index 7092cb70643..1eba85f4279 100644 --- a/mobile/app/index.tsx +++ b/mobile/app/index.tsx @@ -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>([]) // 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) diff --git a/mobile/src/transport/client-context.test.ts b/mobile/src/transport/client-context.test.ts index ff9158f7e0c..2699c9af40d 100644 --- a/mobile/src/transport/client-context.test.ts +++ b/mobile/src/transport/client-context.test.ts @@ -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 readonly closeHost: (hostId: string) => void readonly forceReconnect: (hostId: string) => Promise - 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 { } 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() } diff --git a/mobile/src/transport/client-context.tsx b/mobile/src/transport/client-context.tsx index 4961a4516df..0b35f88dbf9 100644 --- a/mobile/src/transport/client-context.tsx +++ b/mobile/src/transport/client-context.tsx @@ -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 { } // 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 } diff --git a/mobile/src/transport/existing-host-relay-routing-lifecycle.test.ts b/mobile/src/transport/existing-host-relay-routing-lifecycle.test.ts new file mode 100644 index 00000000000..adc0d32c4c3 --- /dev/null +++ b/mobile/src/transport/existing-host-relay-routing-lifecycle.test.ts @@ -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((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((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 + } + } +} diff --git a/mobile/src/transport/existing-host-relay-routing.ts b/mobile/src/transport/existing-host-relay-routing.ts index fc6674088cc..c832ad90010 100644 --- a/mobile/src/transport/existing-host-relay-routing.ts +++ b/mobile/src/transport/existing-host-relay-routing.ts @@ -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 + beforePublish?: () => Promise, + endpointGeneration = getHostEndpointPublicationGeneration(host.id) ): Promise { 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 + writeBundle: (bundle: MobileRelayCredentialBundle) => Promise, + endpointGeneration = getHostEndpointPublicationGeneration(host.id) ): Promise { 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') } } diff --git a/mobile/src/transport/host-force-reconnect.test.ts b/mobile/src/transport/host-force-reconnect.test.ts index ebe52ae96d3..9f013367756 100644 --- a/mobile/src/transport/host-force-reconnect.test.ts +++ b/mobile/src/transport/host-force-reconnect.test.ts @@ -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() + }) }) diff --git a/mobile/src/transport/host-force-reconnect.ts b/mobile/src/transport/host-force-reconnect.ts index 69b0a3acecd..9b1fa8f0bab 100644 --- a/mobile/src/transport/host-force-reconnect.ts +++ b/mobile/src/transport/host-force-reconnect.ts @@ -15,7 +15,7 @@ type HostReconnectOperation = { profileVersion: number getEntry: () => HostReconnectEntry | undefined getListenerCount: () => number - removeEntry: () => void + removeEntry: (expected: HostReconnectEntry) => void cancelPendingOpen: () => void openReplacement: () => Promise } @@ -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, diff --git a/mobile/src/transport/host-profile-publication.ts b/mobile/src/transport/host-profile-publication.ts index 3d0ed3aed8e..eb800ee5cbb 100644 --- a/mobile/src/transport/host-profile-publication.ts +++ b/mobile/src/transport/host-profile-publication.ts @@ -2,11 +2,22 @@ import type { HostProfile } from './types' const pendingByHost = new Map>() const revisionByHost = new Map() +const endpointGenerationByHost = new Map() 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( hostId: string, publish: () => Promise diff --git a/mobile/src/transport/host-reconnect-profile-cache.test.ts b/mobile/src/transport/host-reconnect-profile-cache.test.ts new file mode 100644 index 00000000000..bd25287ea9a --- /dev/null +++ b/mobile/src/transport/host-reconnect-profile-cache.test.ts @@ -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) + }) +}) diff --git a/mobile/src/transport/host-reconnect-profile-cache.ts b/mobile/src/transport/host-reconnect-profile-cache.ts index 162915d6099..e15e981c626 100644 --- a/mobile/src/transport/host-reconnect-profile-cache.ts +++ b/mobile/src/transport/host-reconnect-profile-cache.ts @@ -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 diff --git a/mobile/src/transport/mobile-endpoint-lifecycle.ts b/mobile/src/transport/mobile-endpoint-lifecycle.ts index b873847bd8f..b01ffb41dad 100644 --- a/mobile/src/transport/mobile-endpoint-lifecycle.ts +++ b/mobile/src/transport/mobile-endpoint-lifecycle.ts @@ -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, diff --git a/mobile/src/transport/mobile-endpoint-supervisor-contract.ts b/mobile/src/transport/mobile-endpoint-supervisor-contract.ts index f5fbd9f017d..11bbfd5d1b9 100644 --- a/mobile/src/transport/mobile-endpoint-supervisor-contract.ts +++ b/mobile/src/transport/mobile-endpoint-supervisor-contract.ts @@ -15,7 +15,7 @@ export type MobileEndpointSupervisorDependencies = { resolveRelay: typeof resolveMobileRelayEndpoint readBundle: (hostId: string) => Promise writeBundle: (bundle: MobileRelayCredentialBundle) => Promise - saveHost: (host: HostProfile) => Promise + saveHost: (host: HostProfile, beforePublish?: () => Promise) => Promise onHostUpdated?: (host: HostProfile) => void now: () => number randomBytes: (length: number) => Uint8Array diff --git a/mobile/src/transport/mobile-relay-rpc-session.test.ts b/mobile/src/transport/mobile-relay-rpc-session.test.ts index 3c60003735f..12c7a379c7c 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.test.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.test.ts @@ -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() + } + }) }) diff --git a/mobile/src/transport/mobile-relay-rpc-session.ts b/mobile/src/transport/mobile-relay-rpc-session.ts index 0e287c331fa..53853469658 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.ts @@ -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( + () => (!closed && state === 'connected' ? true : null), + probeControlPlane + ) let controlProbeTimer: ReturnType | 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 { diff --git a/mobile/src/transport/rpc-client-context-contract.ts b/mobile/src/transport/rpc-client-context-contract.ts index 3a55f8cc98a..92a0ea24ff9 100644 --- a/mobile/src/transport/rpc-client-context-contract.ts +++ b/mobile/src/transport/rpc-client-context-contract.ts @@ -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 } diff --git a/mobile/src/transport/rpc-client-request-deadline.test.ts b/mobile/src/transport/rpc-client-request-deadline.test.ts index 06d758ab7a9..e2943a2002b 100644 --- a/mobile/src/transport/rpc-client-request-deadline.test.ts +++ b/mobile/src/transport/rpc-client-request-deadline.test.ts @@ -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) diff --git a/mobile/src/transport/rpc-client.ts b/mobile/src/transport/rpc-client.ts index 780d7ec3bc4..d58161fede6 100644 --- a/mobile/src/transport/rpc-client.ts +++ b/mobile/src/transport/rpc-client.ts @@ -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 | null = null let handshakeTimer: ReturnType | null = null let activityProbeTimer: ReturnType | null = null - let activityProbeInFlight = false + const activityProbeFollowUp = new RpcControlProbeFollowUp( + () => (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 diff --git a/mobile/src/transport/rpc-control-probe-follow-up.ts b/mobile/src/transport/rpc-control-probe-follow-up.ts new file mode 100644 index 00000000000..56cfe28cec3 --- /dev/null +++ b/mobile/src/transport/rpc-control-probe-follow-up.ts @@ -0,0 +1,36 @@ +export class RpcControlProbeFollowUp { + 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)) + } + } +}