diff --git a/docs/reference/relay-regional-placement.md b/docs/reference/relay-regional-placement.md new file mode 100644 index 00000000000..d0acfb5a777 --- /dev/null +++ b/docs/reference/relay-regional-placement.md @@ -0,0 +1,20 @@ +# Relay regional placement + +Orca selects a Relay region in the Electron main process before requesting a new assignment. The +director publishes an allowlisted region catalog containing only HTTPS cell subdomains of that +director; Orca takes three bounded `/health` latency samples per region and caches the stable choice +for 24 hours. A cached region changes only when the alternative is materially faster. + +The assignment request sends only `preferredRegion`. It does not send latency, IP address, country, +pairing data, or credentials. Catalog, probe, and cache failures fall back to an assignment without +a region preference. A rolled-back director that rejects the new field is retried once without +only that field while preserving reconnect behavior. + +The selection measures the desktop network path. Folder workspaces and SSH workspaces share the +same local broker and do not run probes on remote hosts. The phone continues to connect to the +exact cell URL in the desktop pairing payload, so its location is not measured independently and +no mobile protocol update is required. + +For deterministic local diagnostics, set `ORCA_RELAY_REGION_OVERRIDE` to `us-central1` or +`asia-east2` before launching Orca. The override is not an end-user setting and is not written to +the preference cache. diff --git a/src/main/global-fetch-call-site-audit.test.ts b/src/main/global-fetch-call-site-audit.test.ts index 8153cbd992c..094c1b3a50e 100644 --- a/src/main/global-fetch-call-site-audit.test.ts +++ b/src/main/global-fetch-call-site-audit.test.ts @@ -24,6 +24,7 @@ const AUDITED_GLOBAL_FETCH_LINES = new Map([ ['main/orca-profiles/profile-cloud-org-members-client.ts', 1], ['main/rate-limits/codex-fetcher.ts', 3], ['main/runtime/relay/relay-http-client.ts', 2], + ['main/runtime/relay/relay-region-preference.ts', 3], ['main/source-control/hosted-review-api-request.ts', 1], ['main/speech/openai-transcription-client.ts', 1], // fetch appears only inside injected-page script source strings, not as a diff --git a/src/main/runtime/relay/desktop-relay-service.ts b/src/main/runtime/relay/desktop-relay-service.ts index 575cbe23bbc..6ce62270750 100644 --- a/src/main/runtime/relay/desktop-relay-service.ts +++ b/src/main/runtime/relay/desktop-relay-service.ts @@ -18,6 +18,7 @@ import type { import type { DeviceCredentialInstallAuthorization } from './relay-control-requests' import { deriveRelayHostId } from './relay-http-client' import { RelayDemandLedger } from './relay-demand-ledger' +import { createRelayRegionPreferenceReader } from './relay-region-preference' type DesktopRelayServiceOptions = { authConfig: OrcaCloudAuthConfig @@ -69,6 +70,7 @@ export class DesktopRelayService { revokeOutbox: this.revokeOutbox, relayHostId: deriveRelayHostId(keypair.publicKey) }) + const resolvePreferredRegion = createRelayRegionPreferenceReader(options) this.coordinator = new RelayAuthCoordinator({ readContext: () => readRelayAuthContext(options.authConfig, options.userDataPath), hasDemand: ({ identity }) => @@ -85,6 +87,7 @@ export class DesktopRelayService { mobileSocketWiring, isCurrent, refreshAccessToken, + resolvePreferredRegion, onStatus: options.onStatus }) void this.flushRevokeOutbox(broker) diff --git a/src/main/runtime/relay/relay-http-client.test.ts b/src/main/runtime/relay/relay-http-client.test.ts index 213d27629d6..c90592be69b 100644 --- a/src/main/runtime/relay/relay-http-client.test.ts +++ b/src/main/runtime/relay/relay-http-client.test.ts @@ -81,6 +81,42 @@ describe('relay HTTP client', () => { }) }) + it('sends only the coarse region and preserves reconnect when removing it', async () => { + const fetch = vi + .fn() + .mockResolvedValueOnce(Response.json({ error: 'invalid_request' }, { status: 400 })) + .mockResolvedValueOnce( + Response.json({ + v: 1, + cellUrl: 'https://relay-c1.example', + assignmentEpoch: 4, + lease: 'lease-jwt' + }) + ) + + await expect( + requestRelayAssignment({ + directorUrl: 'https://relay.example', + relayToken: 'scoped-token', + relayHostId: 'AbCdEf0123_-xyZ9', + reconnect: true, + preferredRegion: 'asia-east2', + fetch + }) + ).resolves.toMatchObject({ assignmentEpoch: 4 }) + expect(JSON.parse(String(fetch.mock.calls[0]?.[1]?.body))).toEqual({ + v: 1, + relayHostId: 'AbCdEf0123_-xyZ9', + preferredRegion: 'asia-east2', + reconnect: true + }) + expect(JSON.parse(String(fetch.mock.calls[1]?.[1]?.body))).toEqual({ + v: 1, + relayHostId: 'AbCdEf0123_-xyZ9', + reconnect: true + }) + }) + it('retries once unhinted when a rolled-back director rejects the hint', async () => { const fetch = vi .fn() diff --git a/src/main/runtime/relay/relay-http-client.ts b/src/main/runtime/relay/relay-http-client.ts index 0cb370f5ba2..c1ffba04eb2 100644 --- a/src/main/runtime/relay/relay-http-client.ts +++ b/src/main/runtime/relay/relay-http-client.ts @@ -2,6 +2,7 @@ import { createHash } from 'node:crypto' import { z } from 'zod' import type { E2EEKeypair } from '../e2ee-keypair' import { cancelUnreadResponseBody } from '../../lib/unread-response-body' +import type { RelayRegion } from './relay-region-preference' const RELAY_HTTP_REQUEST_DEADLINE_MS = 15_000 const RELAY_RETRY_AFTER_MAX_MS = 5 * 60_000 @@ -117,6 +118,7 @@ export async function requestRelayAssignment(input: { relayToken: string relayHostId: string reconnect?: boolean + preferredRegion?: RelayRegion fetch?: typeof globalThis.fetch requestDeadlineMs?: number }): Promise { @@ -133,6 +135,7 @@ export async function requestRelayAssignment(input: { body: JSON.stringify({ v: 1, relayHostId: input.relayHostId, + ...(input.preferredRegion ? { preferredRegion: input.preferredRegion } : {}), // Declares likely reconnection so the director can verify and admit // through its bounded fast lane instead of the placement queue. ...(input.reconnect ? { reconnect: true } : {}) @@ -141,6 +144,11 @@ export async function requestRelayAssignment(input: { if (!response.ok) { const retryAfterMs = relayRetryAfterMs(response.headers.get('retry-after')) await cancelUnreadResponseBody(response) + if (input.preferredRegion && response.status === 400) { + // A rolled-back director rejects the regional hint; preserve the + // reconnect lane while retrying without only that field. + return await requestRelayAssignment({ ...input, preferredRegion: undefined }) + } if (input.reconnect && response.status === 400) { // A rolled-back director rejects unknown fields; retry once unhinted. return await requestRelayAssignment({ ...input, reconnect: false }) diff --git a/src/main/runtime/relay/relay-origin-pool.ts b/src/main/runtime/relay/relay-origin-pool.ts index 6f5f5951bf8..13f02a94075 100644 --- a/src/main/runtime/relay/relay-origin-pool.ts +++ b/src/main/runtime/relay/relay-origin-pool.ts @@ -7,6 +7,7 @@ import type { RelayDrainMessage } from './relay-control-protocol' import { RelayDrainRetrySchedule } from './relay-drain-retry-schedule' import { RelayHttpError, requestRelayAssignment, type RelayAssignment } from './relay-http-client' import type { RelayBrokerStatus, RelayIdentity } from './relay-session-broker-contract' +import type { RelayRegion } from './relay-region-preference' type RelayOriginPoolOptions = { directorUrl: string @@ -17,6 +18,7 @@ type RelayOriginPoolOptions = { mobileSocketWiring: MobileSocketWiring isCurrent: () => boolean onStatus: (status: RelayBrokerStatus) => void + resolvePreferredRegion?: () => Promise fetch?: typeof globalThis.fetch createControlSocket?: (url: string, relayJwt: string) => WebSocket createDataSocket?: (url: string) => WebSocket @@ -158,6 +160,8 @@ export class RelayOriginPool { if (!this.relayJwt) { throw new Error('relay_authorization_unavailable') } + const preferredRegion = await this.options.resolvePreferredRegion?.().catch(() => undefined) + this.assertCurrent() // Why: only the configured director can choose a migration target. const assignment = await requestRelayAssignment({ directorUrl: this.options.directorUrl, @@ -166,6 +170,7 @@ export class RelayOriginPool { // Recovery always follows an established assignment; the director // verifies this and admits through its reconnect fast lane. reconnect: true, + preferredRegion, fetch: this.options.fetch }) this.assertCurrent() diff --git a/src/main/runtime/relay/relay-region-preference.test.ts b/src/main/runtime/relay/relay-region-preference.test.ts new file mode 100644 index 00000000000..bb4001d08e2 --- /dev/null +++ b/src/main/runtime/relay/relay-region-preference.test.ts @@ -0,0 +1,278 @@ +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { cancelTrackingResponse } from '../../lib/unread-response-body.test-fixtures' +import { probeRelayOrigin, RelayRegionPreferenceResolver } from './relay-region-preference' + +const DIRECTOR = 'https://relay.example.test' +const US = 'https://us-c1.relay.example.test' +const US_SECONDARY = 'https://us-c2.relay.example.test' +const ASIA = 'https://asia-c1.relay.example.test' +const tempPaths: string[] = [] + +afterEach(() => { + for (const path of tempPaths.splice(0)) { + rmSync(path, { recursive: true, force: true }) + } +}) + +function userDataPath(): string { + const path = mkdtempSync(join(tmpdir(), 'orca-relay-region-')) + tempPaths.push(path) + return path +} + +function catalogFetch(regions: unknown) { + return vi.fn(async () => Response.json({ v: 1, regions })) +} + +function sampledProbe(samples: Record) { + const calls: string[] = [] + const probe = async (origin: string): Promise => { + calls.push(origin) + return samples[origin]?.shift() ?? null + } + return { calls, probe } +} + +function cachePath(path: string): string { + return join(path, 'orca-relay-region-preference.json') +} + +describe('Relay region preference', () => { + it('measures three rounds across one- and two-origin catalogs and caches Asia', async () => { + const path = userDataPath() + const fetch = catalogFetch([ + { region: 'us-central1', probeOrigins: [US, US_SECONDARY] }, + { region: 'asia-east2', probeOrigins: [ASIA] } + ]) + const { calls, probe } = sampledProbe({ + [US]: [160, 170, 150], + [US_SECONDARY]: [155, 165, 145], + [ASIA]: [35, 40, 30] + }) + const resolver = new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch, + probe, + now: () => 1_000 + }) + + await expect(resolver.resolve()).resolves.toBe('asia-east2') + expect(calls.filter((origin) => origin === US)).toHaveLength(3) + expect(calls.filter((origin) => origin === US_SECONDARY)).toHaveLength(3) + expect(calls.filter((origin) => origin === ASIA)).toHaveLength(3) + expect(JSON.parse(readFileSync(cachePath(path), 'utf8'))).toMatchObject({ + v: 1, + directorUrl: DIRECTOR, + region: 'asia-east2', + latencyMs: 35 + }) + + const offlineFetch = vi.fn(async () => { + throw new Error('offline') + }) + await expect( + new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch: offlineFetch, + now: () => 2_000 + }).resolve() + ).resolves.toBe('asia-east2') + expect(offlineFetch).not.toHaveBeenCalled() + }) + + it('keeps the cached region unless a stable alternative is meaningfully faster', async () => { + const path = userDataPath() + writeFileSync( + cachePath(path), + JSON.stringify({ + v: 1, + directorUrl: DIRECTOR, + region: 'us-central1', + latencyMs: 100, + expiresAt: 999 + }) + ) + const regions = [ + { region: 'us-central1', probeOrigins: [US] }, + { region: 'asia-east2', probeOrigins: [ASIA] } + ] + const first = sampledProbe({ [US]: [95, 100, 105], [ASIA]: [80, 85, 90] }) + await expect( + new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch: catalogFetch(regions), + probe: first.probe, + now: () => 1_000 + }).resolve() + ).resolves.toBe('us-central1') + + writeFileSync( + cachePath(path), + JSON.stringify({ + v: 1, + directorUrl: DIRECTOR, + region: 'us-central1', + latencyMs: 100, + expiresAt: 999 + }) + ) + const second = sampledProbe({ [US]: [95, 100, 105], [ASIA]: [55, 60, 65] }) + await expect( + new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch: catalogFetch(regions), + probe: second.probe, + now: () => 1_000 + }).resolve() + ).resolves.toBe('asia-east2') + }) + + it('falls back without a hint for corrupt cache, old catalogs, and unstable probes', async () => { + const path = userDataPath() + writeFileSync(cachePath(path), '{not-json') + const invalidCatalogs = [ + [{ region: 'unknown', probeOrigins: [US] }], + [ + { region: 'us-central1', probeOrigins: [US] }, + { region: 'asia-east2', probeOrigins: [US] } + ], + [{ region: 'us-central1', probeOrigins: ['http://us.relay.example.test'] }], + [{ region: 'us-central1', probeOrigins: ['https://external.example.test'] }] + ] + for (const regions of invalidCatalogs) { + await expect( + new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch: catalogFetch(regions), + probe: async () => 10, + now: () => 1_000 + }).resolve() + ).resolves.toBeUndefined() + } + + const unstable = sampledProbe({ [US]: [10, 20, 200] }) + await expect( + new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch: catalogFetch([{ region: 'us-central1', probeOrigins: [US] }]), + probe: unstable.probe, + now: () => 1_000 + }).resolve() + ).resolves.toBeUndefined() + }) + + it('recovers from corrupt cache and cancels an old directors error response', async () => { + const path = userDataPath() + writeFileSync(cachePath(path), '{not-json') + const healthy = sampledProbe({ [ASIA]: [30, 32, 34] }) + await expect( + new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch: catalogFetch([{ region: 'asia-east2', probeOrigins: [ASIA] }]), + probe: healthy.probe, + now: () => 1_000 + }).resolve() + ).resolves.toBe('asia-east2') + + rmSync(cachePath(path), { force: true }) + let cancelled = 0 + const oldDirector = vi.fn(async () => + cancelTrackingResponse(404, () => { + cancelled += 1 + }) + ) + await expect( + new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch: oldDirector, + now: () => 1_000 + }).resolve() + ).resolves.toBeUndefined() + expect(cancelled).toBe(1) + }) + + it('rejects a cache expiry beyond the 24-hour bound', async () => { + const path = userDataPath() + writeFileSync( + cachePath(path), + JSON.stringify({ + v: 1, + directorUrl: DIRECTOR, + region: 'us-central1', + latencyMs: 100, + expiresAt: 10 * 24 * 60 * 60_000 + }) + ) + const healthy = sampledProbe({ [ASIA]: [30, 32, 34] }) + + await expect( + new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch: catalogFetch([{ region: 'asia-east2', probeOrigins: [ASIA] }]), + probe: healthy.probe, + now: () => 1_000 + }).resolve() + ).resolves.toBe('asia-east2') + }) + + it('uses a valid diagnostic override without network or cache mutation', async () => { + const path = userDataPath() + const fetch = vi.fn() + const resolver = new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + diagnosticOverride: 'asia-east2', + fetch + }) + + await expect(resolver.resolve()).resolves.toBe('asia-east2') + expect(fetch).not.toHaveBeenCalled() + }) + + it('bounds an offline catalog request and returns no preference', async () => { + const fetch = vi.fn( + async (_url, init) => + await new Promise((_resolve, reject) => { + init?.signal?.addEventListener('abort', () => reject(init.signal?.reason), { once: true }) + }) + ) + + await expect( + new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: userDataPath(), + fetch, + requestTimeoutMs: 5 + }).resolve() + ).resolves.toBeUndefined() + expect(fetch).toHaveBeenCalledOnce() + }) + + it('probes only the canonical health path and cancels its body', async () => { + let cancelled = 0 + const fetch = vi.fn(async () => + cancelTrackingResponse(200, () => { + cancelled += 1 + }) + ) + const times = [10, 42] + + await expect(probeRelayOrigin(ASIA, fetch, () => times.shift()!)).resolves.toBe(32) + expect(fetch.mock.calls[0]?.[0]).toBe(`${ASIA}/health`) + expect(fetch.mock.calls[0]?.[1]).toMatchObject({ method: 'GET' }) + expect(fetch.mock.calls[0]?.[1]?.headers).toBeUndefined() + expect(cancelled).toBe(1) + }) +}) diff --git a/src/main/runtime/relay/relay-region-preference.ts b/src/main/runtime/relay/relay-region-preference.ts new file mode 100644 index 00000000000..a35ee7f0815 --- /dev/null +++ b/src/main/runtime/relay/relay-region-preference.ts @@ -0,0 +1,322 @@ +import { existsSync, readFileSync, statSync } from 'node:fs' +import { join } from 'node:path' +import { performance } from 'node:perf_hooks' +import { z } from 'zod' +import { cancelUnreadResponseBody } from '../../lib/unread-response-body' +import { readFetchResponseJsonWithinLimit } from '../../../shared/fetch-response-body' +import { hardenExistingSecureFile, writeSecureJsonFile } from '../../../shared/secure-file' + +export const RELAY_REGIONS = ['us-central1', 'asia-east2'] as const +export type RelayRegion = (typeof RELAY_REGIONS)[number] + +const RELAY_REGION_CACHE_FILENAME = 'orca-relay-region-preference.json' +const CACHE_MAX_BYTES = 8 * 1024 +const CATALOG_MAX_BYTES = 16 * 1024 +const CACHE_TTL_MS = 24 * 60 * 60_000 +const PROBE_SAMPLES = 3 +const PROBE_TIMEOUT_MS = 1_500 +const SWITCH_MINIMUM_MS = 25 +const SWITCH_RATIO = 0.8 + +const RelayRegionSchema = z.enum(RELAY_REGIONS) +const RelayProbeOriginSchema = z.string().max(2_048).refine(isCanonicalHttpsOrigin) +const RelayRegionCatalogSchema = z + .object({ + v: z.literal(1), + regions: z + .array( + z + .object({ + region: RelayRegionSchema, + probeOrigins: z.array(RelayProbeOriginSchema).min(1).max(2) + }) + .strict() + ) + .max(RELAY_REGIONS.length) + }) + .strict() + .superRefine((catalog, context) => { + const regions = new Set() + const origins = new Set() + for (const [regionIndex, entry] of catalog.regions.entries()) { + if (regions.has(entry.region)) { + context.addIssue({ + code: 'custom', + message: 'duplicate relay region', + path: ['regions', regionIndex, 'region'] + }) + } + regions.add(entry.region) + for (const [originIndex, origin] of entry.probeOrigins.entries()) { + if (origins.has(origin)) { + context.addIssue({ + code: 'custom', + message: 'duplicate relay probe origin', + path: ['regions', regionIndex, 'probeOrigins', originIndex] + }) + } + origins.add(origin) + } + } + }) + +const RelayRegionCacheSchema = z + .object({ + v: z.literal(1), + directorUrl: z.string().max(2_048), + region: RelayRegionSchema, + latencyMs: z.number().finite().nonnegative().max(60_000), + expiresAt: z.number().int().positive().max(Number.MAX_SAFE_INTEGER) + }) + .strict() + +type RelayRegionCatalog = z.infer +type RelayRegionCache = z.infer +type RegionMeasurement = { region: RelayRegion; latencyMs: number } + +type RelayRegionPreferenceOptions = { + directorUrl: string + userDataPath: string + fetch?: typeof globalThis.fetch + now?: () => number + measureNow?: () => number + diagnosticOverride?: string + probe?: (origin: string) => Promise + requestTimeoutMs?: number +} + +export class RelayRegionPreferenceResolver { + private readonly options: RelayRegionPreferenceOptions + private pending: Promise | null = null + + constructor(options: RelayRegionPreferenceOptions) { + this.options = options + } + + async resolve(): Promise { + const override = RelayRegionSchema.safeParse( + this.options.diagnosticOverride ?? process.env.ORCA_RELAY_REGION_OVERRIDE + ) + if (override.success) { + return override.data + } + + const now = (this.options.now ?? Date.now)() + const cache = readRelayRegionCache(this.options.userDataPath, this.options.directorUrl, now) + if (cache && cache.expiresAt > now) { + return cache.region + } + if (this.pending) { + return await this.pending + } + + this.pending = this.refresh(cache, now).catch(() => undefined) + try { + return await this.pending + } finally { + this.pending = null + } + } + + private async refresh( + previous: RelayRegionCache | null, + now: number + ): Promise { + const fetch = this.options.fetch ?? globalThis.fetch + const catalog = await fetchRelayRegionCatalog( + this.options.directorUrl, + fetch, + this.options.requestTimeoutMs ?? PROBE_TIMEOUT_MS + ) + const probe = + this.options.probe ?? + ((origin: string) => + probeRelayOrigin( + origin, + fetch, + this.options.measureNow ?? (() => performance.now()), + this.options.requestTimeoutMs ?? PROBE_TIMEOUT_MS + )) + const measurements = ( + await Promise.all(catalog.regions.map((entry) => measureRegion(entry, probe))) + ).filter((measurement): measurement is RegionMeasurement => measurement !== null) + const selected = selectRegionMeasurement(measurements, previous) + if (!selected) { + return undefined + } + + try { + writeSecureJsonFile(join(this.options.userDataPath, RELAY_REGION_CACHE_FILENAME), { + v: 1, + directorUrl: this.options.directorUrl, + region: selected.region, + latencyMs: selected.latencyMs, + expiresAt: now + CACHE_TTL_MS + } satisfies RelayRegionCache) + } catch { + // A cache write must not block an otherwise valid Relay assignment. + } + return selected.region + } +} + +export function createRelayRegionPreferenceReader(input: { + authConfig: { relayDirectorUrl: string } + userDataPath: string +}): () => Promise { + const resolver = new RelayRegionPreferenceResolver({ + directorUrl: input.authConfig.relayDirectorUrl, + userDataPath: input.userDataPath + }) + return () => resolver.resolve() +} + +async function fetchRelayRegionCatalog( + directorUrl: string, + fetch: typeof globalThis.fetch, + timeoutMs: number +): Promise { + if (!isCanonicalDirectorOrigin(directorUrl)) { + throw new Error('invalid relay director origin') + } + const response = await fetch(`${directorUrl}/v1/regions`, { + method: 'GET', + cache: 'no-store', + redirect: 'error', + signal: AbortSignal.timeout(timeoutMs) + }) + if (!response.ok) { + await cancelUnreadResponseBody(response) + throw new Error(`relay region catalog failed (${response.status})`) + } + const body = await readFetchResponseJsonWithinLimit(response, CATALOG_MAX_BYTES, { + structuralTokens: 64, + nestingDepth: 8 + }) + const catalog = RelayRegionCatalogSchema.parse(body) + if ( + catalog.regions.some((entry) => + entry.probeOrigins.some((origin) => !isProbeOriginForDirector(origin, directorUrl)) + ) + ) { + throw new Error('relay probe origin does not belong to the director') + } + return catalog +} + +export async function probeRelayOrigin( + origin: string, + fetch: typeof globalThis.fetch, + now = () => performance.now(), + timeoutMs = PROBE_TIMEOUT_MS +): Promise { + if (!RelayProbeOriginSchema.safeParse(origin).success) { + return null + } + const startedAt = now() + try { + const response = await fetch(`${origin}/health`, { + method: 'GET', + cache: 'no-store', + redirect: 'error', + signal: AbortSignal.timeout(timeoutMs) + }) + const latencyMs = now() - startedAt + await cancelUnreadResponseBody(response) + return response.ok && Number.isFinite(latencyMs) && latencyMs >= 0 ? latencyMs : null + } catch { + return null + } +} + +async function measureRegion( + entry: RelayRegionCatalog['regions'][number], + probe: (origin: string) => Promise +): Promise { + const samples: number[] = [] + for (let sample = 0; sample < PROBE_SAMPLES; sample++) { + const latencies = (await Promise.all(entry.probeOrigins.map(probe))).filter( + (latency): latency is number => latency !== null + ) + if (latencies.length === 0) { + return null + } + samples.push(Math.min(...latencies)) + } + samples.sort((left, right) => left - right) + const median = samples[1]! + const spread = samples[2]! - samples[0]! + if (spread > Math.max(20, median * 0.5)) { + return null + } + return { region: entry.region, latencyMs: median } +} + +function selectRegionMeasurement( + measurements: RegionMeasurement[], + previous: RelayRegionCache | null +): RegionMeasurement | null { + const order = new Map(RELAY_REGIONS.map((region, index) => [region, index])) + const sorted = [...measurements].sort( + (left, right) => + left.latencyMs - right.latencyMs || order.get(left.region)! - order.get(right.region)! + ) + const best = sorted[0] + if (!best || !previous || best.region === previous.region) { + return best ?? null + } + const current = measurements.find((measurement) => measurement.region === previous.region) + if (!current) { + return best + } + const meaningful = + current.latencyMs - best.latencyMs >= SWITCH_MINIMUM_MS && + best.latencyMs <= current.latencyMs * SWITCH_RATIO + return meaningful ? best : current +} + +function readRelayRegionCache(userDataPath: string, directorUrl: string, now: number) { + const path = join(userDataPath, RELAY_REGION_CACHE_FILENAME) + try { + if (!existsSync(path)) { + return null + } + hardenExistingSecureFile(path) + if (statSync(path).size > CACHE_MAX_BYTES) { + return null + } + const parsed = RelayRegionCacheSchema.safeParse(JSON.parse(readFileSync(path, 'utf8'))) + return parsed.success && + parsed.data.directorUrl === directorUrl && + parsed.data.expiresAt <= now + CACHE_TTL_MS + ? parsed.data + : null + } catch { + return null + } +} + +function isCanonicalHttpsOrigin(value: string): boolean { + try { + const url = new URL(value) + return url.protocol === 'https:' && url.origin === value + } catch { + return false + } +} + +function isCanonicalDirectorOrigin(value: string): boolean { + try { + const url = new URL(value) + const loopback = ['127.0.0.1', 'localhost', '[::1]'].includes(url.hostname) + return ( + url.origin === value && (url.protocol === 'https:' || (url.protocol === 'http:' && loopback)) + ) + } catch { + return false + } +} + +function isProbeOriginForDirector(origin: string, directorUrl: string): boolean { + return new URL(origin).hostname.endsWith(`.${new URL(directorUrl).hostname}`) +} diff --git a/src/main/runtime/relay/relay-session-broker-contract.ts b/src/main/runtime/relay/relay-session-broker-contract.ts index 3c7a707e5bd..377ef90416f 100644 --- a/src/main/runtime/relay/relay-session-broker-contract.ts +++ b/src/main/runtime/relay/relay-session-broker-contract.ts @@ -3,6 +3,7 @@ import type { OrcaCloudAuthConfig } from '../../orca-profiles/profile-cloud-auth import type { MobileRelayStatus } from '../../../shared/mobile-relay-status' import type { E2EEKeypair } from '../e2ee-keypair' import type { MobileSocketWiring } from '../rpc/mobile-socket-wiring' +import type { RelayRegion } from './relay-region-preference' export type RelayBrokerStatus = MobileRelayStatus @@ -21,6 +22,7 @@ export type RelaySessionBrokerOptions = { mobileSocketWiring: MobileSocketWiring isCurrent: () => boolean refreshAccessToken: () => Promise + resolvePreferredRegion?: () => Promise onStatus: (status: RelayBrokerStatus) => void fetch?: typeof globalThis.fetch createControlSocket?: (url: string, relayJwt: string) => WebSocket diff --git a/src/main/runtime/relay/relay-session-broker.test.ts b/src/main/runtime/relay/relay-session-broker.test.ts index 1fd172c2537..bea6e072273 100644 --- a/src/main/runtime/relay/relay-session-broker.test.ts +++ b/src/main/runtime/relay/relay-session-broker.test.ts @@ -211,7 +211,20 @@ describe('RelaySessionBroker lifecycle ownership', () => { assignmentEpoch: 2, leaseExpiresAt: 2_000_000 }) - const broker = await RelaySessionBroker.connect(brokerOptions({ onStatus: vi.fn() })) + const resolvePreferredRegion = vi + .fn() + .mockResolvedValueOnce('asia-east2') + .mockResolvedValueOnce('us-central1') + const broker = await RelaySessionBroker.connect( + brokerOptions({ + onStatus: vi.fn(), + resolvePreferredRegion + }) + ) + expect(fakes.assign).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ preferredRegion: 'asia-east2', reconnect: true }) + ) fakes.controls[0]!.options.onConnectionOpen({ connId: 'old-basis', connTicket: 'T'.repeat(43), @@ -227,6 +240,12 @@ describe('RelaySessionBroker lifecycle ownership', () => { }) await vi.waitFor(() => expect(fakes.controls).toHaveLength(2)) + expect(fakes.assign).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ preferredRegion: 'us-central1', reconnect: true }) + ) + expect(resolvePreferredRegion).toHaveBeenCalledTimes(2) + expect(broker.endpoint?.cellUrl).toBe('https://relay-c2.example.test') expect(fakes.transports[0]!.openConnection).toHaveBeenCalledOnce() expect(brokerBasisIds(broker)).toEqual(['old-basis']) diff --git a/src/main/runtime/relay/relay-session-broker.ts b/src/main/runtime/relay/relay-session-broker.ts index e2a8402a04c..e04640a5b2e 100644 --- a/src/main/runtime/relay/relay-session-broker.ts +++ b/src/main/runtime/relay/relay-session-broker.ts @@ -45,6 +45,7 @@ export class RelaySessionBroker { mobileSocketWiring: options.mobileSocketWiring, isCurrent: () => this.isCurrent(), onStatus: (status) => this.publishStatus(status), + resolvePreferredRegion: options.resolvePreferredRegion, fetch: options.fetch, createControlSocket: options.createControlSocket, createDataSocket: options.createDataSocket, @@ -197,12 +198,15 @@ export class RelaySessionBroker { private async open(accessToken: string): Promise { this.publishStatus('connecting') - const authorization = await exchangeRelayAuthorization({ - endpoint: this.options.authConfig.relayTokenEndpoint, - accessToken, - keypair: this.options.keypair, - fetch: this.options.fetch - }) + const [authorization, preferredRegion] = await Promise.all([ + exchangeRelayAuthorization({ + endpoint: this.options.authConfig.relayTokenEndpoint, + accessToken, + keypair: this.options.keypair, + fetch: this.options.fetch + }), + this.options.resolvePreferredRegion?.().catch(() => undefined) ?? Promise.resolve(undefined) + ]) this.assertCurrent() const assignment = await requestRelayAssignment({ directorUrl: this.options.authConfig.relayDirectorUrl, @@ -212,6 +216,7 @@ export class RelaySessionBroker { // director verifies the claim, and first-ever pairing simply falls // through to the placement lane. reconnect: true, + preferredRegion, fetch: this.options.fetch }) this.assertCurrent()