1import { Effect, Exit, Schema } from "effect"
2import { Rpc, RpcSerialization } from "effect/rpc"
3import assert from "node:assert/strict"
4import { cpus } from "node:os"
5import { Bench } from "tinybench"
6
7const User = Schema.Struct({
8 id: Schema.String,
9 displayName: Schema.String,
10 email: Schema.String,
11 active: Schema.Boolean,
12 roles: Schema.Array(Schema.String)
13})
14
15const SearchUsers = Rpc.make("SearchUsers", {
16 payload: Schema.Struct({
17 organizationId: Schema.String,
18 query: Schema.String,
19 page: Schema.Number,
20 pageSize: Schema.Number,
21 filters: Schema.Struct({
22 active: Schema.optional(Schema.Boolean),
23 roles: Schema.Array(Schema.String)
24 })
25 }),
26 success: Schema.Array(User),
27 error: Schema.Struct({
28 code: Schema.String,
29 message: Schema.String
30 })
31})
32
33const Event = Schema.Struct({
34 sequence: Schema.Number,
35 timestamp: Schema.Date,
36 level: Schema.Literals(["info", "warning", "error"]),
37 message: Schema.String,
38 attributes: Schema.Record(Schema.String, Schema.String)
39})
40
41const searchPayload = {
42 organizationId: "org-effect",
43 query: "schema binary",
44 page: 3,
45 pageSize: 25,
46 filters: {
47 active: true,
48 roles: ["maintainer", "contributor"]
49 }
50}
51
52const users = Array.from({ length: 12 }, (_, index) => ({
53 id: `user-${index.toString().padStart(3, "0")}`,
54 displayName: `Benchmark User ${index}`,
55 email: `benchmark-${index}@example.com`,
56 active: index % 4 !== 0,
57 roles: index % 3 === 0 ? ["maintainer", "contributor"] : ["contributor"]
58}))
59
60const makeEvent = (index: number) => ({
61 sequence: index + 1,
62 timestamp: new Date(1_756_000_000_000 + index * 1_000),
63 level: index % 11 === 0 ? "warning" as const : "info" as const,
64 message: `Processed RPC event ${index + 1}`,
65 attributes: {
66 region: ["eu-west-1", "us-east-1", "ap-southeast-2"][index % 3]!,
67 worker: `worker-${index % 8}`,
68 attempt: String(index % 3 + 1)
69 }
70})
71
72const events = [
73 makeEvent(0),
74 ...Array.from({ length: 31 }, (_, index) => makeEvent(index + 1))
75] satisfies Schema.NonEmptyArray<typeof Event>["Type"]
76
77interface BenchmarkCase {
78 readonly name: string
79 readonly schema: Schema.Codec<unknown, unknown>
80 readonly value: unknown
81 readonly envelope: (hole: unknown) => unknown
82 readonly hole: (envelope: unknown) => unknown
83}
84
85const getHole = (key: PropertyKey) => (envelope: unknown): unknown => {
86 assert(typeof envelope === "object" && envelope !== null && key in envelope)
87 return (envelope as Record<PropertyKey, unknown>)[key]
88}
89
90const cases: ReadonlyArray<BenchmarkCase> = [
91 {
92 name: "request / nested search payload",
93 schema: SearchUsers.payloadSchema,
94 value: searchPayload,
95 envelope: (payload) => ({
96 _tag: "Request",
97 id: 1,
98 tag: SearchUsers._tag,
99 payload,
100 headers: [
101 ["authorization", "Bearer benchmark-token"],
102 ["x-request-id", "benchmark-request-0001"]
103 ],
104 traceId: "0123456789abcdef0123456789abcdef",
105 spanId: "0123456789abcdef",
106 sampled: true
107 }),
108 hole: getHole("payload")
109 },
110 {
111 name: "exit / 12-user success",
112 schema: Rpc.exitSchema(SearchUsers),
113 value: Exit.succeed(users),
114 envelope: (exit) => ({
115 _tag: "Exit",
116 requestId: 1,
117 exit
118 }),
119 hole: getHole("exit")
120 },
121 {
122 name: "chunk / 32 events",
123 schema: Schema.NonEmptyArray(Event),
124 value: events,
125 envelope: (values) => ({
126 _tag: "Chunk",
127 requestId: 1,
128 values
129 }),
130 hole: getHole("values")
131 }
132]
133
134const schemaBinary = Effect.runSync(
135 RpcSerialization.RpcSerialization.pipe(Effect.provide(RpcSerialization.layerSchemaBinary()))
136)
137
138const formats = [
139 { name: "NDJSON", serialization: RpcSerialization.ndjson },
140 { name: "SchemaBinary", serialization: schemaBinary }
141] as const
142
143const warmupIterations = 500
144const iterations = 5_000
145// A serialization may carry state from one frame to the next, so the decode
146// task walks a stream in order and never feeds the same frame twice.
147const streamLength = warmupIterations + iterations + 1
148// Frames a connection sends before its serialization has settled.
149const warmupFrames = 8
150
151interface Prepared {
152 readonly caseName: string
153 readonly formatName: string
154 readonly firstFrameSize: number
155 readonly steadyFrameSize: number
156 readonly encode: () => unknown
157 readonly decode: () => unknown
158}
159
160const toBytes = (encoded: Uint8Array | string | undefined): Uint8Array<ArrayBuffer> => {
161 assert(encoded instanceof Uint8Array)
162 return encoded.slice()
163}
164
165const prepared = cases.flatMap((testCase): ReadonlyArray<Prepared> =>
166 formats.map(({ name, serialization }) => {
167 const encodeHole = Schema.encodeUnknownSync(serialization.codecFor(testCase.schema))
168 const decodeHole = Schema.decodeUnknownSync(serialization.codecFor(testCase.schema))
169 // A binary serialization fills the hole with bytes; a JSON-shaped one fills
170 // it with the value itself.
171 const encodedHole = encodeHole(testCase.value)
172 const hole = encodedHole instanceof Uint8Array ? encodedHole.slice() : encodedHole
173 const encoder = serialization.makeUnsafe()
174 const encode = () => encoder.encode(testCase.envelope(encodeHole(testCase.value)))
175 const firstFrame = toBytes(encode())
176 let steadyFrame = firstFrame
177 for (let i = 0; i < warmupFrames; i++) steadyFrame = toBytes(encode())
178
179 const streamWriter = serialization.makeUnsafe()
180 const stream = Array.from(
181 { length: streamLength },
182 () => toBytes(streamWriter.encode(testCase.envelope(hole)) as Uint8Array)
183 )
184 let decoder = serialization.makeUnsafe()
185 let index = 0
186 const decode = () => {
187 if (index === stream.length) {
188 decoder = serialization.makeUnsafe()
189 index = 0
190 }
191 const envelopes = decoder.decode(stream[index++])
192 assert.strictEqual(envelopes.length, 1)
193 return decodeHole(testCase.hole(envelopes[0]))
194 }
195
196 assert.deepStrictEqual(decode(), testCase.value)
197
198 return {
199 caseName: testCase.name,
200 formatName: name,
201 firstFrameSize: firstFrame.length,
202 steadyFrameSize: steadyFrame.length,
203 encode,
204 decode
205 }
206 })
207)
208
209console.log(`${process.platform} ${process.arch}; ${cpus()[0]?.model ?? "unknown CPU"}; Node ${process.version}`)
210console.log(
211 "End-to-end operations include the payload codec plus RPC envelope framing; codec construction is excluded."
212)
213console.log("SchemaBinary fingerprints envelopes only and keeps every frame independently decodable.")
214console.log(
215 "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."
216)
217console.log(
218 `${warmupIterations.toLocaleString()} warmup operations and ${iterations.toLocaleString()} measured operations per case, format, and direction.`
219)
220
221console.table(prepared.map((entry) => ({
222 Case: entry.caseName,
223 Format: entry.formatName,
224 "First frame bytes": entry.firstFrameSize,
225 "Steady frame bytes": entry.steadyFrameSize
226})))
227
228const bench = new Bench({
229 iterations,
230 time: 0,
231 warmupIterations,
232 warmupTime: 0,
233 timestampProvider: "hrtimeNow"
234})
235const labels = new Map<string, {
236 readonly caseName: string
237 readonly formatName: string
238 readonly direction: string
239}>()
240let sink: unknown
241
242for (const entry of prepared) {
243 for (const [direction, run] of [["encode", entry.encode], ["decode", entry.decode]] as const) {
244 const name = `${entry.caseName} / ${entry.formatName} / ${direction}`
245 labels.set(name, {
246 caseName: entry.caseName,
247 formatName: entry.formatName,
248 direction
249 })
250 bench.add(name, () => {
251 sink = run()
252 })
253 }
254}
255
256await bench.run()
257assert.notStrictEqual(sink, undefined)
258
259const ndjsonThroughput = new Map<string, number>()
260for (const task of bench.tasks) {
261 const label = labels.get(task.name)!
262 if (label.formatName === "NDJSON" && task.result?.state === "completed") {
263 ndjsonThroughput.set(`${label.caseName}/${label.direction}`, task.result.throughput.mean)
264 }
265}
266
267console.table(bench.tasks.map((task) => {
268 const label = labels.get(task.name)!
269 const result = task.result
270 if (result?.state === "errored") {
271 return {
272 Case: label.caseName,
273 Format: label.formatName,
274 Direction: label.direction,
275 Error: result.error.message
276 }
277 }
278 if (result?.state !== "completed") {
279 return {
280 Case: label.caseName,
281 Format: label.formatName,
282 Direction: label.direction,
283 State: result?.state ?? "missing result"
284 }
285 }
286 const baseline = ndjsonThroughput.get(`${label.caseName}/${label.direction}`)!
287 return {
288 Case: label.caseName,
289 Format: label.formatName,
290 Direction: label.direction,
291 "Throughput avg (ops/s)": Math.round(result.throughput.mean),
292 "vs NDJSON": `${(result.throughput.mean / baseline).toFixed(2)}x`,
293 "Latency med (us/op)": (result.latency.p50 * 1_000).toFixed(2),
294 "Latency RME": `${result.latency.rme.toFixed(2)}%`,
295 Samples: result.latency.samplesCount
296 }
297}))