Default TCP cluster serialization to SchemaBinary

.changeset/tcp-schema-binary-default.md

@@ -0,0 +1,12 @@
+---
+"@effect/platform-bun": patch
+"@effect/platform-deno": patch
+"@effect/platform-node": patch
+"effect": patch
+---
+
+Use SchemaBinary as the default RPC serialization for TCP cluster connections, including configurable frame limits.
+
+Cluster payloads are encoded with the binary codec on the wire. When a persisted reply cannot be encoded for JSON storage, the defect fallback that storage records is now also the reply delivered to waiting callers, so live replies always match what was persisted.
+
+SchemaBinary codecs are memoized by schema identity and wire mode, so per-message codec requests reuse the derived codec instead of rebuilding it.

packages/effect/benchmark/rpc/RpcSerialization.ts

@@ -211,7 +211,7 @@ console.log(
   "End-to-end operations include the payload codec plus RPC envelope framing; codec construction is excluded."
 )
 console.log(
-  "Msgpack uses RpcSerialization.msgPack defaults, including records. SchemaBinary fingerprints envelopes only and shares one string dictionary across the frames of a connection."
+  "Msgpack uses RpcSerialization.msgPack defaults, including records. SchemaBinary fingerprints envelopes only and keeps every frame independently decodable."
 )
 console.log(
   "First-frame sizes use a fresh serializer; steady sizes and throughput reuse one as on a long-lived connection, and decode walks a stream in frame order."

packages/effect/src/unstable/cluster/MessageStorage.ts

@@ -478,10 +478,23 @@ export type EncodedRepliesOptions<A> = {
  * @since 4.0.0
  */
 export const make = (
-  storage: Omit<
-    MessageStorage["Service"],
-    "registerReplyHandler" | "unregisterReplyHandler" | "unregisterShardReplyHandlers"
-  >
+  storage:
+    & Omit<
+      MessageStorage["Service"],
+      "saveReply" | "registerReplyHandler" | "unregisterReplyHandler" | "unregisterShardReplyHandlers"
+    >
+    & {
+      /**
+       * Save the provided `Reply`, returning the reply as it was persisted.
+       *
+       * Implementations that persist a different representation, such as a
+       * defect reply when the original cannot be encoded, return that fallback
+       * so reply handlers deliver exactly what storage recorded.
+       */
+      readonly saveReply: <R extends Rpc.Any>(
+        reply: Reply.ReplyWithContext<R>
+      ) => Effect.Effect<Reply.ReplyWithContext<R>, PersistenceError | MalformedMessage>
+    }
 ): Effect.Effect<MessageStorage["Service"]> =>
   Effect.sync(() => {
     type ReplyHandler = {
@@ -555,11 +568,11 @@ export const make = (
         }),
       saveReply(reply) {
         const requestId = reply.reply.requestId
-        return Effect.flatMap(storage.saveReply(reply), () => {
+        return Effect.flatMap(storage.saveReply(reply), (persisted) => {
           const handlers = replyHandlers.get(requestId)
           if (!handlers) {
             return Effect.void
-          } else if (reply.reply._tag === "WithExit") {
+          } else if (persisted.reply._tag === "WithExit") {
             replyHandlers.delete(requestId)
             for (let i = 0; i < handlers.length; i++) {
               const handler = handlers[i]
@@ -568,8 +581,8 @@ export const make = (
             }
           }
           return handlers.length === 1
-            ? handlers[0].respond(reply)
-            : Effect.forEach(handlers, (handler) => handler.respond(reply))
+            ? handlers[0].respond(persisted)
+            : Effect.forEach(handlers, (handler) => handler.respond(persisted))
         })
       }
     })
@@ -634,7 +647,22 @@ export const makeEncoded: (encoded: Encoded) => Effect.Effect<
         ),
         Effect.asVoid
       ),
-    saveReply: (reply) => Effect.flatMap(Reply.serializeOrDefect(reply, codecForJson), encoded.saveReply),
+    saveReply: (reply) =>
+      Reply.serialize(reply, codecForJson).pipe(
+        Effect.map((encodedReply) => ({ encodedReply, persisted: reply })),
+        Effect.catchTag("MalformedMessage", (error) => {
+          const persisted = Reply.ReplyWithContext.fromDefect({
+            id: reply.reply.id,
+            requestId: reply.reply.requestId,
+            defect: error
+          })
+          return Effect.map(
+            Effect.orDie(Reply.serialize(persisted, codecForJson)),
+            (encodedReply) => ({ encodedReply, persisted })
+          )
+        }),
+        Effect.flatMap(({ encodedReply, persisted }) => Effect.as(encoded.saveReply(encodedReply), persisted))
+      ),
     clearReplies: encoded.clearReplies,
     repliesFor: Effect.fnUntraced(function*(messages) {
       const requestIds = Arr.empty<string>()
@@ -658,7 +686,7 @@ export const makeEncoded: (encoded: Encoded) => Effect.Effect<
       return encoded.requestIdForPrimaryKey(primaryKey)
     },
     unprocessedMessages(shardIds, options) {
-      const storage = this as MessageStorage["Service"]
+      const storage = this as unknown as MessageStorage["Service"]
       const shards = Array.from(shardIds, (id) => id.toString())
       if (!Arr.isArrayNonEmpty(shards)) return Effect.succeed([])
       if (options?.addresses !== undefined && options.addresses.length === 0) return Effect.succeed([])
@@ -668,7 +696,7 @@ export const makeEncoded: (encoded: Encoded) => Effect.Effect<
       )
     },
     unprocessedMessagesById(messageIds) {
-      const storage = this as MessageStorage["Service"]
+      const storage = this as unknown as MessageStorage["Service"]
       const ids = Array.from(messageIds)
       if (!Arr.isArrayNonEmpty(ids)) return Effect.succeed([])
       return Effect.flatMap(
@@ -799,7 +827,7 @@ export const makeEncoded: (encoded: Encoded) => Effect.Effect<
 export const noop: MessageStorage["Service"] = Effect.runSync(make({
   saveRequest: () => Effect.succeed(SaveResult.Success()),
   saveEnvelope: () => Effect.void,
-  saveReply: () => Effect.void,
+  saveReply: (reply) => Effect.succeed(reply),
   clearReplies: () => Effect.void,
   repliesFor: () => Effect.succeed([]),
   repliesForUnfiltered: () => Effect.succeed([]),

packages/effect/src/unstable/encoding/SchemaBinary.ts

@@ -67,11 +67,17 @@ export interface toCodec<S extends Schema.Constraint> extends
   >
 {}
 
+const codecCache = new WeakMap<Schema.Constraint, toCodec<any>>()
+const fingerprintCodecCache = new WeakMap<Schema.Constraint, toCodec<any>>()
+const directCodecCache = new WeakMap<Schema.Constraint, toCodec<any>>()
+const directFingerprintCodecCache = new WeakMap<Schema.Constraint, toCodec<any>>()
+
 /**
  * Derives a compact binary codec from a schema.
  *
  * The wire layout is compiled from the encoded side of the schema. Each
  * encode/decode handles exactly one frame; use {@link parser} for streams.
+ * Derived codecs are memoized by schema identity and wire mode.
  *
  * Encoded results are arena-backed views and may share a larger buffer. Use
  * `bytes.slice()` when independent ownership is required.
@@ -93,15 +99,20 @@ export interface toCodec<S extends Schema.Constraint> extends
  * @since 4.0.0
  */
 export function toCodec<S extends Schema.Constraint>(schema: S, options?: Options): toCodec<S> {
+  const cache = options?.fingerprint === true ? fingerprintCodecCache : codecCache
+  const cached = cache.get(schema)
+  if (cached !== undefined) return cached
   const { exact, layout, recursive, target } = compileTarget(schema)
   const mode = compileMode(layout, options?.fingerprint)
   const trusted: Trusted | undefined = exact ? { value: undefined } : undefined
-  return assembleCodec(
+  const codec = assembleCodec<S>(
     trusted === undefined ? target : withTrustedDecode(target, trusted, !recursive),
     layout,
     mode,
     trusted
   )
+  cache.set(schema, codec)
+  return codec
 }
 
 /**
@@ -111,24 +122,37 @@ export function toCodec<S extends Schema.Constraint>(schema: S, options?: Option
  * Only schemas the binary layer already validates on its own take the direct
  * path; anything else falls back to {@link toCodec}. A direct codec encodes and
  * decodes the same values as {@link toCodec} but is not a sound `Schema.is`
- * guard, so it is only for callers that never guard on it.
+ * guard, so it is only for callers that never guard on it. Derived codecs are
+ * memoized by schema identity and wire mode.
  *
  * @internal
  */
 export function toCodecDirect<S extends Schema.Constraint>(schema: S, options?: Options): toCodec<S> {
+  const cache = options?.fingerprint === true ? directFingerprintCodecCache : directCodecCache
+  const cached = cache.get(schema)
+  if (cached !== undefined) return cached
   const { exact, exitSuccess, layout, target } = compileTarget(schema)
   if (!exact) {
     if (!exitSuccess) return toCodec(schema, options)
     const trusted: Trusted = { value: undefined }
-    return assembleCodec(
+    const codec = assembleCodec<S>(
       withExitSuccessDecode(target, trusted),
       layout,
       compileMode(layout, options?.fingerprint),
       trusted,
       true
     )
+    cache.set(schema, codec)
+    return codec
   }
-  return assembleCodec(passThrough(target), layout, compileMode(layout, options?.fingerprint), undefined)
+  const codec = assembleCodec<S>(
+    passThrough(target),
+    layout,
+    compileMode(layout, options?.fingerprint),
+    undefined
+  )
+  cache.set(schema, codec)
+  return codec
 }
 
 function assembleCodec<S extends Schema.Constraint>(

packages/effect/src/unstable/rpc/RpcSerialization.ts

@@ -596,19 +596,16 @@ const defaultSchemaBinaryMaxFrameSize = 16 * 1024 * 1024
 const schemaBinaryTextEncoder = new TextEncoder()
 
 const makeSchemaBinary = (options?: {
-  readonly maxFrameSize?: number | undefined
+  readonly maxFrameSize?: number | "unbounded" | undefined
   readonly fingerprintPayloads?: boolean | undefined
 }): RpcSerialization["Service"] => {
-  const maxFrameSize = options?.maxFrameSize ?? defaultSchemaBinaryMaxFrameSize
+  const maxFrameSize = options?.maxFrameSize === "unbounded"
+    ? undefined
+    : options?.maxFrameSize ?? defaultSchemaBinaryMaxFrameSize
   const codecFor: CodecFor = options?.fingerprintPayloads === true
     ? (schema) => SchemaBinary.toCodecDirect(schema, { fingerprint: true })
     : SchemaBinary.toCodecDirect
-  // The envelope repeats itself: the same RPC tag, header names, and trace ids
-  // come back on message after message. A dictionary shared by every frame on
-  // the connection sends each of those once and references it afterwards, so
-  // the writer and the reader here are a matched pair and neither one works
-  // against a peer that was built without the other.
-  const envelopeOptions = { fingerprint: true, dictionary: true } as const
+  const envelopeOptions = { fingerprint: true } as const
   return RpcSerialization.of({
     contentType: "application/vnd.effect.rpc+schema-binary",
     includesFraming: true,
@@ -715,12 +712,13 @@ export const layerMsgPackWith = (
 /**
  * RPC serialization layer that uses SchemaBinary with fingerprinted RPC
  * envelopes. Payload fingerprints are disabled by default to support compatible
- * schema evolution. Frames default to a 16 MiB maximum size.
+ * schema evolution. Frames default to a 16 MiB maximum size. Use `"unbounded"`
+ * to disable the frame-size limit.
  *
  * @category layers
  * @since 4.0.0
  */
 export const layerSchemaBinary = (options?: {
-  readonly maxFrameSize?: number | undefined
+  readonly maxFrameSize?: number | "unbounded" | undefined
   readonly fingerprintPayloads?: boolean | undefined
 }): Layer.Layer<RpcSerialization> => Layer.sync(RpcSerialization)(() => makeSchemaBinary(options))

packages/effect/test/cluster/MessageStorage.test.ts

@@ -207,13 +207,57 @@ describe("MessageStorage", () => {
         yield* latch.await
         yield* Fiber.await(fiber)
       }).pipe(Effect.provide(MemoryLayer)))
+
+    it.effect("reply handlers receive the persisted defect fallback for unencodable replies", () =>
+      Effect.gen(function*() {
+        const storage = yield* MessageStorage.MessageStorage
+        const snowflake = yield* Snowflake.Generator
+        const latch = yield* Latch.make()
+        const request = yield* makeRequest({ rpc: UnknownSuccessRpc })
+        yield* storage.saveRequest(request)
+        let received: Reply.Reply<any> | undefined
+        const fiber = yield* storage.registerReplyHandler(
+          new Message.OutgoingRequest({
+            ...request,
+            respond: (reply) => {
+              received = reply
+              return latch.open
+            }
+          })
+        ).pipe(Effect.forkChild)
+        yield* TestClock.adjust(1)
+        yield* storage.saveReply(
+          new Reply.ReplyWithContext({
+            reply: new Reply.WithExit({
+              id: snowflake.nextUnsafe(),
+              requestId: request.envelope.requestId,
+              exit: Exit.succeed(new Error("not json encodable")) as any
+            }),
+            context: request.context,
+            rpc: request.rpc
+          })
+        )
+        yield* latch.await
+        yield* Fiber.await(fiber)
+        expect(received?._tag).toEqual("WithExit")
+        const exit = (received as Reply.WithExit<any>).exit
+        expect(Exit.isFailure(exit)).toBe(true)
+        expect(JSON.stringify(exit)).toContain("MalformedMessage")
+        const stored = yield* storage.repliesForUnfiltered([request.envelope.requestId])
+        expect(JSON.stringify(stored)).toContain("MalformedMessage")
+      }).pipe(Effect.provide(MemoryLayer)))
   })
 })
 
 export const GetUserRpc = Rpc.make("GetUser", {
   payload: { id: Schema.Number }
 })
 
+const UnknownSuccessRpc = Rpc.make("UnknownSuccess", {
+  payload: { id: Schema.Number },
+  success: Schema.Unknown
+})
+
 export const makeRequest = Effect.fnUntraced(function*(options?: {
   readonly rpc?: Rpc.AnyWithProps
   readonly payload?: any

packages/effect/test/rpc/RpcSerialization.test.ts

@@ -544,6 +544,31 @@ describe("RpcSerialization", () => {
         assert.deepStrictEqual(parser.decode(frame), [request])
       }).pipe(Effect.provide(RpcSerialization.layerSchemaBinary())))
 
+    it.effect("keeps varied frames decodable across parser replacement", () =>
+      Effect.gen(function*() {
+        const serialization = yield* RpcSerialization.RpcSerialization
+        const encoder = serialization.makeUnsafe()
+        const requests: Array<RpcMessage.RequestEncoded> = Array.from({ length: 200 }, (_, index) => ({
+          _tag: "Request",
+          id: index % 2 === 0 ? index : `request-${index}`,
+          tag: `Echo${index % 7}`,
+          payload: Uint8Array.from({ length: index % 17 }, (_, offset) => (index + offset) & 0xFF),
+          headers: Array.from({ length: index % 4 }, (_, offset) => [`x-${offset}`, `${index}`]),
+          ...(index % 3 === 0
+            ? { traceId: `trace-${index}`, spanId: `span-${index}`, sampled: index % 2 === 0 }
+            : undefined)
+        }))
+
+        for (const request of requests) {
+          const frame = encoder.encode(request)
+          assert.instanceOf(frame, Uint8Array)
+          const split = 1 + request.tag.length % (frame.length - 1)
+          const parser = serialization.makeUnsafe()
+          assert.deepStrictEqual(parser.decode(frame.subarray(0, split)), [])
+          assert.deepStrictEqual(parser.decode(frame.subarray(split)), [request])
+        }
+      }).pipe(Effect.provide(RpcSerialization.layerSchemaBinary())))
+
     it.effect("owns encoded frames without copying envelope holes", () =>
       Effect.gen(function*() {
         const serialization = yield* RpcSerialization.RpcSerialization
@@ -586,6 +611,12 @@ describe("RpcSerialization", () => {
         assert.deepStrictEqual(serialization.makeUnsafe().decode(uvarint(4)), [])
         assert.throws(() => serialization.makeUnsafe().decode(uvarint(5)), /frame within maxFrameSize/)
       }).pipe(Effect.provide(RpcSerialization.layerSchemaBinary({ maxFrameSize: 4 }))))
+
+    it.effect("layerSchemaBinary allows an unbounded maxFrameSize", () =>
+      Effect.gen(function*() {
+        const serialization = yield* RpcSerialization.RpcSerialization
+        assert.deepStrictEqual(serialization.makeUnsafe().decode(uvarint(16 * 1024 * 1024 + 1)), [])
+      }).pipe(Effect.provide(RpcSerialization.layerSchemaBinary({ maxFrameSize: "unbounded" }))))
   })
 
   describe("codecFor", () => {