feat(mobile): race direct and relay pairing

Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
Jinwoo-H
2026-07-13 01:37:47 -04:00
co-authored by Orca
parent de47a6125b
commit 9a7dfec237
7 changed files with 572 additions and 10 deletions
@@ -0,0 +1,135 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
const fakes = vi.hoisted(() => ({
channelOptions: null as null | {
onAuthenticated(): void
onText(value: string): void
},
start: vi.fn(),
handleMessage: vi.fn(),
sendText: vi.fn(() => true),
dispose: vi.fn()
}))
vi.mock('./mobile-e2ee-v2-client-session', () => ({
MobileE2EEV2ClientSession: { create: vi.fn(() => ({ hello: {} })) }
}))
vi.mock('./mobile-e2ee-v2-physical-channel', () => ({
MobileE2EEV2PhysicalChannel: class {
constructor(options: NonNullable<typeof fakes.channelOptions>) {
fakes.channelOptions = options
}
start = fakes.start
handleMessage = fakes.handleMessage
sendText = fakes.sendText
dispose = fakes.dispose
}
}))
import { connectMobileRelayForPairing, RelayOuterError } from './mobile-relay-physical-client'
class FakeSocket {
readonly OPEN = 1
readyState = 1
bufferedAmount = 0
sent: unknown[] = []
onopen: (() => void) | null = null
onmessage: ((event: { data: unknown }) => void) | null = null
onerror: (() => void) | null = null
onclose: ((event: { code: number }) => void) | null = null
send(value: unknown): void {
this.sent.push(value)
}
close(): void {}
receive(data: unknown): void {
this.onmessage?.({ data })
}
}
const relay = {
v: 1 as const,
directorUrl: 'https://relay.onorca.dev',
cellUrl: 'https://relay-c1.onorca.dev',
assignmentEpoch: 7,
relayHostId: 'AbCdEf0123_-xyZ9',
inviteToken: 'abcdefghijklmnopqrstuvwxyzABCDEFGH012345678',
inviteExpiresAt: Date.now() + 300_000,
e2eeFraming: 2 as const
}
describe('mobile relay physical pairing client', () => {
beforeEach(() => {
vi.clearAllMocks()
fakes.channelOptions = null
fakes.sendText.mockReturnValue(true)
})
it('uses first-frame outer auth, waits for host attach, then carries RPC over E2EE v2', async () => {
const socket = new FakeSocket()
let openedUrl = ''
const client = connectMobileRelayForPairing({
relay,
deviceToken: 'device-token',
desktopPublicKeyB64: 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=',
createSocket: (url) => {
openedUrl = url
return socket as unknown as WebSocket
}
})
socket.onopen?.()
expect(openedUrl).toBe('wss://relay-c1.onorca.dev/v1/connect/AbCdEf0123_-xyZ9')
expect(openedUrl).not.toContain('?')
expect(JSON.parse(socket.sent[0] as string)).toEqual({
type: 'relay-auth',
v: 1,
mode: 'connect',
credential: relay.inviteToken
})
socket.receive(
JSON.stringify({
type: 'relay-hello',
ok: true,
credentialKind: 'invite',
leaseExpiresAt: Date.now() + 60_000
})
)
await vi.waitFor(() => expect(fakes.start).toHaveBeenCalledOnce())
fakes.channelOptions!.onAuthenticated()
const responsePromise = client.sendRequest('status.get')
await vi.waitFor(() => expect(fakes.sendText).toHaveBeenCalledOnce())
const request = JSON.parse(fakes.sendText.mock.calls[0]![0] as string)
expect(request).toEqual({
id: 'relay-pair-1',
deviceToken: 'device-token',
method: 'status.get'
})
fakes.channelOptions!.onText(
JSON.stringify({
id: request.id,
ok: true,
result: { path: 'relay' },
_meta: { runtimeId: 'runtime-1' }
})
)
await expect(responsePromise).resolves.toMatchObject({ ok: true, result: { path: 'relay' } })
})
it('surfaces a typed endpoint-scoped outer rejection before E2EE', async () => {
const socket = new FakeSocket()
const client = connectMobileRelayForPairing({
relay,
deviceToken: 'device-token',
desktopPublicKeyB64: 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=',
createSocket: () => socket as unknown as WebSocket
})
const status = client.sendRequest('status.get')
socket.receive(JSON.stringify({ type: 'relay-hello', ok: false, code: 4404 }))
await expect(status).rejects.toEqual(new RelayOuterError(4404))
expect(fakes.start).not.toHaveBeenCalled()
})
})
@@ -0,0 +1,184 @@
import type { PairingRelay } from '../../../src/shared/mobile-relay-pairing-offer'
import { RelayPhoneHelloSchema } from '../../../src/shared/mobile-relay-phone-protocol'
import { MobileE2EEV2ClientSession } from './mobile-e2ee-v2-client-session'
import { MobileE2EEV2PhysicalChannel } from './mobile-e2ee-v2-physical-channel'
import { isRpcResponse } from './rpc-response-shape'
import type { RpcResponse } from './types'
import { websocketPayloadToUint8 } from './websocket-payload-bytes'
type PendingRequest = {
resolve: (response: RpcResponse) => void
reject: (error: Error) => void
timer: ReturnType<typeof setTimeout>
}
export type PairingCandidateClient = {
sendRequest(method: string, params?: unknown): Promise<RpcResponse>
close(): void
}
export class RelayOuterError extends Error {
constructor(readonly code: number) {
super(`relay_outer_${code}`)
}
}
export function connectMobileRelayForPairing(args: {
relay: PairingRelay
deviceToken: string
desktopPublicKeyB64: string
requestTimeoutMs?: number
createSocket?: (url: string) => WebSocket
}): PairingCandidateClient {
const requestTimeoutMs = args.requestTimeoutMs ?? 30_000
const socketUrl = relayPhoneWebSocketUrl(args.relay)
const socket = (args.createSocket ?? ((url) => new WebSocket(url)))(socketUrl)
const session = MobileE2EEV2ClientSession.create({
desktopPublicKeyB64: args.desktopPublicKeyB64,
transport: 'relay',
relayHostId: args.relay.relayHostId
})
const pending = new Map<string, PendingRequest>()
let requestCounter = 0
let closed = false
let outerReady = false
let authenticated = false
let resolveAuthenticated!: () => void
let rejectAuthenticated!: (error: Error) => void
const authenticatedPromise = new Promise<void>((resolve, reject) => {
resolveAuthenticated = resolve
rejectAuthenticated = reject
})
const channel = new MobileE2EEV2PhysicalChannel({
session,
socket,
deviceToken: args.deviceToken,
decodeBinary: websocketPayloadToUint8,
onAuthenticated: () => {
authenticated = true
resolveAuthenticated()
},
onText: (plaintext) => {
let value: unknown
try {
value = JSON.parse(plaintext)
} catch {
return
}
if (!isRpcResponse(value)) {
return
}
const request = pending.get(value.id)
if (request) {
clearTimeout(request.timer)
pending.delete(value.id)
request.resolve(value)
}
},
onBinary: () => {},
onError: fail
})
socket.onopen = () => {
socket.send(
JSON.stringify({
type: 'relay-auth',
v: 1,
mode: 'connect',
credential: args.relay.inviteToken
})
)
}
let inboundChain: Promise<void> = Promise.resolve()
socket.onmessage = (event) => {
inboundChain = inboundChain
.then(async () => {
if (closed) {
return
}
if (!outerReady) {
acceptRelayHello(event.data)
return
}
await channel.handleMessage(event.data)
})
.catch((error: unknown) => fail(asError(error)))
}
socket.onerror = () => fail(new Error('relay transport error'))
socket.onclose = (event) => fail(new RelayOuterError(event.code || 1006))
function acceptRelayHello(raw: unknown): void {
if (typeof raw !== 'string') {
throw new Error('expected plaintext relay hello')
}
let value: unknown
try {
value = JSON.parse(raw)
} catch {
throw new Error('invalid relay hello JSON')
}
const parsed = RelayPhoneHelloSchema.safeParse(value)
if (!parsed.success) {
throw new Error('invalid relay hello')
}
if (!parsed.data.ok) {
throw new RelayOuterError(parsed.data.code)
}
if (parsed.data.credentialKind !== 'invite') {
throw new Error('pairing invite resolved as an unexpected credential kind')
}
outerReady = true
channel.start()
}
function fail(error: Error): void {
if (closed) {
return
}
closed = true
channel.dispose()
rejectAuthenticated(error)
for (const request of pending.values()) {
clearTimeout(request.timer)
request.reject(error)
}
pending.clear()
socket.close()
}
return {
async sendRequest(method, params) {
await authenticatedPromise
if (closed || !authenticated) {
throw new Error('relay pairing client closed')
}
const id = `relay-pair-${++requestCounter}`
return new Promise<RpcResponse>((resolve, reject) => {
const timer = setTimeout(() => {
pending.delete(id)
reject(new Error(`relay pairing RPC timed out: ${method}`))
}, requestTimeoutMs)
pending.set(id, { resolve, reject, timer })
if (
!channel.sendText(JSON.stringify({ id, deviceToken: args.deviceToken, method, params }))
) {
clearTimeout(timer)
pending.delete(id)
reject(new Error('relay E2EE channel not ready'))
}
})
},
close: () => fail(new Error('relay pairing client closed'))
}
}
export function relayPhoneWebSocketUrl(relay: PairingRelay): string {
const url = new URL(relay.cellUrl)
url.protocol = 'wss:'
url.pathname = `/v1/connect/${encodeURIComponent(relay.relayHostId)}`
return url.toString()
}
function asError(error: unknown): Error {
return error instanceof Error ? error : new Error(String(error))
}
@@ -0,0 +1,59 @@
import type { PairingCandidateClient } from './mobile-relay-physical-client'
export type PairingCandidate = {
path: 'direct' | 'relay'
client: PairingCandidateClient
}
export function racePairingCandidates(
candidates: readonly PairingCandidate[]
): Promise<PairingCandidate> {
return new Promise((resolve, reject) => {
const successes: PairingCandidate[] = []
let failures = 0
let settled = false
let selectionQueued = false
for (const candidate of candidates) {
void candidate.client.sendRequest('status.get').then(
(response) => {
if (!response.ok) {
failures++
rejectIfFinished()
return
}
successes.push(candidate)
if (selectionQueued) {
return
}
selectionQueued = true
// Why: defer one microtask so simultaneous successes are visible and
// direct deterministically wins the exact tie regardless of callback order.
queueMicrotask(() => {
if (settled) {
return
}
settled = true
const winner = successes.find(({ path }) => path === 'direct') ?? successes[0]!
for (const loser of candidates) {
if (loser !== winner) {
loser.client.close()
}
}
resolve(winner)
})
},
() => {
failures++
rejectIfFinished()
}
)
}
function rejectIfFinished(): void {
if (!settled && failures === candidates.length && successes.length === 0) {
settled = true
reject(new Error('direct and relay pairing paths both failed'))
}
}
})
}
@@ -1,6 +1,7 @@
import { describe, expect, it, vi } from 'vitest'
import type { MobileRelayCredentialBundle } from './mobile-relay-credential-bundle'
import type { MobileRelayPairingJournal } from './mobile-relay-pairing-journal'
import { racePairingCandidates } from './pairing-candidate-race'
import { startPreProfilePairing } from './pre-profile-pairing-coordinator'
import type { HostProfile, PairingOffer, RpcResponse } from './types'
import type { RpcClient } from './rpc-client'
@@ -53,8 +54,13 @@ function fakeClient(responses: RpcResponse[]) {
}
function dependencies(client: RpcClient, events: string[]) {
const unavailableRelay = fakeClient([])
;(unavailableRelay.sendRequest as ReturnType<typeof vi.fn>).mockRejectedValue(
new Error('relay unavailable')
)
return {
connectDirect: vi.fn(() => (events.push('connect'), client)),
connectRelay: vi.fn(() => unavailableRelay),
getNextHostName: vi.fn(async () => 'Blue Whale'),
saveHost: vi.fn(async (_host: HostProfile) => {
events.push('save-host')
@@ -77,6 +83,33 @@ function dependencies(client: RpcClient, events: string[]) {
}
describe('pre-profile pairing coordinator', () => {
it('chooses direct when both post-E2EE status successes settle in the same turn', async () => {
let resolveDirect!: (response: RpcResponse) => void
let resolveRelay!: (response: RpcResponse) => void
const direct = fakeClient([])
const relay = fakeClient([])
;(direct.sendRequest as ReturnType<typeof vi.fn>).mockReturnValue(
new Promise<RpcResponse>((resolve) => {
resolveDirect = resolve
})
)
;(relay.sendRequest as ReturnType<typeof vi.fn>).mockReturnValue(
new Promise<RpcResponse>((resolve) => {
resolveRelay = resolve
})
)
const racing = racePairingCandidates([
{ path: 'direct', client: direct },
{ path: 'relay', client: relay }
])
resolveRelay(success({ path: 'relay' }))
resolveDirect(success({ path: 'direct' }))
await expect(racing).resolves.toMatchObject({ path: 'direct' })
expect(relay.close).toHaveBeenCalledOnce()
expect(direct.close).not.toHaveBeenCalled()
})
it('keeps a legacy offer direct-only through the shared path', async () => {
const events: string[] = []
const client = fakeClient([success({ version: '1.0.0' })])
@@ -208,6 +241,68 @@ describe('pre-profile pairing coordinator', () => {
])
})
it('uses relay-basis provisioning when only the relay reaches post-E2EE status', async () => {
const direct = fakeClient([])
;(direct.sendRequest as ReturnType<typeof vi.fn>).mockRejectedValue(new Error('LAN down'))
let journal: MobileRelayPairingJournal | null = null
const relay = {
sendRequest: vi.fn(async (method: string) => {
if (method === 'status.get') {
return success({ path: 'relay' })
}
const installed = {
v: 1 as const,
reqId: journal!.metadata.installReqId,
authorizationMode: 'relay-basis' as const,
currentVersion: 1,
resumeExpiresAt: now + 86_400_000
}
if (method === 'pairing.provisionRelay') {
return success(installed)
}
return success({
v: 1,
relay: {
v: 1,
directorUrl: relayOffer.relay!.directorUrl,
cellUrl: relayOffer.relay!.cellUrl,
assignmentEpoch: 7,
relayHostId: relayOffer.relay!.relayHostId,
e2eeFraming: 2
},
installStatus: {
v: 1,
reqId: journal!.metadata.installReqId,
state: 'committed',
result: installed
}
})
}),
close: vi.fn()
} as unknown as RpcClient
const deps = dependencies(direct, [])
deps.connectRelay.mockReturnValue(relay)
deps.saveJournal.mockImplementation(async (value) => {
journal = value
})
const attempt = startPreProfilePairing({
offer: relayOffer,
timeoutMs: 5_000,
dependencies: deps
})
await expect(attempt.result).resolves.toEqual({ hostId: `host-${now}` })
expect(direct.close).toHaveBeenCalled()
expect(relay.sendRequest).toHaveBeenNthCalledWith(2, 'pairing.provisionRelay', {
reqId: journal!.metadata.installReqId,
newResumeTokenHash: journal!.metadata.pendingResumeTokenHash
})
expect(deps.writeCredentialBundle).toHaveBeenCalledWith(
expect.objectContaining({ current: expect.objectContaining({ version: 1 }) })
)
})
it('cancels the disposable physical client without publishing a host', async () => {
let resolveStatus!: (response: RpcResponse) => void
const status = new Promise<RpcResponse>((resolve) => {
@@ -5,7 +5,7 @@ import {
type DeviceCredentialInstalled,
type MobileRelayEndpoint
} from '../../../src/shared/mobile-relay-credential-contract'
import { connect, type ConnectOptions, type RpcClient } from './rpc-client'
import { connect, type ConnectOptions } from './rpc-client'
import { getNextHostName, saveHost } from './host-store'
import type { HostProfile, PairingOffer, RpcResponse } from './types'
import {
@@ -21,6 +21,11 @@ import {
promotePairingJournalCredential,
writeMobileRelayCredentialBundle
} from './mobile-relay-credential-bundle'
import {
connectMobileRelayForPairing,
type PairingCandidateClient
} from './mobile-relay-physical-client'
import { racePairingCandidates, type PairingCandidate } from './pairing-candidate-race'
export type PreProfilePairingAttempt = {
readonly result: Promise<{ hostId: string }>
@@ -30,6 +35,7 @@ export type PreProfilePairingAttempt = {
type Dependencies = {
connectDirect: typeof connect
connectRelay: typeof connectMobileRelayForPairing
getNextHostName: typeof getNextHostName
saveHost: typeof saveHost
saveJournal: typeof saveMobileRelayPairingJournal
@@ -42,6 +48,7 @@ type Dependencies = {
const defaultDependencies: Dependencies = {
connectDirect: connect,
connectRelay: connectMobileRelayForPairing,
getNextHostName,
saveHost,
saveJournal: saveMobileRelayPairingJournal,
@@ -59,7 +66,7 @@ export function startPreProfilePairing(args: {
dependencies?: Partial<Dependencies>
}): PreProfilePairingAttempt {
const dependencies = { ...defaultDependencies, ...args.dependencies }
const clients = new Set<RpcClient>()
const clients = new Set<PairingCandidateClient>()
let disposed = false
let timedOut = false
let timer: ReturnType<typeof setTimeout> | null = null
@@ -115,7 +122,7 @@ async function runPairing(
offer: PairingOffer,
connectOptions: ConnectOptions | undefined,
dependencies: Dependencies,
clients: Set<RpcClient>,
clients: Set<PairingCandidateClient>,
isDisposed: () => boolean
): Promise<{ hostId: string }> {
const now = dependencies.now()
@@ -134,14 +141,24 @@ async function runPairing(
assertActive(isDisposed)
}
const client = dependencies.connectDirect(
const directClient = dependencies.connectDirect(
offer.endpoint,
offer.deviceToken,
offer.publicKeyB64,
connectOptions
)
clients.add(client)
requireSuccess(await client.sendRequest('status.get'))
clients.add(directClient)
const candidates: PairingCandidate[] = [{ path: 'direct', client: directClient }]
if (journal) {
const relayClient = dependencies.connectRelay({
relay: offer.relay!,
deviceToken: offer.deviceToken,
desktopPublicKeyB64: offer.publicKeyB64
})
clients.add(relayClient)
candidates.push({ path: 'relay', client: relayClient })
}
const winner = await racePairingCandidates(candidates)
assertActive(isDisposed)
if (!journal) {
@@ -153,16 +170,19 @@ async function runPairing(
...journal,
metadata: {
...journal.metadata,
winner: 'direct',
authorizationMode: 'authenticated-direct'
winner: winner.path,
authorizationMode: winner.path === 'direct' ? 'authenticated-direct' : 'relay-basis'
}
}
await dependencies.updateJournal(journal.metadata.journalId, () => journal!.metadata)
const provision = await client.sendRequest('pairing.provisionRelay', {
const provision = await winner.client.sendRequest('pairing.provisionRelay', {
reqId: journal.metadata.installReqId,
newResumeTokenHash: journal.metadata.pendingResumeTokenHash
})
if (isMethodNotFound(provision)) {
if (winner.path !== 'direct') {
throw new Error('relay pairing RPC unavailable after relay path authentication')
}
await dependencies.saveHost(baseHost(offer, hostId, hostName, now))
await dependencies.clearJournal(journal.metadata.journalId)
return { hostId }
@@ -170,7 +190,7 @@ async function runPairing(
const installed = DeviceCredentialInstalledSchema.parse(requireSuccess(provision))
const endpoints = PairingGetEndpointsResultSchema.parse(
requireSuccess(
await client.sendRequest('pairing.getEndpoints', {
await winner.client.sendRequest('pairing.getEndpoints', {
installReqId: journal.metadata.installReqId
})
)
@@ -0,0 +1,34 @@
import { describe, expect, it } from 'vitest'
import { RelayPhoneHelloSchema } from './mobile-relay-phone-protocol'
describe('relay phone outer protocol', () => {
it('accepts only exact invite, resume, and failure hello variants', () => {
expect(
RelayPhoneHelloSchema.safeParse({
type: 'relay-hello',
ok: true,
credentialKind: 'invite',
leaseExpiresAt: 1
}).success
).toBe(true)
expect(
RelayPhoneHelloSchema.safeParse({
type: 'relay-hello',
ok: false,
code: 4404,
cellUrl: 'https://untrusted.example'
}).success
).toBe(false)
expect(
RelayPhoneHelloSchema.safeParse({
type: 'relay-hello',
ok: true,
credentialKind: 'resume',
leaseExpiresAt: 10,
acceptedCredentialVersion: 2,
acceptedAs: 'grace',
resumeExpiresAt: 8
}).success
).toBe(true)
})
})
+35
View File
@@ -0,0 +1,35 @@
import { z } from 'zod'
const EpochMsSchema = z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER)
export const RelayPhoneHelloSchema = z.union([
z
.object({
type: z.literal('relay-hello'),
ok: z.literal(false),
code: z.number().int().min(4000).max(4999)
})
.strict(),
z
.object({
type: z.literal('relay-hello'),
ok: z.literal(true),
credentialKind: z.literal('invite'),
leaseExpiresAt: EpochMsSchema
})
.strict(),
z
.object({
type: z.literal('relay-hello'),
ok: z.literal(true),
credentialKind: z.literal('resume'),
leaseExpiresAt: EpochMsSchema,
acceptedCredentialVersion: z.number().int().positive(),
acceptedAs: z.enum(['current', 'grace']),
resumeExpiresAt: EpochMsSchema,
graceExpiresAt: EpochMsSchema.optional()
})
.strict()
])
export type RelayPhoneHello = z.infer<typeof RelayPhoneHelloSchema>