Cap RPC streaming decoder buffers

.changeset/cap-rpc-streaming-buffers.md

@@ -0,0 +1,5 @@
+---
+"@effect/rpc": patch
+---
+
+Cap NDJSON and MessagePack streaming decoder buffers with configurable limits.

.changeset/configure-cluster-rpc-buffer-limits.md

@@ -0,0 +1,6 @@
+---
+"@effect/platform-node": patch
+"@effect/platform-bun": patch
+---
+
+Allow configuring cluster RPC serialization buffer limits.

packages/platform-bun/src/BunClusterHttp.ts

@@ -55,6 +55,7 @@ export const layer = <
 >(options: {
   readonly transport: "http" | "websocket"
   readonly serialization?: "msgpack" | "ndjson" | undefined
+  readonly serializationMaxBufferSize?: number | "unbounded" | undefined
   readonly clientOnly?: ClientOnly | undefined
   readonly storage?: Storage | undefined
   readonly shardingConfig?: Partial<ShardingConfig.ShardingConfig["Type"]> | undefined
@@ -117,7 +118,9 @@ export const layer = <
     ),
     Layer.provide(ShardingConfig.layerFromEnv(options?.shardingConfig)),
     Layer.provide(
-      options?.serialization === "ndjson" ? RpcSerialization.layerNdjson : RpcSerialization.layerMsgPack
+      options?.serialization === "ndjson"
+        ? RpcSerialization.layerNdjsonWith({ maxBufferSize: options.serializationMaxBufferSize })
+        : RpcSerialization.layerMsgPackWith({ maxBufferSize: options.serializationMaxBufferSize })
     )
   ) as any
 }

packages/platform-bun/src/BunClusterSocket.ts

@@ -40,6 +40,7 @@ export const layer = <
 >(
   options?: {
     readonly serialization?: "msgpack" | "ndjson" | undefined
+    readonly serializationMaxBufferSize?: number | "unbounded" | undefined
     readonly clientOnly?: ClientOnly | undefined
     readonly storage?: Storage | undefined
     readonly shardingConfig?: Partial<ShardingConfig.ShardingConfig["Type"]> | undefined
@@ -95,7 +96,9 @@ export const layer = <
     ),
     Layer.provide(ShardingConfig.layerFromEnv(options?.shardingConfig)),
     Layer.provide(
-      options?.serialization === "ndjson" ? RpcSerialization.layerNdjson : RpcSerialization.layerMsgPack
+      options?.serialization === "ndjson"
+        ? RpcSerialization.layerNdjsonWith({ maxBufferSize: options?.serializationMaxBufferSize })
+        : RpcSerialization.layerMsgPackWith({ maxBufferSize: options?.serializationMaxBufferSize })
     )
   ) as any
 }

packages/platform-node/src/NodeClusterHttp.ts

@@ -46,6 +46,7 @@ export const layer = <
 >(options: {
   readonly transport: "http" | "websocket"
   readonly serialization?: "msgpack" | "ndjson" | undefined
+  readonly serializationMaxBufferSize?: number | "unbounded" | undefined
   readonly clientOnly?: ClientOnly | undefined
   readonly storage?: Storage | undefined
   readonly runnerHealth?: Health | undefined
@@ -112,7 +113,9 @@ export const layer = <
     ),
     Layer.provide(ShardingConfig.layerFromEnv(options?.shardingConfig)),
     Layer.provide(
-      options?.serialization === "ndjson" ? RpcSerialization.layerNdjson : RpcSerialization.layerMsgPack
+      options?.serialization === "ndjson"
+        ? RpcSerialization.layerNdjsonWith({ maxBufferSize: options.serializationMaxBufferSize })
+        : RpcSerialization.layerMsgPackWith({ maxBufferSize: options.serializationMaxBufferSize })
     )
   ) as any
 }

packages/platform-node/src/NodeClusterSocket.ts

@@ -47,6 +47,7 @@ export const layer = <
 >(
   options?: {
     readonly serialization?: "msgpack" | "ndjson" | undefined
+    readonly serializationMaxBufferSize?: number | "unbounded" | undefined
     readonly clientOnly?: ClientOnly | undefined
     readonly storage?: Storage | undefined
     readonly runnerHealth?: Health | undefined
@@ -106,7 +107,9 @@ export const layer = <
     ),
     Layer.provide(ShardingConfig.layerFromEnv(options?.shardingConfig)),
     Layer.provide(
-      options?.serialization === "ndjson" ? RpcSerialization.layerNdjson : RpcSerialization.layerMsgPack
+      options?.serialization === "ndjson"
+        ? RpcSerialization.layerNdjsonWith({ maxBufferSize: options?.serializationMaxBufferSize })
+        : RpcSerialization.layerMsgPackWith({ maxBufferSize: options?.serializationMaxBufferSize })
     )
   ) as any
 }

packages/rpc/src/RpcSerialization.ts

@@ -2,11 +2,44 @@
  * @since 1.0.0
  */
 import * as Context from "effect/Context"
+import * as Data from "effect/Data"
 import * as Layer from "effect/Layer"
 import { hasProperty } from "effect/Predicate"
 import * as Msgpackr from "msgpackr"
 import type * as RpcMessage from "./RpcMessage.js"
 
