fix(desktop): copy IPC payloads out of the shared msgpack buffer (#49696)

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<typeof createIpcCodec>
   readonly onMessage: (event: Electron.MessageEvent) => void
   readonly onClose: () => void
   readonly unbindEvents: Effect.Effect<void>
@@ -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

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<RpcMessage.FromServerEncoded>()
         const onMessage = (event: MessageEvent) => {
           try {

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])
+  })
+})

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
+    },
+  }
+}