diff --git a/mobile/src/transport/client-context.test.ts b/mobile/src/transport/client-context.test.ts index 70006816021..3289ac20ecc 100644 --- a/mobile/src/transport/client-context.test.ts +++ b/mobile/src/transport/client-context.test.ts @@ -511,7 +511,7 @@ describe('useHostClient', () => { expect(connectMock.mock.calls[2]?.[0]).toEqual(updatedHost) expect(oldReplacement.closeMock).toHaveBeenCalledOnce() expect(newReplacement.sendRequest).toHaveBeenCalledWith('status.get', undefined, { - timeoutMs: 15_000, + timeoutMs: expect.any(Number), budgetSpansConnect: true, strictDeadline: true }) diff --git a/mobile/src/transport/existing-host-relay-routing.ts b/mobile/src/transport/existing-host-relay-routing.ts index 91c33afcc79..1995d65a11c 100644 --- a/mobile/src/transport/existing-host-relay-routing.ts +++ b/mobile/src/transport/existing-host-relay-routing.ts @@ -1,28 +1,38 @@ import { HostProfileSchema, type HostProfile } from './types' -import { mutateStoredHosts } from './host-store' +import { loadHosts } from './host-store' +import { serializeHostProfilePublication } from './host-profile-publication' import * as hostListLoads from './host-list-load-sharing' import { saveMobileRelayHostOverlay } from './mobile-relay-host-overlay-store' export class MobileRelayUpgradeHostRemovedError extends Error {} +export class MobileRelayUpgradeHostSupersededError extends Error {} -export async function saveExistingHostRelayRouting(host: HostProfile): Promise { +export async function saveExistingHostRelayRouting( + host: HostProfile, + beforePublish?: () => Promise +): Promise { const validated = HostProfileSchema.parse(host) - await mutateStoredHosts((hosts) => { - const existing = hosts.find(({ id }) => id === validated.id) + await serializeHostProfilePublication(validated.id, async () => { + const existing = (await loadHosts()).find(({ id }) => id === validated.id) if (!existing || existing.publicKeyB64 !== validated.publicKeyB64) { throw new MobileRelayUpgradeHostRemovedError('mobile relay upgrade host was removed') } - return hosts + if (existing.deviceToken !== validated.deviceToken) { + throw new MobileRelayUpgradeHostSupersededError('mobile relay upgrade host was re-paired') + } + const { endpoints, relayHostId, relay } = validated + if (!endpoints || !relayHostId || !relay) { + throw new Error('mobile relay upgrade routing metadata missing') + } + // Why: serialize the credential before its routing overlay so neither side can cross a re-pair. + await beforePublish?.() + await saveMobileRelayHostOverlay({ + v: 2, + hostId: validated.id, + endpoints, + relayHostId, + relay + }) + hostListLoads.dropSharedHostListLoad() }) - if (!validated.endpoints || !validated.relayHostId || !validated.relay) { - throw new Error('mobile relay upgrade routing metadata missing') - } - await saveMobileRelayHostOverlay({ - v: 2, - hostId: validated.id, - endpoints: validated.endpoints, - relayHostId: validated.relayHostId, - relay: validated.relay - }) - hostListLoads.dropSharedHostListLoad() } diff --git a/mobile/src/transport/force-reconnect-rpc-health.ts b/mobile/src/transport/force-reconnect-rpc-health.ts index 9731aa8b0dd..5047657769d 100644 --- a/mobile/src/transport/force-reconnect-rpc-health.ts +++ b/mobile/src/transport/force-reconnect-rpc-health.ts @@ -2,10 +2,12 @@ import type { RpcClient } from './rpc-client' import { isRpcDeliveryUnknown } from './rpc-delivery-ambiguity' import { isLogicalClientCutoverError } from './stable-logical-rpc-client' -const FORCE_RECONNECT_HEALTH_TIMEOUT_MS = 15_000 +export const FORCE_RECONNECT_TIMEOUT_MS = 15_000 -export async function verifyForceReconnectRpcHealth(client: RpcClient): Promise { - const deadline = Date.now() + FORCE_RECONNECT_HEALTH_TIMEOUT_MS +export async function verifyForceReconnectRpcHealth( + client: RpcClient, + deadline = Date.now() + FORCE_RECONNECT_TIMEOUT_MS +): Promise { let lastError: unknown = null for (;;) { const timeoutMs = deadline - Date.now() diff --git a/mobile/src/transport/host-force-reconnect.test.ts b/mobile/src/transport/host-force-reconnect.test.ts new file mode 100644 index 00000000000..0fe5595a7f7 --- /dev/null +++ b/mobile/src/transport/host-force-reconnect.test.ts @@ -0,0 +1,30 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { HostForceReconnectCoordinator } from './host-force-reconnect' + +describe('HostForceReconnectCoordinator', () => { + afterEach(() => vi.useRealTimers()) + + it('bounds a first reconnect when its same-profile open never settles', async () => { + vi.useFakeTimers() + const cancelPendingOpen = vi.fn() + const coordinator = new HostForceReconnectCoordinator() + const reconnect = coordinator.run({ + hostId: 'host-1', + profileVersion: 0, + getEntry: () => undefined, + getListenerCount: () => 1, + removeEntry: vi.fn(), + cancelPendingOpen, + openReplacement: () => new Promise(() => {}) + }) + const outcome = reconnect.catch((error: Error) => error.message) + + expect(cancelPendingOpen).toHaveBeenCalledOnce() + await vi.advanceTimersByTimeAsync(14_999) + expect(cancelPendingOpen).toHaveBeenCalledOnce() + await vi.advanceTimersByTimeAsync(1) + + await expect(outcome).resolves.toBe('Force Reconnect timed out') + expect(cancelPendingOpen).toHaveBeenCalledTimes(2) + }) +}) diff --git a/mobile/src/transport/host-force-reconnect.ts b/mobile/src/transport/host-force-reconnect.ts index fb16671f9bf..c8e687046bb 100644 --- a/mobile/src/transport/host-force-reconnect.ts +++ b/mobile/src/transport/host-force-reconnect.ts @@ -1,5 +1,8 @@ import type { RpcClient } from './rpc-client' -import { verifyForceReconnectRpcHealth } from './force-reconnect-rpc-health' +import { + FORCE_RECONNECT_TIMEOUT_MS, + verifyForceReconnectRpcHealth +} from './force-reconnect-rpc-health' export type HostReconnectEntry = { client: RpcClient @@ -47,9 +50,14 @@ export class HostForceReconnectCoordinator { if (pending) { generation += 1 this.generations.set(operation.hostId, generation) - operation.cancelPendingOpen() } - const reconnect = this.replaceAndVerify(operation, generation) + // Why: a first reconnect must not join an ordinary open whose Keychain read may never settle. + operation.cancelPendingOpen() + const reconnect = this.replaceAndVerify( + operation, + generation, + Date.now() + FORCE_RECONNECT_TIMEOUT_MS + ) this.pendingByHost.set(operation.hostId, { profileVersion: operation.profileVersion, generation, @@ -66,7 +74,8 @@ export class HostForceReconnectCoordinator { private async replaceAndVerify( operation: HostReconnectOperation, - generation: number + generation: number, + deadline: number ): Promise { if (this.wasCancelled(operation.hostId, generation)) { return @@ -78,7 +87,7 @@ export class HostForceReconnectCoordinator { entry.client.close() operation.removeEntry() } - const fresh = await operation.openReplacement() + const fresh = await this.openBeforeDeadline(operation, deadline) if (this.wasCancelled(operation.hostId, generation)) { return } @@ -87,7 +96,7 @@ export class HostForceReconnectCoordinator { } fresh.refCount = savedRefCount try { - await verifyForceReconnectRpcHealth(fresh.client) + await verifyForceReconnectRpcHealth(fresh.client, deadline) } catch (error) { if (!this.wasCancelled(operation.hostId, generation)) { throw error @@ -95,6 +104,32 @@ export class HostForceReconnectCoordinator { } } + private async openBeforeDeadline( + operation: HostReconnectOperation, + deadline: number + ): Promise { + const timeoutMs = deadline - Date.now() + if (timeoutMs <= 0) { + throw new Error('Force Reconnect timed out') + } + let timer: ReturnType | null = null + try { + return await Promise.race([ + operation.openReplacement(), + new Promise((_resolve, reject) => { + timer = setTimeout(() => { + operation.cancelPendingOpen() + reject(new Error('Force Reconnect timed out')) + }, timeoutMs) + }) + ]) + } finally { + if (timer) { + clearTimeout(timer) + } + } + } + private generation(hostId: string): number { return this.generations.get(hostId) ?? 0 } diff --git a/mobile/src/transport/host-profile-publication.ts b/mobile/src/transport/host-profile-publication.ts new file mode 100644 index 00000000000..d0a477aa016 --- /dev/null +++ b/mobile/src/transport/host-profile-publication.ts @@ -0,0 +1,33 @@ +import type { HostProfile } from './types' + +const pendingByHost = new Map>() + +export function serializeHostProfilePublication( + hostId: string, + publish: () => Promise +): Promise { + const previous = pendingByHost.get(hostId) ?? Promise.resolve() + const result = previous.then(publish) + const settled = result.then( + () => undefined, + () => undefined + ) + pendingByHost.set(hostId, settled) + void settled.then(() => { + if (pendingByHost.get(hostId) === settled) { + pendingByHost.delete(hostId) + } + }) + return result +} + +export function publishHostProfileTransaction( + host: HostProfile, + beforeHostSave: (() => Promise) | null, + saveHost: (host: HostProfile) => Promise +): Promise { + return serializeHostProfilePublication(host.id, async () => { + await beforeHostSave?.() + await saveHost(host) + }) +} diff --git a/mobile/src/transport/host-store.test.ts b/mobile/src/transport/host-store.test.ts index dc50e5f8e0c..23a0974f397 100644 --- a/mobile/src/transport/host-store.test.ts +++ b/mobile/src/transport/host-store.test.ts @@ -45,9 +45,11 @@ import { } from './host-store' import { MobileRelayUpgradeHostRemovedError, + MobileRelayUpgradeHostSupersededError, saveExistingHostRelayRouting } from './existing-host-relay-routing' import { resetMobileRelayHostOverlayStoreForTests } from './mobile-relay-host-overlay-store' +import { publishHostProfileTransaction } from './host-profile-publication' const HOSTS_STORAGE_KEY = 'orca:hosts' const OVERLAY_STORAGE_KEY = 'orca:mobile-relay:host-overlays:v2' @@ -286,6 +288,74 @@ describe('host-store list mutations', () => { ) }) + it('rejects relay routing from a lifecycle superseded by same-host re-pairing', async () => { + await saveHost({ ...HOST_ONE, deviceToken: 'replacement-token' }) + const relayHostId = 'AbCdEf0123_-xyZ9' + + await expect( + saveExistingHostRelayRouting({ + ...HOST_ONE, + deviceToken: 'token-1', + endpoints: [ + { id: 'direct-primary', kind: 'lan', url: HOST_ONE.endpoint }, + { id: 'relay-primary', kind: 'relay', url: 'wss://relay.onorca.dev/v1/connect/host' } + ], + relayHostId, + relay: { + v: 1, + directorUrl: 'https://relay.onorca.dev', + cellUrl: 'https://relay.onorca.dev', + assignmentEpoch: 1, + relayHostId, + e2eeFraming: 2 + } + }) + ).rejects.toBeInstanceOf(MobileRelayUpgradeHostSupersededError) + + expect(JSON.parse(storedOverlayRaw ?? '[]')).toEqual([]) + }) + + it('orders an in-flight relay publication before same-host re-pair persistence', async () => { + let releaseOldToken: ((token: string | null) => void) | null = null + secureStoreMock.getItemAsync.mockImplementationOnce( + () => + new Promise((resolve) => { + releaseOldToken = resolve + }) + ) + const relayHostId = 'AbCdEf0123_-xyZ9' + const stalePublication = saveExistingHostRelayRouting({ + ...HOST_ONE, + deviceToken: 'token-1', + endpoints: [ + { id: 'direct-primary', kind: 'lan', url: HOST_ONE.endpoint }, + { id: 'relay-primary', kind: 'relay', url: 'wss://relay.onorca.dev/v1/connect/host' } + ], + relayHostId, + relay: { + v: 1, + directorUrl: 'https://relay.onorca.dev', + cellUrl: 'https://relay.onorca.dev', + assignmentEpoch: 1, + relayHostId, + e2eeFraming: 2 + } + }) + await vi.waitFor(() => expect(secureStoreMock.getItemAsync).toHaveBeenCalledOnce()) + + const replacement = publishHostProfileTransaction( + { ...HOST_ONE, deviceToken: 'replacement-token' }, + null, + saveHost + ) + await Promise.resolve() + expect(secureStoreMock.setItemAsync).not.toHaveBeenCalled() + + releaseOldToken?.('token-1') + await Promise.all([stalePublication, replacement]) + expect(JSON.parse(storedOverlayRaw ?? '[]')).toEqual([]) + }) + it('awaits cleanup scheduling after metadata commit', async () => { let resolveSchedule: (() => void) | null = null scheduleCleanupMock.mockReturnValue( diff --git a/mobile/src/transport/mobile-endpoint-supervisor-support.ts b/mobile/src/transport/mobile-endpoint-supervisor-support.ts index 28318ffd4a8..952e6293af8 100644 --- a/mobile/src/transport/mobile-endpoint-supervisor-support.ts +++ b/mobile/src/transport/mobile-endpoint-supervisor-support.ts @@ -20,14 +20,19 @@ export function relayWebSocketUrl(relay: { cellUrl: string; relayHostId: string export async function persistRelayHost( host: HostProfile, relay: MobileRelayEndpoint, - saveHost: (host: HostProfile) => Promise + saveHost: (host: HostProfile, beforePublish?: () => Promise) => Promise, + beforePublish?: () => Promise ): Promise { const endpoints = [ ...(host.endpoints ?? [{ id: 'direct-primary', kind: 'lan' as const, url: host.endpoint }]) ].filter(({ kind }) => kind !== 'relay') endpoints.push({ id: 'relay-primary', kind: 'relay', url: relayWebSocketUrl(relay) }) const updated = { ...host, endpoints, relayHostId: relay.relayHostId, relay } - await saveHost(updated) + if (beforePublish) { + await saveHost(updated, beforePublish) + } else { + await saveHost(updated) + } return updated } diff --git a/mobile/src/transport/mobile-relay-direct-upgrade.test.ts b/mobile/src/transport/mobile-relay-direct-upgrade.test.ts index 832febfe190..746abc6826a 100644 --- a/mobile/src/transport/mobile-relay-direct-upgrade.test.ts +++ b/mobile/src/transport/mobile-relay-direct-upgrade.test.ts @@ -1,6 +1,9 @@ import { describe, expect, it, vi } from 'vitest' import type { MobileRelayEndpoint } from '../../../src/shared/mobile-relay-credential-contract' -import { MobileRelayUpgradeHostRemovedError } from './existing-host-relay-routing' +import { + MobileRelayUpgradeHostRemovedError, + MobileRelayUpgradeHostSupersededError +} from './existing-host-relay-routing' import { createMobileRelayDirectUpgradeJournal, type MobileRelayDirectUpgradeJournal @@ -64,7 +67,9 @@ function dependencies(journal: MobileRelayDirectUpgradeJournal | null = null) { }), writeBundle: vi.fn(async () => {}), deleteBundle: vi.fn(async () => {}), - saveHost: vi.fn(async () => {}), + saveHost: vi.fn(async (_host: HostProfile, beforePublish?: () => Promise) => { + await beforePublish?.() + }), randomBytes: (length: number) => new Uint8Array(length).fill(7) } } @@ -113,11 +118,37 @@ describe('existing direct pairing relay upgrade', () => { reqId: journal!.reqId, newResumeTokenHash: journal!.pendingResumeTokenHash }) - expect(deps.writeBundle).toHaveBeenCalledBefore(deps.saveHost) + expect(deps.saveHost).toHaveBeenCalledWith(expect.any(Object), expect.any(Function)) + expect(deps.writeBundle).toHaveBeenCalledOnce() expect(result?.host.relay).toEqual(relay) expect(deps.clearJournal).toHaveBeenCalledWith(host.id) }) + it('does not delete a replacement pairing bundle when the old upgrade is superseded', async () => { + const journal = createMobileRelayDirectUpgradeJournal(host.id, (length) => + new Uint8Array(length).fill(6) + ) + const committed = installed(journal) + const deps = dependencies(journal) + deps.saveHost.mockRejectedValue( + new MobileRelayUpgradeHostSupersededError('mobile relay upgrade host was re-paired') + ) + const client = clientWith([ + success({ + v: 1, + relay, + installStatus: { v: 1, reqId: journal.reqId, state: 'committed', result: committed } + }) + ]) + + await expect( + upgradeDirectMobileRelay({ client, host, dependencies: deps }) + ).rejects.toBeInstanceOf(MobileRelayUpgradeHostSupersededError) + expect(deps.writeBundle).not.toHaveBeenCalled() + expect(deps.deleteBundle).not.toHaveBeenCalled() + expect(deps.clearJournal).toHaveBeenCalledWith(host.id) + }) + it('recovers an already committed install without authorizing a second one', async () => { const journal = createMobileRelayDirectUpgradeJournal(host.id, (length) => new Uint8Array(length).fill(3) diff --git a/mobile/src/transport/mobile-relay-direct-upgrade.ts b/mobile/src/transport/mobile-relay-direct-upgrade.ts index a2d907bc8a5..aa390aee4cd 100644 --- a/mobile/src/transport/mobile-relay-direct-upgrade.ts +++ b/mobile/src/transport/mobile-relay-direct-upgrade.ts @@ -7,6 +7,7 @@ import { } from '../../../src/shared/mobile-relay-credential-contract' import { MobileRelayUpgradeHostRemovedError, + MobileRelayUpgradeHostSupersededError, saveExistingHostRelayRouting } from './existing-host-relay-routing' import { persistRelayHost } from './mobile-endpoint-supervisor-support' @@ -118,15 +119,17 @@ async function publishCommitted( expiresAt: installed.resumeExpiresAt } }) - // Why: the overlay must never advertise relay without its matching credential. - await dependencies.writeBundle(bundle) let updatedHost: HostProfile try { - updatedHost = await persistRelayHost(host, endpoints.relay, dependencies.saveHost) + updatedHost = await persistRelayHost(host, endpoints.relay, dependencies.saveHost, () => + dependencies.writeBundle(bundle) + ) } catch (error) { if (error instanceof MobileRelayUpgradeHostRemovedError) { await dependencies.deleteBundle(host.id) await dependencies.clearJournal(host.id) + } else if (error instanceof MobileRelayUpgradeHostSupersededError) { + await dependencies.clearJournal(host.id) } throw error } diff --git a/mobile/src/transport/mobile-relay-pairing-recovery.ts b/mobile/src/transport/mobile-relay-pairing-recovery.ts index 0d5ba7e5e07..c21869fd02d 100644 --- a/mobile/src/transport/mobile-relay-pairing-recovery.ts +++ b/mobile/src/transport/mobile-relay-pairing-recovery.ts @@ -26,6 +26,7 @@ import { } from './mobile-relay-physical-client' import { createRecoveringPairingRelayCandidate } from './pairing-relay-candidate' import type { HostProfile, RpcResponse } from './types' +import { publishHostProfileTransaction } from './host-profile-publication' export type MobileRelayPairingRecoveryResult = 'none' | 'recovered' | 'deferred' @@ -243,10 +244,15 @@ async function publishCommitted( if (journal.metadata.authorizationMode !== installed.authorizationMode) { await dependencies.updateJournal(journal.metadata.journalId, () => reconciledJournal.metadata) } - await dependencies.writeCredentialBundle( - promotePairingJournalCredential({ journal: reconciledJournal, installed }) + const host = relayHost(reconciledJournal, endpoints.relay) + await publishHostProfileTransaction( + host, + () => + dependencies.writeCredentialBundle( + promotePairingJournalCredential({ journal: reconciledJournal, installed }) + ), + dependencies.saveHost ) - await dependencies.saveHost(relayHost(reconciledJournal, endpoints.relay)) await dependencies.clearJournal(journal.metadata.journalId) } diff --git a/mobile/src/transport/mobile-relay-rpc-connect-wait.ts b/mobile/src/transport/mobile-relay-rpc-connect-wait.ts new file mode 100644 index 00000000000..2210307764c --- /dev/null +++ b/mobile/src/transport/mobile-relay-rpc-connect-wait.ts @@ -0,0 +1,32 @@ +import type { ConnectionState } from './types' + +export function waitForMobileRelayRpcConnected(args: { + getState: () => ConnectionState + subscribe: (listener: (state: ConnectionState) => void) => () => void + timeoutMs: number +}): Promise { + const state = args.getState() + if (state === 'connected') { + return Promise.resolve() + } + if (state === 'disconnected' || state === 'auth-failed') { + return Promise.reject(new Error(`relay session ${state}`)) + } + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + unsubscribe() + reject(new Error('relay session connection timed out')) + }, args.timeoutMs) + const unsubscribe = args.subscribe((next) => { + if (next === 'connected') { + clearTimeout(timer) + unsubscribe() + resolve() + } else if (next === 'disconnected' || next === 'auth-failed') { + clearTimeout(timer) + unsubscribe() + reject(new Error(`relay session ${next}`)) + } + }) + }) +} diff --git a/mobile/src/transport/mobile-relay-rpc-session.test.ts b/mobile/src/transport/mobile-relay-rpc-session.test.ts index 406293ed058..82ac52b2f19 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.test.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.test.ts @@ -4,6 +4,8 @@ import { encodeBrowserScreencastFrame } from '../../../src/shared/browser-screencast-protocol' import { encodeTerminalStreamFrame, TerminalStreamOpcode } from './terminal-stream-protocol' +import { verifyForceReconnectRpcHealth } from './force-reconnect-rpc-health' +import { MobileE2EEAuthenticationError } from './mobile-e2ee-v2-physical-channel' import { isRpcDeliveryUnknown } from './rpc-delivery-ambiguity' const fakes = vi.hoisted(() => ({ @@ -335,6 +337,50 @@ describe('mobile relay RPC session', () => { } }) + it('marks a Relay session auth-failed when its periodic probe is unauthorized', async () => { + vi.useFakeTimers() + try { + const { session } = await authenticateSession() + + await vi.advanceTimersByTimeAsync(20_000) + const probe = fakes.sendText.mock.calls + .map(([payload]) => JSON.parse(payload as string) as { id: string; method: string }) + .findLast(({ method }) => method === 'status.get')! + fakes.linkOptions!.onText( + JSON.stringify({ + id: probe.id, + ok: false, + error: { code: 'unauthorized', message: 'Invalid device token' }, + _meta: {} + }) + ) + + expect(session.getFailure()).toBeInstanceOf(MobileE2EEAuthenticationError) + expect(session.getState()).toBe('auth-failed') + expect(fakes.close).toHaveBeenCalledOnce() + } finally { + vi.useRealTimers() + } + }) + + it('rejects Force Reconnect health when Relay authorization was revoked', async () => { + const { session } = await authenticateSession() + const verification = verifyForceReconnectRpcHealth(session) + await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledOnce()) + const probe = JSON.parse(fakes.sendText.mock.calls[0]![0] as string) as { id: string } + fakes.linkOptions!.onText( + JSON.stringify({ + id: probe.id, + ok: false, + error: { code: 'unauthorized', message: 'Invalid device token' }, + _meta: {} + }) + ) + + await expect(verification).rejects.toBeInstanceOf(MobileE2EEAuthenticationError) + expect(session.getState()).toBe('auth-failed') + }) + it('keeps a Relay session when a fresh timeout probe proves it live', async () => { const { session } = await authenticateSession() vi.useFakeTimers() diff --git a/mobile/src/transport/mobile-relay-rpc-session.ts b/mobile/src/transport/mobile-relay-rpc-session.ts index 6b72419f472..e9204acc04d 100644 --- a/mobile/src/transport/mobile-relay-rpc-session.ts +++ b/mobile/src/transport/mobile-relay-rpc-session.ts @@ -5,6 +5,7 @@ import { } from '../../../src/shared/mobile-relay-credential-contract' import { MobileRelayE2eeLink } from './mobile-relay-e2ee-link' 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 { openRpcRequestBudget, resolvePostConnectRequestTimeout } from './rpc-request-budget' @@ -192,6 +193,10 @@ export function connectMobileRelayRpcSession(args: { if (!isRpcResponse(value)) { return } + if (!value.ok && value.error.code === 'unauthorized') { + fail(new MobileE2EEAuthenticationError()) + return + } const request = pending.get(value.id) if (request || timedOutControlRequestIds.delete(value.id)) { controlResponseSequence += 1 @@ -253,30 +258,10 @@ export function connectMobileRelayRpcSession(args: { } function waitForConnected(timeoutMs = requestTimeoutMs): Promise { - if (state === 'connected') { - return Promise.resolve() - } - return new Promise((resolve, reject) => { - let timer: ReturnType | null = null - const unsubscribe = client.onStateChange((next) => { - if (next === 'connected') { - finish() - resolve() - } else if (next === 'disconnected' || next === 'auth-failed') { - finish() - reject(new Error(`relay session ${next}`)) - } - }) - timer = setTimeout(() => { - finish() - reject(new Error('relay session connection timed out')) - }, timeoutMs) - function finish(): void { - if (timer) { - clearTimeout(timer) - } - unsubscribe() - } + return waitForMobileRelayRpcConnected({ + getState: () => state, + subscribe: (listener) => client.onStateChange(listener), + timeoutMs }) } @@ -309,7 +294,9 @@ export function connectMobileRelayRpcSession(args: { // Why: pending entries only exist after their frame reached the authenticated // link (sendFrame failures delete them synchronously), so the desktop may // have processed them — mark the ambiguity for callers. - markRpcDeliveryUnknown(error) + if (!(error instanceof MobileE2EEAuthenticationError)) { + markRpcDeliveryUnknown(error) + } for (const request of pending.values()) { clearTimeout(request.timer) request.reject(error) diff --git a/mobile/src/transport/pre-profile-pairing-coordinator.ts b/mobile/src/transport/pre-profile-pairing-coordinator.ts index 82bd989be03..33bba30055c 100644 --- a/mobile/src/transport/pre-profile-pairing-coordinator.ts +++ b/mobile/src/transport/pre-profile-pairing-coordinator.ts @@ -21,6 +21,7 @@ import { promotePairingJournalCredential, writeMobileRelayCredentialBundle } from './mobile-relay-credential-bundle' +import { publishHostProfileTransaction } from './host-profile-publication' import { connectMobileRelayForPairing, type PairingCandidateClient @@ -28,6 +29,7 @@ import { import { racePairingCandidates, type PairingCandidate } from './pairing-candidate-race' import { resolvePairingInviteThroughDirector } from './mobile-relay-invite-director' import { createRecoveringPairingRelayCandidate } from './pairing-relay-candidate' +import { relayWebSocketUrl } from './mobile-endpoint-supervisor-support' export type PreProfilePairingAttempt = { readonly result: Promise<{ hostId: string }> @@ -190,7 +192,11 @@ async function runPairing( assertActive(isDisposed) if (!journal) { - await dependencies.saveHost(baseHost(offer, hostId, hostName, now)) + await publishHostProfileTransaction( + baseHost(offer, hostId, hostName, now), + null, + dependencies.saveHost + ) return { hostId } } @@ -211,7 +217,11 @@ async function runPairing( if (winner.path !== 'direct') { throw new Error('relay pairing RPC unavailable after relay path authentication') } - await dependencies.saveHost(baseHost(offer, hostId, hostName, now)) + await publishHostProfileTransaction( + baseHost(offer, hostId, hostName, now), + null, + dependencies.saveHost + ) await dependencies.clearJournal(journal.metadata.journalId) return { hostId } } @@ -228,9 +238,17 @@ async function runPairing( throw new Error('desktop returned no relay endpoint after credential install') } assertActive(isDisposed) - await dependencies.writeCredentialBundle(promotePairingJournalCredential({ journal, installed })) - await dependencies.saveHost(relayHost(journal, endpoints.relay)) - await dependencies.clearJournal(journal.metadata.journalId) + const committedJournal = journal + const host = relayHost(committedJournal, endpoints.relay) + await publishHostProfileTransaction( + host, + () => + dependencies.writeCredentialBundle( + promotePairingJournalCredential({ journal: committedJournal, installed }) + ), + dependencies.saveHost + ) + await dependencies.clearJournal(committedJournal.metadata.journalId) return { hostId } } @@ -264,13 +282,6 @@ function relayHost(journal: MobileRelayPairingJournal, relay: MobileRelayEndpoin } } -function relayWebSocketUrl(relay: MobileRelayEndpoint): string { - const url = new URL(relay.cellUrl) - url.protocol = 'wss:' - url.pathname = `/v1/connect/${encodeURIComponent(relay.relayHostId)}` - return url.toString() -} - function requireSuccess(response: RpcResponse): unknown { if (!response.ok) { throw new Error(`${response.error.code}: ${response.error.message}`)