+const defaultMaxBufferSize = 16 * 1024 * 1024
+
+/**
+ * @since 1.0.0
+ * @category errors
+ */
+export class RpcSerializationError extends Data.TaggedError("RpcSerializationError")<{
+  readonly reason: "BufferSizeExceeded"
+  readonly maxBufferSize: number
+  readonly bufferSize: number
+}> {
+  /**
+   * @since 1.0.0
+   */
+  get message() {
+    return `RPC serialization buffer exceeded the maximum size of ${this.maxBufferSize}`
+  }
+}
+
+/**
+ * @since 1.0.0
+ * @category serialization
+ */
+export interface StreamOptions {
+  readonly maxBufferSize?: number | "unbounded" | undefined
+}
+
+const isBufferSizeExceeded = (
+  bufferSize: number,
+  maxBufferSize: number | "unbounded"
+): maxBufferSize is number => maxBufferSize !== "unbounded" && bufferSize > maxBufferSize
+
 /**
  * @since 1.0.0
  * @category serialization
@@ -46,41 +79,68 @@ export const json: RpcSerialization["Type"] = RpcSerialization.of({
  * @since 1.0.0
  * @category serialization
  */
-export const ndjson: RpcSerialization["Type"] = RpcSerialization.of({
-  contentType: "application/ndjson",
-  includesFraming: true,
-  unsafeMake: () => {
-    const decoder = new TextDecoder()
-    let buffer = ""
-    return ({
-      decode: (bytes) => {
-        buffer += typeof bytes === "string" ? bytes : decoder.decode(bytes)
-        let position = 0
-        let nlIndex = buffer.indexOf("\n", position)
-        const items: Array<unknown> = []
-        while (nlIndex !== -1) {
-          const item = JSON.parse(buffer.slice(position, nlIndex))
-          items.push(item)
-          position = nlIndex + 1
-          nlIndex = buffer.indexOf("\n", position)
-        }
-        buffer = buffer.slice(position)
-        return items
-      },
-      encode: (response) => {
-        if (Array.isArray(response)) {
-          if (response.length === 0) return undefined
-          let data = ""
-          for (let i = 0; i < response.length; i++) {
-            data += JSON.stringify(response[i]) + "\n"
+export const makeNdjson = (options?: StreamOptions): RpcSerialization["Type"] => {
+  const maxBufferSize = options?.maxBufferSize ?? defaultMaxBufferSize
+  return RpcSerialization.of({
+    contentType: "application/ndjson",
+    includesFraming: true,
+    unsafeMake: () => {
+      const decoder = new TextDecoder()
+      let buffer = ""
+      return ({
+        decode: (bytes) => {
+          buffer += typeof bytes === "string" ? bytes : decoder.decode(bytes)
+          let position = 0
+          let nlIndex = buffer.indexOf("\n", position)
+          const items: Array<unknown> = []
+          while (nlIndex !== -1) {
+            const bufferSize = nlIndex - position
+            if (isBufferSizeExceeded(bufferSize, maxBufferSize)) {
+              buffer = ""
+              throw new RpcSerializationError({
+                reason: "BufferSizeExceeded",
+                maxBufferSize,
+                bufferSize
+              })
+            }
+            const item = JSON.parse(buffer.slice(position, nlIndex))
+            items.push(item)
+            position = nlIndex + 1
+            nlIndex = buffer.indexOf("\n", position)
+          }
+          buffer = buffer.slice(position)
+          const bufferSize = buffer.length
+          if (isBufferSizeExceeded(bufferSize, maxBufferSize)) {
+            buffer = ""
+            throw new RpcSerializationError({
+              reason: "BufferSizeExceeded",
+              maxBufferSize,
+              bufferSize
+            })
+          }
+          return items
+        },
+        encode: (response) => {
+          if (Array.isArray(response)) {
+            if (response.length === 0) return undefined
+            let data = ""
+            for (let i = 0; i < response.length; i++) {
+              data += JSON.stringify(response[i]) + "\n"
+            }
+            return data
           }
-          return data
+          return JSON.stringify(response) + "\n"
         }
-        return JSON.stringify(response) + "\n"
-      }
-    })
-  }
-})
+      })
+    }
+  })
+}
+
+/**
+ * @since 1.0.0
+ * @category serialization
+ */
+export const ndjson: RpcSerialization["Type"] = makeNdjson()
 
 /**
  * @since 1.0.0
@@ -119,12 +179,13 @@ export const jsonRpc = (options?: {
  */
 export const ndJsonRpc = (options?: {
   readonly contentType?: string | undefined
+  readonly maxBufferSize?: number | "unbounded" | undefined
 }): RpcSerialization["Type"] =>
   RpcSerialization.of({
     contentType: options?.contentType ?? "application/json-rpc",
     includesFraming: true,
     unsafeMake: () => {
-      const parser = ndjson.unsafeMake()
+      const parser = makeNdjson(options).unsafeMake()
       const batches = new Map<string, {
         readonly size: number
         readonly responses: Map<string, RpcMessage.FromServerEncoded>
@@ -397,13 +458,16 @@ type JsonRpcMessage = JsonRpcRequest | JsonRpcResponse
  * @since 1.0.0
  * @category serialization
  */
-export const makeMsgPack = (options?: Msgpackr.Options | undefined): RpcSerialization["Type"] =>
-  RpcSerialization.of({
+export const makeMsgPack = (
+  options?: Msgpackr.Options & StreamOptions | undefined
+): RpcSerialization["Type"] => {
+  const { maxBufferSize = defaultMaxBufferSize, ...msgpackOptions } = options ?? {}
+  return RpcSerialization.of({
     contentType: "application/msgpack",
     includesFraming: true,
     unsafeMake() {
-      const unpackr = new Msgpackr.Unpackr(options)
-      const packr = new Msgpackr.Packr(options)
+      const unpackr = new Msgpackr.Unpackr(msgpackOptions)
+      const packr = new Msgpackr.Packr(msgpackOptions)
       const encoder = new TextEncoder()
       let incomplete: Uint8Array | undefined = undefined
       return {
@@ -422,7 +486,16 @@ export const makeMsgPack = (options?: Msgpackr.Options | undefined): RpcSerializ
           } catch (error_) {
             const error = error_ as any
             if (error.incomplete) {
-              incomplete = buf.subarray(error.lastPosition)
+              const nextIncomplete = buf.subarray(error.lastPosition)
+              const bufferSize = nextIncomplete.length
+              if (isBufferSizeExceeded(bufferSize, maxBufferSize)) {
+                throw new RpcSerializationError({
+                  reason: "BufferSizeExceeded",
+                  maxBufferSize,
+                  bufferSize
+                })
+              }
+              incomplete = nextIncomplete
               return error.values ?? []
             }
             throw error_
@@ -432,6 +505,7 @@ export const makeMsgPack = (options?: Msgpackr.Options | undefined): RpcSerializ
       }
     }
   })
+}
 
 /**
  * @since 1.0.0
@@ -461,6 +535,13 @@ export const layerJson: Layer.Layer<RpcSerialization> = Layer.succeed(RpcSeriali
  */
 export const layerNdjson: Layer.Layer<RpcSerialization> = Layer.succeed(RpcSerialization, ndjson)
 
+/**
+ * @since 1.0.0
+ * @category serialization
+ */
+export const layerNdjsonWith = (options?: StreamOptions): Layer.Layer<RpcSerialization> =>
+  Layer.succeed(RpcSerialization, makeNdjson(options))
+
 /**
  * A rpc serialization layer that uses JSON-RPC for serialization.
  *
@@ -480,6 +561,7 @@ export const layerJsonRpc = (options?: {
  */
 export const layerNdJsonRpc = (options?: {
   readonly contentType?: string | undefined
+  readonly maxBufferSize?: number | "unbounded" | undefined
 }): Layer.Layer<RpcSerialization> => Layer.succeed(RpcSerialization, ndJsonRpc(options))
 
 /**
@@ -492,3 +574,11 @@ export const layerNdJsonRpc = (options?: {
  * @category serialization
  */
 export const layerMsgPack: Layer.Layer<RpcSerialization> = Layer.succeed(RpcSerialization, msgPack)
+
+/**
+ * @since 1.0.0
+ * @category serialization
+ */
+export const layerMsgPackWith = (
+  options?: Msgpackr.Options & StreamOptions | undefined
+): Layer.Layer<RpcSerialization> => Layer.succeed(RpcSerialization, makeMsgPack(options))

packages/rpc/src/RpcServer.ts

@@ -8,7 +8,7 @@ import * as HttpRouter from "@effect/platform/HttpRouter"
 import type * as HttpServerError from "@effect/platform/HttpServerError"
 import * as HttpServerRequest from "@effect/platform/HttpServerRequest"
 import * as HttpServerResponse from "@effect/platform/HttpServerResponse"
-import type * as Socket from "@effect/platform/Socket"
+import * as Socket from "@effect/platform/Socket"
 import * as SocketServer from "@effect/platform/SocketServer"
 import * as Transferable from "@effect/platform/Transferable"
 import type { WorkerError } from "@effect/platform/WorkerError"
@@ -29,6 +29,7 @@ import * as Layer from "effect/Layer"
 import * as Mailbox from "effect/Mailbox"
 import * as Option from "effect/Option"
 import { type ParseError, TreeFormatter } from "effect/ParseResult"
+import * as Predicate from "effect/Predicate"
 import * as Runtime from "effect/Runtime"
 import * as RuntimeFlags from "effect/RuntimeFlags"
 import * as Schedule from "effect/Schedule"
@@ -1484,6 +1485,9 @@ const makeSocketProtocol = Effect.gen(function*() {
           step: constVoid
         })
       } catch (cause) {
+        if (Predicate.isTagged(cause, "RpcSerializationError")) {
+          return writeRaw(new Socket.CloseEvent(1009, (cause as RpcSerialization.RpcSerializationError).message))
+        }
         return writeRaw(parser.encode(ResponseDefectEncoded(cause))!)
       }
     }).pipe(