diff --git a/docs/reference/relay-regional-placement.md b/docs/reference/relay-regional-placement.md index d0acfb5a777..2e984ab0789 100644 --- a/docs/reference/relay-regional-placement.md +++ b/docs/reference/relay-regional-placement.md @@ -2,8 +2,27 @@ 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. +director. Orca discards one warm-up `/health` request per probe origin — a cold request pays TCP and +TLS setup that can exceed the round trip it measures — then takes three bounded samples and compares +regions by their minimum. A wide spread still rejects a region, but only a genuinely flapping one. +The stable choice is cached for 24 hours, and a cached region changes only when the alternative is +materially faster. + +A region wins only against a measured competitor. If any region in the catalog is rejected or cannot +be measured, Orca sends no hint rather than selecting the sole survivor. Sending no hint is not +neutral placement: the director assigns `preferredRegion ?? RELAY_DEFAULT_REGION`, and the default +is `us-central1`. So an `asia-east2` user whose `us-central1` probe fails or flaps once is placed in +`us-central1` for that refresh. That trade is accepted because the relay database is +`us-central1`-only, and it is bounded: the withheld hint is cached for one hour, not the 24 hours a +chosen region gets, so the next hour re-measures. An origin that fails its warm-up probe is dropped +before the sampling rounds, so an unreachable region costs one probe timeout rather than four. + +After a control socket registers, Orca probes the cell it actually landed on, once per cell URL per +process. The cache is deleted only when it names a region other than the best measured one and the +assigned cell is more than three times slower than that region — a far cell under a cache that still +names the best region means the director declined the hint, and re-measuring would return the same +answer. Self-heal skips an absent, expired, or no-hint cache, and never runs under +`ORCA_RELAY_REGION_OVERRIDE`. 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 diff --git a/src/main/global-fetch-call-site-audit.test.ts b/src/main/global-fetch-call-site-audit.test.ts index e4c0539fbfb..63801dd3eed 100644 --- a/src/main/global-fetch-call-site-audit.test.ts +++ b/src/main/global-fetch-call-site-audit.test.ts @@ -25,6 +25,7 @@ const AUDITED_GLOBAL_FETCH_LINES = new Map([ ['main/rate-limits/codex-fetcher.ts', 3], ['main/runtime/relay/relay-http-client.ts', 2], ['main/runtime/relay/relay-region-preference.ts', 3], + ['main/runtime/relay/relay-region-probe.ts', 1], ['main/source-control/hosted-review-api-request.ts', 1], ['main/speech/openai-transcription-client.ts', 1], // Main HTTP port: one type declaration plus the Node fallback call. The fallback diff --git a/src/main/runtime/relay/desktop-relay-service.ts b/src/main/runtime/relay/desktop-relay-service.ts index def786e7758..0ce44918870 100644 --- a/src/main/runtime/relay/desktop-relay-service.ts +++ b/src/main/runtime/relay/desktop-relay-service.ts @@ -71,7 +71,7 @@ export class DesktopRelayService { revokeOutbox: this.revokeOutbox, relayHostId: deriveRelayHostId(keypair.publicKey) }) - const resolvePreferredRegion = createRelayRegionPreferenceReader(options) + const regionPreference = createRelayRegionPreferenceReader(options) this.coordinator = new RelayAuthCoordinator({ readContext: () => readRelayAuthContext(options.authConfig, options.userDataPath), hasDemand: ({ identity }) => @@ -88,7 +88,8 @@ export class DesktopRelayService { mobileSocketWiring, isCurrent, refreshAccessToken, - resolvePreferredRegion, + resolvePreferredRegion: regionPreference.resolvePreferredRegion, + onAssignedCellActive: regionPreference.noteAssignedCell, onStatus: options.onStatus }) void this.flushRevokeOutbox(broker) diff --git a/src/main/runtime/relay/relay-region-preference.test.ts b/src/main/runtime/relay/relay-region-preference.test.ts index bb4001d08e2..61a9846072f 100644 --- a/src/main/runtime/relay/relay-region-preference.test.ts +++ b/src/main/runtime/relay/relay-region-preference.test.ts @@ -1,17 +1,24 @@ -import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs' +import { existsSync, 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' +import { RelayRegionPreferenceResolver } from './relay-region-preference' +import { probeRelayOrigin } from './relay-region-probe' 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 CELL = 'https://cell-7.relay.example.test' +const BOTH_REGIONS = [ + { region: 'us-central1', probeOrigins: [US] }, + { region: 'asia-east2', probeOrigins: [ASIA] } +] const tempPaths: string[] = [] afterEach(() => { + vi.unstubAllEnvs() for (const path of tempPaths.splice(0)) { rmSync(path, { recursive: true, force: true }) } @@ -27,6 +34,7 @@ function catalogFetch(regions: unknown) { return vi.fn(async () => Response.json({ v: 1, regions })) } +// Each list starts with the discarded warm-up probe, then the three kept samples. function sampledProbe(samples: Record) { const calls: string[] = [] const probe = async (origin: string): Promise => { @@ -36,21 +44,35 @@ function sampledProbe(samples: Record) { return { calls, probe } } +function writeNoHintCache(path: string, expiresAt: number): void { + writeFileSync( + cachePath(path), + JSON.stringify({ v: 1, directorUrl: DIRECTOR, region: null, expiresAt }) + ) +} + function cachePath(path: string): string { return join(path, 'orca-relay-region-preference.json') } +function writeCache(path: string, region: string, expiresAt = 999): void { + writeFileSync( + cachePath(path), + JSON.stringify({ v: 1, directorUrl: DIRECTOR, region, latencyMs: 100, expiresAt }) + ) +} + describe('Relay region preference', () => { - it('measures three rounds across one- and two-origin catalogs and caches Asia', async () => { + it('measures a warm-up plus three rounds across one- and two-origin catalogs', 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] + [US]: [400, 160, 170, 150], + [US_SECONDARY]: [390, 155, 165, 145], + [ASIA]: [90, 35, 40, 30] }) const resolver = new RelayRegionPreferenceResolver({ directorUrl: DIRECTOR, @@ -61,14 +83,14 @@ describe('Relay region preference', () => { }) 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(calls.filter((origin) => origin === US)).toHaveLength(4) + expect(calls.filter((origin) => origin === US_SECONDARY)).toHaveLength(4) + expect(calls.filter((origin) => origin === ASIA)).toHaveLength(4) expect(JSON.parse(readFileSync(cachePath(path), 'utf8'))).toMatchObject({ v: 1, directorUrl: DIRECTOR, region: 'asia-east2', - latencyMs: 35 + latencyMs: 30 }) const offlineFetch = vi.fn(async () => { @@ -85,55 +107,173 @@ describe('Relay region preference', () => { expect(offlineFetch).not.toHaveBeenCalled() }) - it('keeps the cached region unless a stable alternative is meaningfully faster', async () => { + it('discards the warm-up probe instead of counting it as the region latency', 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] }) + const { calls, probe } = sampledProbe({ + [US]: [5, 40, 42, 44], + [ASIA]: [7, 300, 302, 304] + }) + await expect( new RelayRegionPreferenceResolver({ directorUrl: DIRECTOR, userDataPath: path, - fetch: catalogFetch(regions), + fetch: catalogFetch(BOTH_REGIONS), + probe, + now: () => 1_000 + }).resolve() + ).resolves.toBe('us-central1') + expect(calls.filter((origin) => origin === US)).toHaveLength(4) + expect(calls.filter((origin) => origin === ASIA)).toHaveLength(4) + // 5 and 7 were the warm-ups; the cached latency is the best kept sample. + expect(JSON.parse(readFileSync(cachePath(path), 'utf8'))).toMatchObject({ latencyMs: 40 }) + }) + + it.each([ + { + name: 'us-central1', + samples: { [US]: [85, 90, 36, 36], [ASIA]: [230, 220, 218, 218] } + }, + { + name: 'asia-east2', + samples: { [US]: [230, 220, 218, 218], [ASIA]: [85, 90, 36, 36] } + } + ])('picks the near region $name despite a cold first sample', async ({ name, samples }) => { + const path = userDataPath() + const { probe } = sampledProbe(samples) + + await expect( + new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch: catalogFetch(BOTH_REGIONS), + probe, + now: () => 1_000 + }).resolve() + ).resolves.toBe(name) + expect(JSON.parse(readFileSync(cachePath(path), 'utf8'))).toMatchObject({ + region: name, + latencyMs: 36 + }) + }) + + it.each([ + { name: 'a flapping near region', near: [50, 10, 20, 400] }, + { name: 'an unreachable near region', near: [] } + ])('sends no hint when $name leaves a sole survivor', async ({ near }) => { + const path = userDataPath() + const { probe } = sampledProbe({ [US]: near, [ASIA]: [230, 220, 218, 218] }) + + await expect( + new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch: catalogFetch(BOTH_REGIONS), + probe, + now: () => 1_000 + }).resolve() + ).resolves.toBeUndefined() + // The withheld hint is remembered briefly so a reconnect does not re-probe. + const cached = JSON.parse(readFileSync(cachePath(path), 'utf8')) + expect(cached).toEqual({ v: 1, directorUrl: DIRECTOR, region: null, expiresAt: 3_601_000 }) + }) + + it('reuses the short-lived no-hint cache instead of re-probing on reconnect', async () => { + const path = userDataPath() + const { calls, probe } = sampledProbe({ [ASIA]: [230, 220, 218, 218] }) + await expect( + new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch: catalogFetch(BOTH_REGIONS), + probe, + now: () => 1_000 + }).resolve() + ).resolves.toBeUndefined() + // An unreachable region costs its warm-up probe only, not three more rounds. + expect(calls.filter((origin) => origin === US)).toHaveLength(1) + + const fetch = vi.fn() + await expect( + new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch, + now: () => 3_600_000 + }).resolve() + ).resolves.toBeUndefined() + expect(fetch).not.toHaveBeenCalled() + }) + + it('drops an origin that failed its warm-up without losing the region', async () => { + const path = userDataPath() + const { calls, probe } = sampledProbe({ + [US]: [300, 36, 38, 40], + [ASIA]: [400, 218, 220, 222] + }) + + await expect( + new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch: catalogFetch([ + { region: 'us-central1', probeOrigins: [US, US_SECONDARY] }, + { region: 'asia-east2', probeOrigins: [ASIA] } + ]), + probe, + now: () => 1_000 + }).resolve() + ).resolves.toBe('us-central1') + expect(calls.filter((origin) => origin === US_SECONDARY)).toHaveLength(1) + expect(calls.filter((origin) => origin === US)).toHaveLength(4) + }) + + it('keeps the cached region unless a stable alternative is meaningfully faster', async () => { + const path = userDataPath() + writeCache(path, 'us-central1') + const first = sampledProbe({ [US]: [300, 95, 100, 105], [ASIA]: [300, 80, 85, 90] }) + await expect( + new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch: catalogFetch(BOTH_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] }) + writeCache(path, 'us-central1') + const second = sampledProbe({ [US]: [300, 95, 100, 105], [ASIA]: [300, 55, 60, 65] }) await expect( new RelayRegionPreferenceResolver({ directorUrl: DIRECTOR, userDataPath: path, - fetch: catalogFetch(regions), + fetch: catalogFetch(BOTH_REGIONS), probe: second.probe, now: () => 1_000 }).resolve() ).resolves.toBe('asia-east2') }) + it('switches away from a cached far region once both regions measure', async () => { + const path = userDataPath() + writeCache(path, 'asia-east2') + const { probe } = sampledProbe({ [US]: [85, 90, 36, 36], [ASIA]: [230, 220, 218, 218] }) + + await expect( + new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch: catalogFetch(BOTH_REGIONS), + probe, + now: () => 1_000 + }).resolve() + ).resolves.toBe('us-central1') + expect(JSON.parse(readFileSync(cachePath(path), 'utf8'))).toMatchObject({ + region: 'us-central1' + }) + }) + it('falls back without a hint for corrupt cache, old catalogs, and unstable probes', async () => { const path = userDataPath() writeFileSync(cachePath(path), '{not-json') @@ -158,7 +298,7 @@ describe('Relay region preference', () => { ).resolves.toBeUndefined() } - const unstable = sampledProbe({ [US]: [10, 20, 200] }) + const unstable = sampledProbe({ [US]: [15, 10, 20, 400] }) await expect( new RelayRegionPreferenceResolver({ directorUrl: DIRECTOR, @@ -173,7 +313,7 @@ describe('Relay region preference', () => { 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] }) + const healthy = sampledProbe({ [ASIA]: [90, 30, 32, 34] }) await expect( new RelayRegionPreferenceResolver({ directorUrl: DIRECTOR, @@ -204,17 +344,8 @@ describe('Relay region preference', () => { 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] }) + writeCache(path, 'us-central1', 10 * 24 * 60 * 60_000) + const healthy = sampledProbe({ [ASIA]: [90, 30, 32, 34] }) await expect( new RelayRegionPreferenceResolver({ @@ -241,6 +372,27 @@ describe('Relay region preference', () => { expect(fetch).not.toHaveBeenCalled() }) + it('lets the environment override win and never self-heals its cache', async () => { + const path = userDataPath() + writeCache(path, 'us-central1', 50_000_000) + vi.stubEnv('ORCA_RELAY_REGION_OVERRIDE', 'asia-east2') + const fetch = vi.fn() + const resolver = new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch, + probe: async () => 900, + now: () => 1_000 + }) + + await expect(resolver.resolve()).resolves.toBe('asia-east2') + await resolver.invalidateIfAssignedCellIsFar(CELL) + expect(fetch).not.toHaveBeenCalled() + expect(JSON.parse(readFileSync(cachePath(path), 'utf8'))).toMatchObject({ + region: 'us-central1' + }) + }) + it('bounds an offline catalog request and returns no preference', async () => { const fetch = vi.fn( async (_url, init) => @@ -276,3 +428,89 @@ describe('Relay region preference', () => { expect(cancelled).toBe(1) }) }) + +describe('Relay region cache self-heal', () => { + const LIVE_EXPIRY = 50_000_000 + + function resolverFor(path: string, cellMs: number[]) { + const { calls, probe } = sampledProbe({ + [US]: [300, 36, 38, 40], + [ASIA]: [400, 218, 220, 222], + [CELL]: cellMs + }) + const fetch = catalogFetch(BOTH_REGIONS) + return { + calls, + fetch, + resolver: new RelayRegionPreferenceResolver({ + directorUrl: DIRECTOR, + userDataPath: path, + fetch, + probe, + now: () => 1_000 + }) + } + } + + it('deletes a cache that names the wrong region once the cell measures far', async () => { + const path = userDataPath() + writeCache(path, 'asia-east2', LIVE_EXPIRY) + const { calls, resolver } = resolverFor(path, [800, 700, 710, 720]) + + await resolver.invalidateIfAssignedCellIsFar(CELL) + expect(existsSync(cachePath(path))).toBe(false) + expect(calls.filter((origin) => origin === CELL)).toHaveLength(4) + }) + + it('keeps a wrong cache whose assigned cell is close to the best region', async () => { + const path = userDataPath() + writeCache(path, 'asia-east2', LIVE_EXPIRY) + const { resolver } = resolverFor(path, [300, 40, 42, 44]) + + await resolver.invalidateIfAssignedCellIsFar(CELL) + expect(JSON.parse(readFileSync(cachePath(path), 'utf8'))).toMatchObject({ + region: 'asia-east2' + }) + }) + + it('keeps a correct cache the director placed away from, without probing the cell', async () => { + const path = userDataPath() + writeCache(path, 'us-central1', LIVE_EXPIRY) + const { calls, resolver } = resolverFor(path, [800, 700, 710, 720]) + + await resolver.invalidateIfAssignedCellIsFar(CELL) + expect(JSON.parse(readFileSync(cachePath(path), 'utf8'))).toMatchObject({ + region: 'us-central1' + }) + expect(calls.filter((origin) => origin === CELL)).toHaveLength(0) + }) + + it.each([ + { name: 'no cache', write: () => {} }, + { name: 'an expired cache', write: (path: string) => writeCache(path, 'asia-east2', 999) }, + { name: 'a no-hint cache', write: (path: string) => writeNoHintCache(path, LIVE_EXPIRY) } + ])('skips the probes and stays unarmed for $name', async ({ write }) => { + const path = userDataPath() + write(path) + const first = resolverFor(path, [800, 700, 710, 720]) + + await first.resolver.invalidateIfAssignedCellIsFar(CELL) + expect(first.fetch).not.toHaveBeenCalled() + expect(first.calls).toHaveLength(0) + + // Nothing was checked, so a cache written later must still be checkable. + writeCache(path, 'asia-east2', LIVE_EXPIRY) + await first.resolver.invalidateIfAssignedCellIsFar(CELL) + expect(existsSync(cachePath(path))).toBe(false) + }) + + it('probes a given cell only once per process', async () => { + const path = userDataPath() + writeCache(path, 'asia-east2', LIVE_EXPIRY) + const { calls, resolver } = resolverFor(path, [800, 700, 710, 720]) + + await resolver.invalidateIfAssignedCellIsFar(CELL) + await resolver.invalidateIfAssignedCellIsFar(CELL) + expect(calls.filter((origin) => origin === CELL)).toHaveLength(4) + }) +}) diff --git a/src/main/runtime/relay/relay-region-preference.ts b/src/main/runtime/relay/relay-region-preference.ts index a35ee7f0815..e435263db0d 100644 --- a/src/main/runtime/relay/relay-region-preference.ts +++ b/src/main/runtime/relay/relay-region-preference.ts @@ -1,78 +1,49 @@ -import { existsSync, readFileSync, statSync } from 'node:fs' +import { existsSync, readFileSync, rmSync, 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' +import { + measureOriginLatency, + RELAY_REGIONS, + measureRegion, + probeRelayOrigin, + PROBE_TIMEOUT_MS, + RelayRegionCatalogSchema, + RelayRegionSchema, + type RegionMeasurement, + type RelayProbe, + type RelayRegion, + type RelayRegionCatalog +} from './relay-region-probe' -export const RELAY_REGIONS = ['us-central1', 'asia-east2'] as const -export type RelayRegion = (typeof RELAY_REGIONS)[number] +export { RELAY_REGIONS, type RelayRegion } from './relay-region-probe' 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 +// A withheld hint is cheap to revisit but expensive to re-measure on every +// reconnect, so it is remembered for far less time than a chosen region. +const NO_HINT_TTL_MS = 60 * 60_000 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 FAR_CELL_RATIO = 3 const RelayRegionCacheSchema = z .object({ v: z.literal(1), directorUrl: z.string().max(2_048), - region: RelayRegionSchema, - latencyMs: z.number().finite().nonnegative().max(60_000), + // Null records a deliberate "no hint"; the field is absent only for a region. + region: RelayRegionSchema.nullable(), + latencyMs: z.number().finite().nonnegative().max(60_000).optional(), 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 @@ -81,30 +52,29 @@ type RelayRegionPreferenceOptions = { now?: () => number measureNow?: () => number diagnosticOverride?: string - probe?: (origin: string) => Promise + probe?: RelayProbe requestTimeoutMs?: number } export class RelayRegionPreferenceResolver { private readonly options: RelayRegionPreferenceOptions private pending: Promise | null = null + private readonly selfHealedCells = new Set() 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 override = this.overrideRegion() + if (override) { + return override } const now = (this.options.now ?? Date.now)() - const cache = readRelayRegionCache(this.options.userDataPath, this.options.directorUrl, now) + const cache = readRelayRegionCache(this.cachePath(), this.options.directorUrl, now) if (cache && cache.expiresAt > now) { - return cache.region + return cache.region ?? undefined } if (this.pending) { return await this.pending @@ -118,17 +88,90 @@ export class RelayRegionPreferenceResolver { } } + // Why: a cache written from a bad measurement pins the desktop to a distant + // cell for a full day. Probing the cell we actually landed on catches that. + async invalidateIfAssignedCellIsFar(assignedCellOrigin: string): Promise { + if (this.overrideRegion() || this.selfHealedCells.has(assignedCellOrigin)) { + return + } + const now = (this.options.now ?? Date.now)() + const cache = readRelayRegionCache(this.cachePath(), this.options.directorUrl, now) + // An absent, expired, or no-hint cache is already re-measured by resolve(). + if (!cache?.region || cache.expiresAt <= now) { + return + } + this.selfHealedCells.add(assignedCellOrigin) + try { + const fetch = this.options.fetch ?? globalThis.fetch + const catalog = await this.fetchCatalog(fetch) + const probe = this.createProbe(fetch) + const best = bestMeasurement(await measureCatalogRegions(catalog, probe)) + // A far cell under a cache that still names the best region is the + // director declining the hint; deleting it would only re-probe. + if (!best || best.region === cache.region) { + return + } + const assignedMs = await measureOriginLatency(assignedCellOrigin, probe) + if (assignedMs !== null && assignedMs > best.latencyMs * FAR_CELL_RATIO) { + rmSync(this.cachePath(), { force: true }) + } + } catch { + // Self-heal is best effort; a failed probe must never disturb the session. + } + } + 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 catalog = await this.fetchCatalog(fetch) + const measurements = await measureCatalogRegions(catalog, this.createProbe(fetch)) + // Why: a region may only win against a measured competitor. With a rejected + // or unmeasurable peer, director default placement beats a lone survivor. + const selected = + measurements.length < catalog.regions.length + ? null + : selectRegionMeasurement(measurements, previous?.region ?? null) + this.writeCache( + selected + ? { region: selected.region, latencyMs: selected.latencyMs, ttlMs: CACHE_TTL_MS } + : { region: null, ttlMs: NO_HINT_TTL_MS }, + now ) - const probe = + return selected?.region + } + + private writeCache( + entry: { region: RelayRegion | null; latencyMs?: number; ttlMs: number }, + now: number + ): void { + try { + writeSecureJsonFile(this.cachePath(), { + v: 1, + directorUrl: this.options.directorUrl, + region: entry.region, + ...(entry.latencyMs === undefined ? {} : { latencyMs: entry.latencyMs }), + expiresAt: now + entry.ttlMs + } satisfies RelayRegionCache) + } catch { + // A cache write must not block an otherwise valid Relay assignment. + } + } + + private overrideRegion(): RelayRegion | undefined { + const override = RelayRegionSchema.safeParse( + this.options.diagnosticOverride ?? process.env.ORCA_RELAY_REGION_OVERRIDE + ) + return override.success ? override.data : undefined + } + + private cachePath(): string { + return join(this.options.userDataPath, RELAY_REGION_CACHE_FILENAME) + } + + private createProbe(fetch: typeof globalThis.fetch): RelayProbe { + return ( this.options.probe ?? ((origin: string) => probeRelayOrigin( @@ -137,38 +180,41 @@ export class RelayRegionPreferenceResolver { 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 + private async fetchCatalog(fetch: typeof globalThis.fetch): Promise { + return await fetchRelayRegionCatalog( + this.options.directorUrl, + fetch, + this.options.requestTimeoutMs ?? PROBE_TIMEOUT_MS + ) } } export function createRelayRegionPreferenceReader(input: { authConfig: { relayDirectorUrl: string } userDataPath: string -}): () => Promise { +}): { + resolvePreferredRegion: () => Promise + noteAssignedCell: (cellUrl: string) => void +} { const resolver = new RelayRegionPreferenceResolver({ directorUrl: input.authConfig.relayDirectorUrl, userDataPath: input.userDataPath }) - return () => resolver.resolve() + return { + resolvePreferredRegion: () => resolver.resolve(), + noteAssignedCell: (cellUrl) => void resolver.invalidateIfAssignedCellIsFar(cellUrl) + } +} + +async function measureCatalogRegions( + catalog: RelayRegionCatalog, + probe: RelayProbe +): Promise { + const measured = await Promise.all(catalog.regions.map((entry) => measureRegion(entry, probe))) + return measured.filter((measurement): measurement is RegionMeasurement => measurement !== null) } async function fetchRelayRegionCatalog( @@ -204,68 +250,25 @@ async function fetchRelayRegionCatalog( 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 bestMeasurement(measurements: RegionMeasurement[]): RegionMeasurement | null { + const order = new Map(RELAY_REGIONS.map((region, index) => [region, index])) + return ( + [...measurements].sort( + (left, right) => + left.latencyMs - right.latencyMs || order.get(left.region)! - order.get(right.region)! + )[0] ?? null + ) } function selectRegionMeasurement( measurements: RegionMeasurement[], - previous: RelayRegionCache | null + previousRegion: RelayRegion | 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 best = bestMeasurement(measurements) + if (!best || !previousRegion || best.region === previousRegion) { + return best } - const current = measurements.find((measurement) => measurement.region === previous.region) + const current = measurements.find((measurement) => measurement.region === previousRegion) if (!current) { return best } @@ -275,8 +278,7 @@ function selectRegionMeasurement( return meaningful ? best : current } -function readRelayRegionCache(userDataPath: string, directorUrl: string, now: number) { - const path = join(userDataPath, RELAY_REGION_CACHE_FILENAME) +function readRelayRegionCache(path: string, directorUrl: string, now: number) { try { if (!existsSync(path)) { return null @@ -296,15 +298,6 @@ function readRelayRegionCache(userDataPath: string, directorUrl: string, now: nu } } -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) diff --git a/src/main/runtime/relay/relay-region-probe.ts b/src/main/runtime/relay/relay-region-probe.ts new file mode 100644 index 00000000000..d0843b73541 --- /dev/null +++ b/src/main/runtime/relay/relay-region-probe.ts @@ -0,0 +1,140 @@ +import { performance } from 'node:perf_hooks' +import { z } from 'zod' +import { cancelUnreadResponseBody } from '../../lib/unread-response-body' + +export const RELAY_REGIONS = ['us-central1', 'asia-east2'] as const +export type RelayRegion = (typeof RELAY_REGIONS)[number] + +export const PROBE_TIMEOUT_MS = 1_500 +const PROBE_SAMPLES = 3 +// Absolute floor for the flap check: a warmed keep-alive path still jitters, and +// a floor below TLS-scale noise rejects healthy regions on nearly every run. +const SPREAD_FLOOR_MS = 150 + +export const RelayRegionSchema = z.enum(RELAY_REGIONS) +export const RelayProbeOriginSchema = z.string().max(2_048).refine(isCanonicalHttpsOrigin) +export 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) + } + } + }) + +export type RelayRegionCatalog = z.infer +export type RelayRegionCatalogEntry = RelayRegionCatalog['regions'][number] +export type RegionMeasurement = { region: RelayRegion; latencyMs: number } +export type RelayProbe = (origin: string) => Promise + +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 + } +} + +// The first request of a process pays TCP and TLS setup, which can exceed the +// round trip it is meant to measure, so it is discarded before sampling. +async function sampleMinLatencies(origins: string[], probe: RelayProbe): Promise { + const warmup = await Promise.all(origins.map(probe)) + // An origin that failed its warm-up would spend one probe timeout per round + // to report nothing, so the sampling rounds skip it entirely. + const live = origins.filter((_origin, index) => warmup[index] !== null) + if (live.length === 0) { + return null + } + const samples: number[] = [] + for (let sample = 0; sample < PROBE_SAMPLES; sample++) { + const latencies = (await Promise.all(live.map(probe))).filter( + (latency): latency is number => latency !== null + ) + if (latencies.length === 0) { + return null + } + samples.push(Math.min(...latencies)) + } + return samples.sort((left, right) => left - right) +} + +export async function measureOriginLatency( + origin: string, + probe: RelayProbe +): Promise { + return (await sampleMinLatencies([origin], probe))?.[0] ?? null +} + +export async function measureRegion( + entry: RelayRegionCatalogEntry, + probe: RelayProbe +): Promise { + const samples = await sampleMinLatencies(entry.probeOrigins, probe) + if (!samples) { + return null + } + const [min, median, max] = samples as [number, number, number] + // Regions compare by their best round trip; the spread check only rejects a + // path that is genuinely flapping, not one that warmed up. + if (max - min > Math.max(SPREAD_FLOOR_MS, median)) { + return null + } + return { region: entry.region, latencyMs: min } +} + +function isCanonicalHttpsOrigin(value: string): boolean { + try { + const url = new URL(value) + return url.protocol === 'https:' && url.origin === value + } catch { + return false + } +} diff --git a/src/main/runtime/relay/relay-session-broker-contract.ts b/src/main/runtime/relay/relay-session-broker-contract.ts index 377ef90416f..c48c2bb6a07 100644 --- a/src/main/runtime/relay/relay-session-broker-contract.ts +++ b/src/main/runtime/relay/relay-session-broker-contract.ts @@ -23,6 +23,7 @@ export type RelaySessionBrokerOptions = { isCurrent: () => boolean refreshAccessToken: () => Promise resolvePreferredRegion?: () => Promise + onAssignedCellActive?: (cellUrl: string) => void 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 6f27b4f2e14..7e058333733 100644 --- a/src/main/runtime/relay/relay-session-broker.test.ts +++ b/src/main/runtime/relay/relay-session-broker.test.ts @@ -312,6 +312,40 @@ describe('RelaySessionBroker lifecycle ownership', () => { expect(fakes.controls[1]!.confirmResume).toHaveBeenCalledOnce() }) + it('reports the assigned cell each time an origin registers', async () => { + fakes.controlConnect.mockResolvedValue({ + type: 'host-hello-ack', + v: 1, + generation: 1, + controlResumeSecret: 'A'.repeat(43), + leaseExpiresAt: 1_000_000, + activeConnIds: [], + pendingConns: [] + } satisfies RelayHostHelloAckMessage) + fakes.assign + .mockResolvedValueOnce({ + cellUrl: 'https://cell-a.relay.example.test', + assignmentEpoch: 1, + leaseExpiresAt: 1_000_000 + }) + .mockResolvedValueOnce({ + cellUrl: 'https://cell-b.relay.example.test', + assignmentEpoch: 2, + leaseExpiresAt: 2_000_000 + }) + const onAssignedCellActive = vi.fn() + + await RelaySessionBroker.connect(brokerOptions({ onAssignedCellActive })) + expect(onAssignedCellActive.mock.calls).toEqual([['https://cell-a.relay.example.test']]) + fakes.controls[0]!.options.onDrain({ + type: 'drain', + graceMs: 5_000, + recovery: 'resolve-director' + }) + await vi.waitFor(() => expect(onAssignedCellActive).toHaveBeenCalledTimes(2)) + expect(onAssignedCellActive).toHaveBeenLastCalledWith('https://cell-b.relay.example.test') + }) + it('opens a fresh same-cell generation when process-local rebind state is lost', async () => { const ack: RelayHostHelloAckMessage = { type: 'host-hello-ack', diff --git a/src/main/runtime/relay/relay-session-broker.ts b/src/main/runtime/relay/relay-session-broker.ts index cd83545e9ca..6ff8f3e3bf2 100644 --- a/src/main/runtime/relay/relay-session-broker.ts +++ b/src/main/runtime/relay/relay-session-broker.ts @@ -293,8 +293,15 @@ export class RelaySessionBroker { } private publishStatus(status: RelayBrokerStatus): void { - if (this.isCurrent()) { - this.options.onStatus(status) + if (!this.isCurrent()) { + return + } + this.options.onStatus(status) + const cellUrl = this.originPool.activeAssignment?.cellUrl + if (status === 'registered' && cellUrl) { + // Fire-and-forget: the listener may probe this cell, and nothing about the + // live session is allowed to wait on that. + this.options.onAssignedCellActive?.(cellUrl) } } }