diff --git a/packages/desktop/src/main/ipc-transport.ts b/packages/desktop/src/main/ipc-transport.ts index 4cdddaa7bf5..1a695097007 100644 --- a/packages/desktop/src/main/ipc-transport.ts +++ b/packages/desktop/src/main/ipc-transport.ts @@ -1,13 +1,14 @@ import type { MessagePortMain, WebContents } from "electron" import { Context, Effect, Layer, Option, Queue, Stream } from "effect" import { RpcMessage, RpcSerialization, RpcServer } from "effect/unstable/rpc" +import { createIpcCodec } from "../shared/ipc-codec" import { bindIpcEvents } from "./ipc-events" type PortBinding = { readonly id: number readonly sender: WebContents readonly port: MessagePortMain - readonly parser: RpcSerialization.Parser + readonly parser: ReturnType readonly onMessage: (event: Electron.MessageEvent) => void readonly onClose: () => void readonly unbindEvents: Effect.Effect @@ -58,7 +59,7 @@ export const IpcServerProtocolLive = Layer.unwrap( } const id = nextClientId++ - const parser = serialization.makeUnsafe() + const parser = createIpcCodec(serialization) const onMessage = (event: Electron.MessageEvent) => { try { parser diff --git a/packages/desktop/src/renderer/ipc-client.ts b/packages/desktop/src/renderer/ipc-client.ts index f48ae724565..da84a0f9271 100644 --- a/packages/desktop/src/renderer/ipc-client.ts +++ b/packages/desktop/src/renderer/ipc-client.ts @@ -1,5 +1,6 @@ import { Context, Effect, Layer, ManagedRuntime, Queue, Stream } from "effect" import { RpcClient, RpcMessage, RpcSerialization } from "effect/unstable/rpc" +import { createIpcCodec } from "../shared/ipc-codec" import { DesktopRpcs, type DesktopRpcClient } from "../shared/ipc-rpc" import type { DesktopEvent } from "../shared/ipc-rpc/events" import { IpcTransportPort } from "../shared/ipc-transport" @@ -85,7 +86,7 @@ function clientProtocol(value: MessagePort) { RpcClient.Protocol.make( Effect.fnUntraced(function* (writeResponse, clientIds) { const serialization = yield* RpcSerialization.RpcSerialization - const parser = serialization.makeUnsafe() + const parser = createIpcCodec(serialization) const inbound = yield* Queue.unbounded() const onMessage = (event: MessageEvent) => { try { diff --git a/packages/desktop/src/shared/ipc-codec.test.ts b/packages/desktop/src/shared/ipc-codec.test.ts new file mode 100644 index 00000000000..c6dcf200ed5 --- /dev/null +++ b/packages/desktop/src/shared/ipc-codec.test.ts @@ -0,0 +1,61 @@ +import { describe, expect, test } from "bun:test" +import { RpcSerialization } from "effect/unstable/rpc" +import { createIpcCodec, ipcLargeMessageBytes } from "./ipc-codec" + +type Serialization = RpcSerialization.RpcSerialization["Service"] + +const msgpack = RpcSerialization.makeMsgPack() + +function counting(serialization: Serialization) { + let created = 0 + const counted: Serialization = { + ...serialization, + makeUnsafe: () => { + created++ + return serialization.makeUnsafe() + }, + } + return { created: () => created, serialization: counted } +} + +describe("ipc codec", () => { + test("posts an exact-size buffer instead of a view into the shared target", () => { + const codec = createIpcCodec(msgpack) + // The msgpack target starts at 8 KiB, so a view would drag a larger backing store along. + const encoded = codec.encode({ _tag: "Request", id: "1", tag: "Ping", payload: {} }) + expect(encoded).toBeInstanceOf(Uint8Array) + const bytes = encoded as Uint8Array + expect(bytes.byteOffset).toBe(0) + expect(bytes.buffer.byteLength).toBe(bytes.byteLength) + }) + + test("copies Node Buffers, whose slice is only a view", () => { + const backing = new ArrayBuffer(1024 * 1024) + const view = Buffer.from(backing, 16, 32) + view.fill(7) + const stub: Serialization = { ...msgpack, makeUnsafe: () => ({ decode: () => [], encode: () => view }) } + const encoded = createIpcCodec(stub).encode({}) as Uint8Array + expect(encoded.buffer).not.toBe(backing) + expect(encoded.buffer.byteLength).toBe(32) + expect([...encoded]).toEqual(Array(32).fill(7)) + }) + + test("replaces the encoder after a large message and keeps the decoder", () => { + const spy = counting(msgpack) + const codec = createIpcCodec(spy.serialization) + expect(spy.created()).toBe(2) + codec.encode({ small: true }) + expect(spy.created()).toBe(2) + codec.encode({ large: "x".repeat(ipcLargeMessageBytes) }) + expect(spy.created()).toBe(3) + codec.decode(codec.encode({ after: 1 }) as Uint8Array) + expect(spy.created()).toBe(3) + }) + + test("round-trips messages through the wrapped serialization", () => { + const client = createIpcCodec(msgpack) + const server = createIpcCodec(msgpack) + const message = { _tag: "Request", id: "7", tag: "DraftsSet", payload: { key: "k", value: "v" } } + expect(server.decode(client.encode(message) as Uint8Array)).toEqual([message]) + }) +}) diff --git a/packages/desktop/src/shared/ipc-codec.ts b/packages/desktop/src/shared/ipc-codec.ts new file mode 100644 index 00000000000..6842f3d5e94 --- /dev/null +++ b/packages/desktop/src/shared/ipc-codec.ts @@ -0,0 +1,25 @@ +import type { RpcSerialization } from "effect/unstable/rpc" + +// After a message this large the encoder is replaced so its grown target buffer can be collected. +export const ipcLargeMessageBytes = 1024 * 1024 + +// The MessagePack parser packs into one shared, grow-only target buffer and returns a view into it. +// Posting that view structured-clones the whole backing buffer, so after one large message every +// later message (even a 55-byte ack) would copy the full grown buffer across processes on each +// send. Encoding through this wrapper posts an exact-size copy instead. Decoding keeps a single +// parser for the connection: record structures the peer defined inline must stay known. +export function createIpcCodec(serialization: RpcSerialization.RpcSerialization["Service"]) { + const decoder = serialization.makeUnsafe() + let encoder = serialization.makeUnsafe() + return { + decode: (bytes: Uint8Array | string) => decoder.decode(bytes), + encode(message: unknown) { + const encoded = encoder.encode(message) + if (!(encoded instanceof Uint8Array)) return encoded + // Not `.slice()`: in the main process the packer hands out a Node Buffer, whose slice is a view. + const copy = new Uint8Array(encoded) + if (copy.byteLength > ipcLargeMessageBytes) encoder = serialization.makeUnsafe() + return copy + }, + } +}