From 09c8597fb7cf35bfcbc7b8487d078891172fd7d6 Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Tue, 11 Aug 2026 16:17:17 -0700 Subject: [PATCH] perf(orchestration): route federated reads over shared control (#13814) --- ...environment-federated-read-routing.test.ts | 173 ++++++++++++++++++ ...ent-federated-read-transport.bench.test.ts | 116 ++++++++++++ ...runtime-environment-request-connections.ts | 15 +- .../runtime-environment-transport-routing.ts | 47 ++++- src/shared/remote-runtime-client.ts | 10 +- src/shared/remote-runtime-memory-limits.ts | 9 +- ...-runtime-shared-control-connection.test.ts | 31 ++++ ...emote-runtime-shared-control-connection.ts | 9 +- .../remote-runtime-shared-control-requests.ts | 6 +- 9 files changed, 390 insertions(+), 26 deletions(-) create mode 100644 src/main/ipc/runtime-environment-federated-read-routing.test.ts create mode 100644 src/main/ipc/runtime-environment-federated-read-transport.bench.test.ts diff --git a/src/main/ipc/runtime-environment-federated-read-routing.test.ts b/src/main/ipc/runtime-environment-federated-read-routing.test.ts new file mode 100644 index 00000000000..09e649e73f5 --- /dev/null +++ b/src/main/ipc/runtime-environment-federated-read-routing.test.ts @@ -0,0 +1,173 @@ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { encodePairingOffer } from '../../shared/pairing' +import { REMOTE_RUNTIME_SHARED_CONTROL_CAPABILITY } from '../../shared/protocol-version' +import { addEnvironmentFromPairingCode } from '../../shared/runtime-environment-store' + +const { sendRemoteRuntimeRequestMock, sendRemoteRuntimeSharedControlRequestMock } = vi.hoisted( + () => ({ + sendRemoteRuntimeRequestMock: vi.fn(), + sendRemoteRuntimeSharedControlRequestMock: vi.fn() + }) +) + +vi.mock('../../shared/remote-runtime-client', () => ({ + sendRemoteRuntimeRequest: sendRemoteRuntimeRequestMock +})) + +vi.mock('./runtime-environment-request-connections', () => ({ + sendRemoteRuntimeConnectionRequest: vi.fn(), + sendRemoteRuntimeSharedControlRequest: sendRemoteRuntimeSharedControlRequestMock, + reconnectRemoteRuntimeSharedControlConnection: vi.fn() +})) + +import { + callRuntimeEnvironment, + resetSharedControlSupport +} from './runtime-environment-transport-routing' + +describe('federated read RPC transport routing', () => { + let userDataPath: string + let environmentId: string + + beforeEach(() => { + userDataPath = mkdtempSync(join(tmpdir(), 'orca-federated-read-routing-')) + environmentId = addEnvironmentFromPairingCode(userDataPath, { + name: 'worker', + pairingCode: encodePairingOffer({ + v: 2, + endpoint: 'ws://127.0.0.1:6768', + deviceToken: 'device-token', + publicKeyB64: Buffer.from(new Uint8Array(32).fill(1)).toString('base64') + }) + }).id + resetSharedControlSupport() + sendRemoteRuntimeRequestMock.mockReset() + sendRemoteRuntimeSharedControlRequestMock.mockReset() + }) + + afterEach(() => { + rmSync(userDataPath, { recursive: true, force: true }) + }) + + it('routes an enveloped federation read over shared control with its authority intact', async () => { + const envelope = { + orchestrationCapability: 'capability', + orchestrationContractVersion: 1, + orchestrationRequestId: 'request-1', + compatibilityInvocationId: 'compatibility-1', + orchestrationCompatibilityEvidence: { + terminalHandle: 'term-1', + paneKey: 'pane-1', + launchToken: 'launch-1' + } + } + sendRemoteRuntimeRequestMock.mockResolvedValue({ + id: 'status', + ok: true, + result: { capabilities: [REMOTE_RUNTIME_SHARED_CONTROL_CAPABILITY] }, + _meta: { runtimeId: 'runtime-worker' } + }) + sendRemoteRuntimeSharedControlRequestMock.mockResolvedValue({ + id: 'pull', + ok: true, + result: { runtimeEpoch: 'runtime-worker', items: [] }, + _meta: { runtimeId: 'runtime-worker' } + }) + + await expect( + callRuntimeEnvironment( + userDataPath, + environmentId, + 'orchestration.federationPull', + { dispatchId: 'dispatch-1', afterSequence: 0, limit: 50 }, + 15_000, + undefined, + envelope + ) + ).resolves.toMatchObject({ ok: true }) + + expect(sendRemoteRuntimeRequestMock.mock.calls.map((call) => call[1])).toEqual(['status.get']) + expect(sendRemoteRuntimeSharedControlRequestMock).toHaveBeenCalledWith( + environmentId, + expect.any(Object), + 'orchestration.federationPull', + { dispatchId: 'dispatch-1', afterSequence: 0, limit: 50 }, + 15_000, + envelope + ) + }) + + it('keeps enveloped federation mutations on the one-shot transport', async () => { + const envelope = { orchestrationContractVersion: 1, orchestrationRequestId: 'ack-1' } + sendRemoteRuntimeRequestMock.mockResolvedValue({ + id: 'ack', + ok: true, + result: { acknowledgedThrough: 4 }, + _meta: { runtimeId: 'runtime-worker' } + }) + + await callRuntimeEnvironment( + userDataPath, + environmentId, + 'orchestration.federationAck', + { dispatchId: 'dispatch-1', throughSequence: 4 }, + 15_000, + undefined, + envelope + ) + + expect(sendRemoteRuntimeRequestMock).toHaveBeenCalledWith( + expect.any(Object), + 'orchestration.federationAck', + { dispatchId: 'dispatch-1', throughSequence: 4 }, + 15_000, + envelope + ) + expect(sendRemoteRuntimeSharedControlRequestMock).not.toHaveBeenCalled() + }) + + it('falls back to one-shot with the envelope when shared control is unavailable', async () => { + const envelope = { orchestrationContractVersion: 1 } + sendRemoteRuntimeRequestMock.mockImplementation(async (_pairing, method) => + method === 'status.get' + ? { + id: 'status', + ok: true, + result: { capabilities: [] }, + _meta: { runtimeId: 'runtime-worker' } + } + : { + id: 'pull', + ok: true, + result: { runtimeEpoch: 'runtime-worker', items: [] }, + _meta: { runtimeId: 'runtime-worker' } + } + ) + + await callRuntimeEnvironment( + userDataPath, + environmentId, + 'orchestration.federationPull', + { dispatchId: 'dispatch-1', afterSequence: 0, limit: 50 }, + 15_000, + undefined, + envelope + ) + + expect(sendRemoteRuntimeRequestMock.mock.calls.map((call) => call[1])).toEqual([ + 'status.get', + 'orchestration.federationPull' + ]) + expect(sendRemoteRuntimeRequestMock).toHaveBeenLastCalledWith( + expect.any(Object), + 'orchestration.federationPull', + { dispatchId: 'dispatch-1', afterSequence: 0, limit: 50 }, + 15_000, + envelope + ) + expect(sendRemoteRuntimeSharedControlRequestMock).not.toHaveBeenCalled() + }) +}) diff --git a/src/main/ipc/runtime-environment-federated-read-transport.bench.test.ts b/src/main/ipc/runtime-environment-federated-read-transport.bench.test.ts new file mode 100644 index 00000000000..4543c3c9b34 --- /dev/null +++ b/src/main/ipc/runtime-environment-federated-read-transport.bench.test.ts @@ -0,0 +1,116 @@ +import { readFileSync } from 'node:fs' +import { performance } from 'node:perf_hooks' +import { join } from 'node:path' +import { describe, expect, it } from 'vitest' +import { sendRemoteRuntimeRequest } from '../../shared/remote-runtime-client' +import { RemoteRuntimeSharedControlConnection } from '../../shared/remote-runtime-shared-control-connection' +import { RuntimeEnvironmentStoreSchema } from '../../shared/runtime-environments' + +const runLiveBenchmark = process.env.ORCA_FEDERATED_READ_BENCH === '1' + +it('interpolates even-sized benchmark samples', () => expect(percentile([1, 3], 0.5)).toBe(2)) + +describe.runIf(runLiveBenchmark)('federated read RPC transport benchmark', () => { + it('compares one-shot and shared-control latency on one saved runtime', async () => { + const userDataPath = process.env.ORCA_RUNTIME_USER_DATA_PATH + const environmentName = process.env.ORCA_RUNTIME_ENVIRONMENT + if (!userDataPath || !environmentName) { + throw new Error('Set ORCA_RUNTIME_USER_DATA_PATH and ORCA_RUNTIME_ENVIRONMENT.') + } + const store = RuntimeEnvironmentStoreSchema.parse( + JSON.parse(readFileSync(join(userDataPath, 'orca-environments.json'), 'utf8')) + ) + const environment = store.environments.find((entry) => entry.name === environmentName) + if (!environment) { + throw new Error(`Unknown runtime environment: ${environmentName}`) + } + const endpoint = + environment.endpoints.find((entry) => entry.id === environment.preferredEndpointId) ?? + environment.endpoints[0] + if (!endpoint) { + throw new Error(`Runtime environment ${environmentName} has no endpoint.`) + } + const pairing = { + v: 2 as const, + endpoint: endpoint.endpoint, + deviceToken: endpoint.deviceToken, + publicKeyB64: endpoint.publicKeyB64 + } + const params = { + dispatchId: 'sta3880_benchmark_missing', + afterSequence: 0, + limit: 50 + } + const oneShot = await measureRequests(() => + sendRemoteRuntimeRequest(pairing, 'orchestration.federationPull', params, 15_000, { + orchestrationContractVersion: 1 + }) + ) + const shared = new RemoteRuntimeSharedControlConnection(pairing, { + environmentId: environment.id + }) + const sharedControl = await measureRequests(() => + shared.request('orchestration.federationPull', params, 15_000, { + orchestrationContractVersion: 1 + }) + ) + const diagnostics = shared.getDiagnostics() + shared.close() + + expect(diagnostics).toMatchObject({ state: 'ready', pendingRequestCount: 0 }) + process.stdout.write(`${JSON.stringify({ oneShot, sharedControl }, null, 2)}\n`) + }, 120_000) +}) + +async function measureRequests( + request: () => Promise<{ ok: boolean; error?: { code: string } }>, + count = 30 +): Promise<{ + count: number + meanMs: number + medianMs: number + p95Ms: number + minMs: number + maxMs: number + responseCodes: string[] +}> { + for (let index = 0; index < 3; index++) { + await request() + } + const durations: number[] = [] + const responseCodes = new Set() + for (let index = 0; index < count; index++) { + const startedAt = performance.now() + const response = await request() + durations.push(performance.now() - startedAt) + if (!response.ok && response.error) { + responseCodes.add(response.error.code) + } + } + durations.sort((left, right) => left - right) + return { + count, + meanMs: round(durations.reduce((sum, value) => sum + value, 0) / count), + medianMs: round(percentile(durations, 0.5)), + p95Ms: round(percentile(durations, 0.95)), + minMs: round(durations[0] ?? 0), + maxMs: round(durations.at(-1) ?? 0), + responseCodes: [...responseCodes] + } +} + +function percentile(sorted: number[], fraction: number): number { + if (sorted.length === 0) { + return 0 + } + const rank = (sorted.length - 1) * fraction + const lowerIndex = Math.floor(rank) + const upperIndex = Math.ceil(rank) + const lower = sorted[lowerIndex] ?? 0 + const upper = sorted[upperIndex] ?? lower + return lower + (upper - lower) * (rank - lowerIndex) +} + +function round(value: number): number { + return Math.round(value * 100) / 100 +} diff --git a/src/main/ipc/runtime-environment-request-connections.ts b/src/main/ipc/runtime-environment-request-connections.ts index 946a0d30cf8..58fb95dd6e1 100644 --- a/src/main/ipc/runtime-environment-request-connections.ts +++ b/src/main/ipc/runtime-environment-request-connections.ts @@ -1,5 +1,8 @@ import type { PairingOffer } from '../../shared/pairing' -import type { RuntimeRpcResponse } from '../../shared/runtime-rpc-envelope' +import type { + RuntimeOrchestrationEnvelope, + RuntimeRpcResponse +} from '../../shared/runtime-rpc-envelope' import { RemoteRuntimeRequestConnection } from '../../shared/remote-runtime-request-connection' import { RemoteRuntimeSharedControlConnection } from '../../shared/remote-runtime-shared-control-connection' import type { @@ -52,9 +55,15 @@ export function sendRemoteRuntimeSharedControlRequest( pairing: PairingOffer, method: string, params: unknown, - timeoutMs: number + timeoutMs: number, + envelope?: RuntimeOrchestrationEnvelope ): Promise> { - return getSharedControlConnection(environmentId, pairing).request(method, params, timeoutMs) + return getSharedControlConnection(environmentId, pairing).request( + method, + params, + timeoutMs, + envelope + ) } export function subscribeRemoteRuntimeSharedControlRequest( diff --git a/src/main/ipc/runtime-environment-transport-routing.ts b/src/main/ipc/runtime-environment-transport-routing.ts index d144ee087e1..884b0b2874a 100644 --- a/src/main/ipc/runtime-environment-transport-routing.ts +++ b/src/main/ipc/runtime-environment-transport-routing.ts @@ -1,5 +1,6 @@ import { getPreferredPairingOffer } from '../../shared/runtime-environments' import { resolveEnvironment, markEnvironmentUsed } from '../../shared/runtime-environment-store' +import { isOrchestrationMutation } from '../../shared/orchestration-rpc-contract' import type { RuntimeOrchestrationEnvelope, RuntimeRpcResponse @@ -108,7 +109,8 @@ export async function callRuntimeEnvironment( const pairing = getPreferredPairingOffer(currentEnvironment) endpoint = pairing.endpoint const effectiveTimeoutMs = timeoutMs ?? DEFAULT_REMOTE_RUNTIME_TIMEOUT_MS - if (envelope) { + const sharedControlEnvelope = shouldUseSharedControlEnvelope(method, params, envelope) + if (envelope && !sharedControlEnvelope) { const response = await sendRemoteRuntimeRequest( pairing, method, @@ -135,19 +137,36 @@ export async function callRuntimeEnvironment( !shouldUseOneShotRequest(method) && (await supportsSharedControl(userDataPath, currentEnvironment, pairing, effectiveTimeoutMs)) ) { - const response = await sendRemoteRuntimeSharedControlRequest( - currentEnvironment.id, - pairing, - method, - params, - effectiveTimeoutMs - ) + const response = sharedControlEnvelope + ? await sendRemoteRuntimeSharedControlRequest( + currentEnvironment.id, + pairing, + method, + params, + effectiveTimeoutMs, + sharedControlEnvelope + ) + : await sendRemoteRuntimeSharedControlRequest( + currentEnvironment.id, + pairing, + method, + params, + effectiveTimeoutMs + ) markEnvironmentUsedFromResponse(userDataPath, currentEnvironment.id, response) return response } // Why: startup/control-plane RPCs use the proven one-shot path so repo // hydration cannot be coupled to a stale terminal-control connection. - const response = await sendRemoteRuntimeRequest(pairing, method, params, effectiveTimeoutMs) + const response = sharedControlEnvelope + ? await sendRemoteRuntimeRequest( + pairing, + method, + params, + effectiveTimeoutMs, + sharedControlEnvelope + ) + : await sendRemoteRuntimeRequest(pairing, method, params, effectiveTimeoutMs) markEnvironmentUsedFromResponse(userDataPath, currentEnvironment.id, response) return response }) @@ -253,6 +272,16 @@ function shouldUseCachedRequestConnection(method: string): boolean { return method === 'terminal.send' || method === 'terminal.updateViewport' } +function shouldUseSharedControlEnvelope( + method: string, + params: unknown, + envelope: RuntimeOrchestrationEnvelope | undefined +): RuntimeOrchestrationEnvelope | undefined { + return envelope && method.startsWith('orchestration.') && !isOrchestrationMutation(method, params) + ? envelope + : undefined +} + function shouldUseOneShotRequest(method: string): boolean { // Why: snapshot recovery must remain available while a retained shared-control stream is reconnecting after a HUB restart. return method === 'session.tabs.list' || method === 'session.tabs.listAll' diff --git a/src/shared/remote-runtime-client.ts b/src/shared/remote-runtime-client.ts index f147ed4e7bb..c53565dc8a2 100644 --- a/src/shared/remote-runtime-client.ts +++ b/src/shared/remote-runtime-client.ts @@ -144,16 +144,12 @@ async function sendRemoteRuntimeRequestOnSocket( }) const pendingRequest = { preparedRequest: prepareRemoteRuntimeRequest(new Map(), () => - serializeRemoteRuntimePayload({ - id: requestId, + serializeRemoteRuntimeRpcRequest({ + requestId, deviceToken: pairing.deviceToken, method, params, - orchestrationCapability: envelope?.orchestrationCapability, - orchestrationContractVersion: envelope?.orchestrationContractVersion, - orchestrationRequestId: envelope?.orchestrationRequestId, - compatibilityInvocationId: envelope?.compatibilityInvocationId, - orchestrationCompatibilityEvidence: envelope?.orchestrationCompatibilityEvidence + envelope }) ) } diff --git a/src/shared/remote-runtime-memory-limits.ts b/src/shared/remote-runtime-memory-limits.ts index 8152f58ef66..a41409ffdfe 100644 --- a/src/shared/remote-runtime-memory-limits.ts +++ b/src/shared/remote-runtime-memory-limits.ts @@ -3,6 +3,7 @@ import { stringifyJsonWithinByteLimit } from './node-bounded-json-stringify' import { RemoteRuntimeClientError } from './remote-runtime-client-error' +import type { RuntimeOrchestrationEnvelope } from './runtime-rpc-envelope' export const REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES = 4 * 1024 * 1024 export const REMOTE_RUNTIME_MAX_WEBSOCKET_FRAME_BYTES = 8 * 1024 * 1024 + 64 @@ -63,12 +64,18 @@ export function serializeRemoteRuntimeRpcRequest(args: { deviceToken: string method: string params: unknown + envelope?: RuntimeOrchestrationEnvelope }): string { return serializeRemoteRuntimePayload({ id: args.requestId, deviceToken: args.deviceToken, method: args.method, - params: args.params + params: args.params, + orchestrationCapability: args.envelope?.orchestrationCapability, + orchestrationContractVersion: args.envelope?.orchestrationContractVersion, + orchestrationRequestId: args.envelope?.orchestrationRequestId, + compatibilityInvocationId: args.envelope?.compatibilityInvocationId, + orchestrationCompatibilityEvidence: args.envelope?.orchestrationCompatibilityEvidence }) } diff --git a/src/shared/remote-runtime-shared-control-connection.test.ts b/src/shared/remote-runtime-shared-control-connection.test.ts index 66a57a92e52..f85b1d8fa6b 100644 --- a/src/shared/remote-runtime-shared-control-connection.test.ts +++ b/src/shared/remote-runtime-shared-control-connection.test.ts @@ -78,6 +78,37 @@ describe('RemoteRuntimeSharedControlConnection', () => { connection.close() }) + it('preserves orchestration authority fields on shared-control requests', async () => { + const server = await createServer() + const connection = new RemoteRuntimeSharedControlConnection(server.pairing) + const envelope = { + orchestrationCapability: 'capability', + orchestrationContractVersion: 1, + orchestrationRequestId: 'request-1', + compatibilityInvocationId: 'compatibility-1', + orchestrationCompatibilityEvidence: { + terminalHandle: 'term-1', + paneKey: 'pane-1', + launchToken: 'launch-1' + }, + id: 'forged-id', + deviceToken: 'forged-token', + method: 'orchestration.federationAck', + params: { dispatchId: 'forged-dispatch' } + } + + await connection.request('orchestration.federationPull', {}, 1000, envelope) + + expect(server.requests).toContainEqual({ + ...envelope, + id: expect.any(String), + deviceToken: 'device-token', + method: 'orchestration.federationPull', + params: {} + }) + connection.close() + }) + it('does not expose a binary sender on the shared control protocol surface', () => { expect('sendSharedControlEncryptedBinary' in sharedControlProtocol).toBe(false) }) diff --git a/src/shared/remote-runtime-shared-control-connection.ts b/src/shared/remote-runtime-shared-control-connection.ts index 524b6697931..4fb40525245 100644 --- a/src/shared/remote-runtime-shared-control-connection.ts +++ b/src/shared/remote-runtime-shared-control-connection.ts @@ -1,6 +1,5 @@ import WebSocket from 'ws' import type { PairingOffer } from './pairing' -import type { RuntimeRpcResponse } from './runtime-rpc-envelope' import type { RemoteRuntimeClientError } from './remote-runtime-client-error' import { remoteRuntimeUnavailableError } from './remote-runtime-request-frames' import { openSharedControlSocket } from './remote-runtime-shared-control-open' @@ -66,14 +65,16 @@ export class RemoteRuntimeSharedControlConnection { request( method: string, params: unknown, - timeoutMs: number - ): Promise> { - return requestSharedControl({ + timeoutMs: number, + envelope?: Parameters[0]['envelope'] + ): ReturnType> { + return requestSharedControl({ pendingRequests: this.pendingRequests, deviceToken: this.pairing.deviceToken, method, params, timeoutMs, + envelope, ensureReady: () => this.ensureReadyWithTimeout(timeoutMs), send: (requestId) => this.sendRequest(requestId), retireRequestId: (requestId) => this.retiredRequestIds.retire(requestId) diff --git a/src/shared/remote-runtime-shared-control-requests.ts b/src/shared/remote-runtime-shared-control-requests.ts index 8379e53da91..5c55c3d11b6 100644 --- a/src/shared/remote-runtime-shared-control-requests.ts +++ b/src/shared/remote-runtime-shared-control-requests.ts @@ -6,7 +6,7 @@ import { type RemoteRuntimePreparedRequest } from './remote-runtime-prepared-request-admission' import { remoteRuntimeTimeoutError } from './remote-runtime-request-frames' -import type { RuntimeRpcResponse } from './runtime-rpc-envelope' +import type { RuntimeOrchestrationEnvelope, RuntimeRpcResponse } from './runtime-rpc-envelope' import { toRemoteRuntimeClientError } from './remote-runtime-shared-control-protocol' import { rejectSharedControlPendingRequest } from './remote-runtime-shared-control-state' import type { SharedControlPendingRequest } from './remote-runtime-shared-control-types' @@ -19,6 +19,7 @@ export function requestSharedControl(args: { method: string params: unknown timeoutMs: number + envelope?: RuntimeOrchestrationEnvelope ensureReady: () => Promise send: (requestId: string) => void retireRequestId?: (requestId: string) => void @@ -35,7 +36,8 @@ export function requestSharedControl(args: { requestId, deviceToken: args.deviceToken, method: args.method, - params: args.params + params: args.params, + envelope: args.envelope }) ) } catch (error) {