From 0530c8e5127d65e725cc8bd3b07667ec9afa4ed6 Mon Sep 17 00:00:00 2001 From: Luke Parker <10430890+Hona@users.noreply.github.com> Date: Sun, 20 Sep 2026 16:26:08 +1000 Subject: [PATCH] perf(desktop): talk to the main process without the Effect runtime in the renderer (#50112) --- packages/desktop/src/renderer/ipc-client.ts | 142 ++++++++++---------- 1 file changed, 73 insertions(+), 69 deletions(-) diff --git a/packages/desktop/src/renderer/ipc-client.ts b/packages/desktop/src/renderer/ipc-client.ts index 240f1462834..521c9b12068 100644 --- a/packages/desktop/src/renderer/ipc-client.ts +++ b/packages/desktop/src/renderer/ipc-client.ts @@ -1,10 +1,14 @@ -import { Context, Effect, Layer, ManagedRuntime, Queue, Schema, Stream } from "effect" -import { RpcClient, RpcMessage } from "effect/unstable/rpc" -import { DesktopRpcs, type DesktopRpcClient } from "../shared/ipc-rpc" +import type { Effect } from "effect" +import type { RpcMessage } from "effect/unstable/rpc" +import type { DesktopRpcClient } from "../shared/ipc-rpc" import type { DesktopEvent } from "../shared/ipc-rpc/events" import { IpcTransportPort } from "../shared/ipc-transport" -class DesktopClient extends Context.Service()("opencode/desktop/DesktopClient") {} +// The main process serves Effect's RpcServer over a MessagePort; this side speaks its wire format +// directly. Messages cross by structured clone (no serialization layer, binary stays binary), every +// payload in the contract is JSON-native or a Uint8Array, and the main process is trusted, so the +// renderer needs neither the Effect runtime nor the contract's schemas to talk to it. Keeping them +// out of the renderer's initial module graph is worth about a third of its startup script. type EventTag = DesktopEvent["_tag"] type InvokeTag = Exclude @@ -13,55 +17,48 @@ type InvokeResult = ReturnType extends Effect.Effect ? Value : never type EventValue = Extract +type Pending = { + readonly resolve: (value: unknown) => void + readonly reject: (error: unknown) => void + readonly chunk?: (values: ReadonlyArray) => void +} + +const pending = new Map() +const listeners = new Map void>>() +const beforeDispose = new Set<() => Promise | void>() +let nextId = 0 + const port = new Promise((resolve) => { const onMessage = (event: MessageEvent) => { if (event.source !== window || event.data !== IpcTransportPort) return const value = event.ports[0] if (!value) return window.removeEventListener("message", onMessage) + value.addEventListener("message", (message) => receive(value, message.data as RpcMessage.FromServerEncoded)) + value.start() resolve(value) } window.addEventListener("message", onMessage) }) -const ClientProtocolLive = Layer.unwrap(Effect.promise(() => port).pipe(Effect.map((value) => clientProtocol(value)))) -const ClientLive = Layer.effect(DesktopClient, RpcClient.make(DesktopRpcs)).pipe(Layer.provide(ClientProtocolLive)) -const runtime = ManagedRuntime.make(ClientLive) -const listeners = new Map void>>() -const beforeDispose = new Set<() => Promise | void>() -// Let queued work (storage flushes) hand its messages to the port before the runtime goes away. +// Let queued work (storage flushes) hand its messages to the port before it closes. window.addEventListener( "pagehide", - () => void Promise.allSettled([...beforeDispose].map((callback) => callback())).then(() => runtime.dispose()), + () => void Promise.allSettled([...beforeDispose].map((callback) => callback())).then(() => port.then((p) => p.close())), { once: true }, ) +void request("DesktopEvents", null, (values) => { + for (const value of values as ReadonlyArray) listeners.get(value._tag)?.forEach((fn) => fn(value)) +}) + export function onBeforeDispose(callback: () => Promise | void) { beforeDispose.add(callback) return () => beforeDispose.delete(callback) } -runtime.runFork( - Effect.gen(function* () { - const client = yield* DesktopClient - yield* client - .DesktopEvents() - .pipe( - Stream.runForEach((event) => - Effect.sync(() => listeners.get(event._tag)?.forEach((listener) => listener(event))), - ), - ) - }), -) - export function invoke(tag: Tag, ...payload: InvokeArgs): Promise> { - return runtime.runPromise( - Effect.gen(function* () { - const client = yield* DesktopClient - const method = client[tag] as unknown as (...args: ReadonlyArray) => Effect.Effect - return yield* method(...payload) - }), - ) as Promise> + return request(tag, payload[0] ?? null) as Promise> } export function send(tag: Tag, ...payload: InvokeArgs) { @@ -79,42 +76,49 @@ export function listen(tag: Tag, listener: (value: EventVa } } -// Structured clone over the port, like Effect's worker protocol: no serialization layer, so binary -// payloads stay binary. Buffers are cloned rather than transferred: Electron's MessagePortMain -// drops transferred ArrayBuffers, so a request carrying one would never arrive. -function clientProtocol(value: MessagePort) { - return Layer.effect( - RpcClient.Protocol, - RpcClient.Protocol.make( - Effect.fnUntraced(function* (writeResponse, clientIds) { - const inbound = yield* Queue.unbounded() - const onMessage = (event: MessageEvent) => { - Queue.offerUnsafe(inbound, event.data as RpcMessage.FromServerEncoded) - } - value.addEventListener("message", onMessage) - value.start() - yield* Effect.addFinalizer(() => - Effect.sync(() => { - value.removeEventListener("message", onMessage) - value.close() - }), - ) - yield* Stream.fromQueue(inbound).pipe( - Stream.runForEach((message) => - Effect.forEach(clientIds, (clientId) => writeResponse(clientId, message), { discard: true }), - ), - Effect.forkScoped, - ) - return { - codecFor: Schema.toCodecJson, - send: (_clientId, request) => - Effect.sync(() => { - value.postMessage(request) - }), - supportsAck: true, - supportsTransferables: false, - } - }), - ), - ) +function request(tag: string, payload: unknown, chunk?: Pending["chunk"]) { + const id = nextId++ + return new Promise((resolve, reject) => { + pending.set(id, { resolve, reject, chunk }) + const message: RpcMessage.RequestEncoded = { _tag: "Request", id, tag, payload, headers: [] } + void port.then((p) => p.postMessage(message)) + }) +} + +function receive(p: MessagePort, message: RpcMessage.FromServerEncoded) { + switch (message._tag) { + case "Chunk": { + pending.get(Number(message.requestId))?.chunk?.(message.values) + p.postMessage({ _tag: "Ack", requestId: message.requestId } satisfies RpcMessage.AckEncoded) + return + } + case "Exit": { + const id = Number(message.requestId) + const entry = pending.get(id) + pending.delete(id) + if (!entry) return + if (message.exit._tag === "Success") return entry.resolve(message.exit.value) + return entry.reject(failure(message.exit.cause)) + } + case "Defect": { + const error = new Error("Desktop IPC defect", { cause: message.defect }) + pending.forEach((entry) => entry.reject(error)) + pending.clear() + return + } + case "ClientProtocolError": { + console.error("[desktop-ipc] protocol error", message.error) + return + } + } +} + +// The RPC failure a caller sees is the encoded error the handler failed with, as before; defects +// and interrupts surface as errors. +function failure(cause: ReadonlyArray<{ readonly _tag: string; readonly error?: unknown; readonly defect?: unknown }>) { + const failed = cause.find((item) => item._tag === "Fail") + if (failed) return failed.error + const died = cause.find((item) => item._tag === "Die") + if (died) return new Error("Desktop IPC handler failed", { cause: died.defect }) + return new Error("Desktop IPC request interrupted") }