1import { Effect, Schema, Stream } from "effect"
2import { Ndjson, SchemaBinary } from "effect/encoding"
3import assert from "node:assert/strict"
4import { gzipSync, zstdCompressSync } from "node:zlib"
5import protobuf, { type Message, type Type } from "protobufjs"
6import { Bench } from "tinybench"
7
8const streamBatchSize = 32
9const repeatedRecordStreamSize = 200
10
11const SmallRecord = Schema.Struct({
12 id: Schema.Number,
13 active: Schema.Boolean,
14 score: Schema.Number,
15 retryCount: Schema.Number,
16 region: Schema.String,
17 verified: Schema.Boolean
18})
19
20const LineItem = Schema.Struct({
21 sku: Schema.String,
22 quantity: Schema.Number,
23 unitPrice: Schema.Number
24})
25
26const NestedPayload = Schema.Struct({
27 orderId: Schema.String,
28 customer: Schema.Struct({
29 id: Schema.String,
30 name: Schema.String,
31 email: Schema.String
32 }),
33 shipping: Schema.Struct({
34 street: Schema.String,
35 city: Schema.String,
36 postalCode: Schema.String,
37 country: Schema.String
38 }),
39 lines: Schema.Array(LineItem),
40 metadata: Schema.Struct({
41 source: Schema.String,
42 campaign: Schema.String,
43 priority: Schema.Boolean
44 })
45})
46
47const Collections = Schema.Struct({
48 tags: Schema.Array(Schema.String),
49 metrics: Schema.Record(Schema.String, Schema.Number),
50 samples: Schema.Array(Schema.Tuple([Schema.Number, Schema.Number, Schema.Boolean])),
51 buckets: Schema.Array(Schema.Array(Schema.Number))
52})
53
54const LargeRow = Schema.Struct({
55 transactionIdentifier: Schema.String,
56 customerIdentifier: Schema.String,
57 productDescription: Schema.String,
58 fulfillmentLocation: Schema.String,
59 quantityPurchased: Schema.Number,
60 unitPriceInCents: Schema.Number,
61 discountInBasisPoints: Schema.Number,
62 requiresManualReview: Schema.Boolean
63})
64
65const LargePayload = Schema.Array(LargeRow)
66
67const largeRows = Array.from({ length: repeatedRecordStreamSize }, (_, index) => ({
68 transactionIdentifier: `transaction-${index.toString().padStart(4, "0")}`,
69 customerIdentifier: `customer-${index % 37}`,
70 productDescription: `Product ${index % 19} with a repeated descriptive field value`,
71 fulfillmentLocation: ["London", "New York", "Singapore", "Sydney"][index % 4]!,
72 quantityPurchased: index % 9 + 1,
73 unitPriceInCents: 500 + index % 73 * 25,
74 discountInBasisPoints: index % 5 * 125,
75 requiresManualReview: index % 17 === 0
76}))
77
78const metrics = (count: number) =>
79 Object.fromEntries(Array.from({ length: count }, (_, index) => [`metric-${index}`, index * 1.25]))
80
81const protobufRoot = protobuf.parse(`
82 syntax = "proto3";
83
84 message SmallRecord {
85 double id = 1;
86 bool active = 2;
87 double score = 3;
88 double retryCount = 4;
89 string region = 5;
90 bool verified = 6;
91 }
92
93 message LineItem {
94 string sku = 1;
95 double quantity = 2;
96 double unitPrice = 3;
97 }
98
99 message Customer {
100 string id = 1;
101 string name = 2;
102 string email = 3;
103 }
104
105 message Shipping {
106 string street = 1;
107 string city = 2;
108 string postalCode = 3;
109 string country = 4;
110 }
111
112 message Metadata {
113 string source = 1;
114 string campaign = 2;
115 bool priority = 3;
116 }
117
118 message NestedPayload {
119 string orderId = 1;
120 Customer customer = 2;
121 Shipping shipping = 3;
122 repeated LineItem lines = 4;
123 Metadata metadata = 5;
124 }
125
126 message Sample {
127 double first = 1;
128 double second = 2;
129 bool third = 3;
130 }
131
132 message Bucket {
133 repeated double values = 1;
134 }
135
136 message Collections {
137 repeated string tags = 1;
138 map<string, double> metrics = 2;
139 repeated Sample samples = 3;
140 repeated Bucket buckets = 4;
141 }
142
143 message NumberMap {
144 map<string, double> values = 1;
145 }
146
147 message LargeRow {
148 string transactionIdentifier = 1;
149 string customerIdentifier = 2;
150 string productDescription = 3;
151 string fulfillmentLocation = 4;
152 double quantityPurchased = 5;
153 double unitPriceInCents = 6;
154 double discountInBasisPoints = 7;
155 bool requiresManualReview = 8;
156 }
157
158 message LargePayload {
159 repeated LargeRow values = 1;
160 }
161`).root
162
163interface ProtobufFixture {
164 readonly type: Type
165 readonly encodeInput: (value: unknown) => object
166 readonly decodeOutput: (message: Message) => unknown
167}
168
169const protobufObject = <T extends object>(type: Type, message: Message): T =>
170 type.toObject(message, { arrays: true, defaults: true, objects: true }) as T
171
172const directProtobufFixture = (name: string): ProtobufFixture => {
173 const type = protobufRoot.lookupType(name)
174 return {
175 type,
176 encodeInput: (value) => value as object,
177 decodeOutput: (message) => protobufObject(type, message)
178 }
179}
180
181const collectionsProtobufType = protobufRoot.lookupType("Collections")
182const numberMapProtobufType = protobufRoot.lookupType("NumberMap")
183const largePayloadProtobufType = protobufRoot.lookupType("LargePayload")
184
185const collectionsProtobufFixture: ProtobufFixture = {
186 type: collectionsProtobufType,
187 encodeInput: (value) => {
188 const collections = value as (typeof Collections)["Type"]
189 return {
190 tags: collections.tags,
191 metrics: collections.metrics,
192 samples: collections.samples.map(([first, second, third]) => ({ first, second, third })),
193 buckets: collections.buckets.map((values) => ({ values }))
194 }
195 },
196 decodeOutput: (message) => {
197 const collections = protobufObject<{
198 readonly tags: ReadonlyArray<string>
199 readonly metrics: Readonly<Record<string, number>>
200 readonly samples: ReadonlyArray<{ readonly first: number; readonly second: number; readonly third: boolean }>
201 readonly buckets: ReadonlyArray<{ readonly values: ReadonlyArray<number> }>
202 }>(collectionsProtobufType, message)
203 return {
204 tags: collections.tags,
205 metrics: collections.metrics,
206 samples: collections.samples.map((sample) => [sample.first, sample.second, sample.third]),
207 buckets: collections.buckets.map((bucket) => bucket.values)
208 }
209 }
210}
211
212const numberMapProtobufFixture: ProtobufFixture = {
213 type: numberMapProtobufType,
214 encodeInput: (value) => ({ values: value }),
215 decodeOutput: (message) =>
216 protobufObject<{ readonly values: Readonly<Record<string, number>> }>(numberMapProtobufType, message).values
217}
218
219const largePayloadProtobufFixture: ProtobufFixture = {
220 type: largePayloadProtobufType,
221 encodeInput: (value) => ({ values: value }),
222 decodeOutput: (message) =>
223 protobufObject<{ readonly values: ReadonlyArray<(typeof LargeRow)["Type"]> }>(largePayloadProtobufType, message)
224 .values
225}
226
227const largeRowProtobufFixture = directProtobufFixture("LargeRow")
228
229const cases = [
230 {
231 name: "small record",
232 schema: SmallRecord,
233 value: {
234 id: 42,
235 active: true,
236 score: 98.5,
237 retryCount: 2,
238 region: "eu-west-1",
239 verified: false
240 },
241 protobuf: directProtobufFixture("SmallRecord")
242 },
243 {
244 name: "nested payload",
245 schema: NestedPayload,
246 value: {
247 orderId: "order-2026-000184",
248 customer: {
249 id: "customer-91",
250 name: "Ada Lovelace",
251 email: "ada@example.com"
252 },
253 shipping: {
254 street: "12 Analytical Engine Way",
255 city: "London",
256 postalCode: "SW1A 1AA",
257 country: "GB"
258 },
259 lines: [
260 { sku: "widget-blue", quantity: 2, unitPrice: 12.5 },
261 { sku: "adapter-pro", quantity: 1, unitPrice: 48 },
262 { sku: "cable-2m", quantity: 3, unitPrice: 8.25 }
263 ],
264 metadata: {
265 source: "partner-api",
266 campaign: "summer-2026",
267 priority: true
268 }
269 },
270 protobuf: directProtobufFixture("NestedPayload")
271 },
272 {
273 name: "collections",
274 schema: Collections,
275 value: {
276 tags: Array.from({ length: 24 }, (_, index) => `tag-${index}`),
277 metrics: Object.fromEntries(Array.from({ length: 24 }, (_, index) => [`metric-${index}`, index * 1.25])),
278 samples: Array.from({ length: 48 }, (_, index) => [index, index / 10, index % 3 === 0] as const),
279 buckets: Array.from({ length: 8 }, (_, bucket) => Array.from({ length: 16 }, (_, index) => bucket * 100 + index))
280 },
281 protobuf: collectionsProtobufFixture
282 },
283 {
284 name: "index signatures / 128 keys",
285 schema: Schema.Record(Schema.String, Schema.Number),
286 value: metrics(128),
287 protobuf: numberMapProtobufFixture
288 },
289 {
290 name: "index signatures / 512 keys",
291 schema: Schema.Record(Schema.String, Schema.Number),
292 value: metrics(512),
293 protobuf: numberMapProtobufFixture
294 },
295 {
296 name: "200-row array payload",
297 schema: LargePayload,
298 value: largeRows,
299 protobuf: largePayloadProtobufFixture
300 }
301] as const
302
303interface Format {
304 readonly name: string
305 readonly encodedSize: number
306 readonly gzipSize: number
307 readonly zstdSize: number
308 readonly encode: () => unknown
309 readonly decode: () => unknown
310}
311
312interface StreamFormat {
313 readonly name: string
314 readonly framesPerOp: number
315 readonly decode: () => ReadonlyArray<unknown> | Promise<ReadonlyArray<unknown>>
316}
317
318interface StreamSize {
319 readonly name: string
320 readonly frames: number
321 readonly encodedSize: number
322 readonly gzipSize: number
323 readonly zstdSize: number
324}
325
326const textEncoder = new TextEncoder()
327
328const concatFrames = (frames: ReadonlyArray<Uint8Array>): Uint8Array<ArrayBuffer> => {
329 const out = new Uint8Array(frames.reduce((length, frame) => length + frame.length, 0))
330 let offset = 0
331 for (const frame of frames) {
332 out.set(frame, offset)
333 offset += frame.length
334 }
335 return out
336}
337
338const sizes = (encoded: Uint8Array): Pick<Format, "encodedSize" | "gzipSize" | "zstdSize"> => ({
339 encodedSize: encoded.length,
340 gzipSize: gzipSync(encoded).length,
341 zstdSize: zstdCompressSync(encoded).length
342})
343
344const prepare = <S extends Schema.ConstraintCodec<unknown, unknown>>(
345 schema: S,
346 value: S["Type"],
347 protobufFixture: ProtobufFixture
348): {
349 readonly formats: ReadonlyArray<Format>
350} => {
351 const jsonSchema = Schema.toCodecJson(schema)
352 const binaryCodec = SchemaBinary.toCodec(schema)
353 const fingerprintCodec = SchemaBinary.toCodec(schema, { fingerprint: true })
354 const jsonCodec = Schema.fromJsonString(jsonSchema)
355
356 const binaryEncode = Schema.encodeUnknownSync(binaryCodec)
357 const binaryDecode = Schema.decodeUnknownSync(binaryCodec)
358 const fingerprintEncode = Schema.encodeUnknownSync(fingerprintCodec)
359 const fingerprintDecode = Schema.decodeUnknownSync(fingerprintCodec)
360 const jsonEncode = Schema.encodeUnknownSync(jsonCodec)
361 const jsonDecode = Schema.decodeUnknownSync(jsonCodec)
362 const protobufValue = protobufFixture.encodeInput(value)
363
364 const binary = binaryEncode(value)
365 const fingerprint = fingerprintEncode(value).slice()
366 const json = jsonEncode(value)
367 const jsonBytes = textEncoder.encode(json)
368 const protobufBytes = protobufFixture.type.encode(protobufValue).finish()
369 const protobufDecode = () => protobufFixture.decodeOutput(protobufFixture.type.decode(protobufBytes))
370
371 assert.deepStrictEqual(binaryDecode(binary), value)
372 assert.deepStrictEqual(fingerprintDecode(fingerprint), value)
373 assert.deepStrictEqual(jsonDecode(json), value)
374 assert.deepStrictEqual(protobufDecode(), value)
375
376 return {
377 formats: [
378 {
379 name: "SchemaBinary",
380 ...sizes(binary),
381 encode: () => binaryEncode(value),
382 decode: () => binaryDecode(binary)
383 },
384 {
385 name: "SchemaBinary fingerprint",
386 ...sizes(fingerprint),
387 encode: () => fingerprintEncode(value),
388 decode: () => fingerprintDecode(fingerprint)
389 },
390 {
391 name: "JSON",
392 ...sizes(jsonBytes),
393 encode: () => jsonEncode(value),
394 decode: () => jsonDecode(json)
395 },
396 {
397 name: "Protobuf",
398 ...sizes(protobufBytes),
399 encode: () => protobufFixture.type.encode(protobufValue).finish(),
400 decode: protobufDecode
401 }
402 ]
403 }
404}
405
406const prepared = cases.map((testCase) => ({
407 name: testCase.name,
408 ...prepare(testCase.schema, testCase.value, testCase.protobuf)
409}))
410
411const prepareStream = async <S extends Schema.ConstraintCodec<unknown, unknown>>(
412 schema: S,
413 values: ReadonlyArray<S["Type"]>,
414 protobufFixture: ProtobufFixture
415): Promise<{ readonly formats: ReadonlyArray<StreamFormat>; readonly sizes: ReadonlyArray<StreamSize> }> => {
416 const binaryCodec = SchemaBinary.toCodec(schema)
417 const binaryEncode = Schema.encodeUnknownSync(binaryCodec)
418 const binaryFrames = values.map((value) => binaryEncode(value).slice())
419 const binaryStream = concatFrames(binaryFrames)
420 const binaryFragments = binaryFrames.map((frame) => {
421 return [frame.subarray(0, 1), frame.subarray(1)] as const
422 })
423
424 const fingerprintEncode = Schema.encodeUnknownSync(SchemaBinary.toCodec(schema, { fingerprint: true }))
425 const fingerprintFrames = values.map((value) => fingerprintEncode(value).slice())
426 const fingerprintStream = concatFrames(fingerprintFrames)
427 const fingerprintFragments = fingerprintFrames.map((frame) => {
428 return [frame.subarray(0, 1), frame.subarray(1)] as const
429 })
430
431 const jsonSchema = Schema.toCodecJson(schema)
432 const encodeJsonValue = Schema.encodeUnknownSync(jsonSchema)
433
434 const protobufFrames = values.map((value) =>
435 protobufFixture.type.encodeDelimited(protobufFixture.encodeInput(value)).finish()
436 )
437 const protobufStream = concatFrames(protobufFrames)
438 const decodeProtobufStream = () => {
439 const reader = protobuf.Reader.create(protobufStream)
440 const decoded: Array<unknown> = []
441 while (reader.pos < reader.len) {
442 decoded.push(protobufFixture.decodeOutput(protobufFixture.type.decodeDelimited(reader)))
443 }
444 return decoded
445 }
446
447 const ndjsonFrames = values.map((value) => textEncoder.encode(`${JSON.stringify(encodeJsonValue(value))}\n`))
448 const ndjsonStream = concatFrames(ndjsonFrames)
449 const ndjsonFragments = ndjsonFrames.map((frame) => {
450 return [frame.subarray(0, 1), frame.subarray(1)] as const
451 })
452 const ndjsonDecoder = Ndjson.decodeSchema(jsonSchema)()
453 const runNdjson = (chunks: ReadonlyArray<Uint8Array>) =>
454 Effect.runPromise(
455 Stream.fromIterable(chunks).pipe(
456 Stream.pipeThroughChannel(ndjsonDecoder),
457 Stream.runCollect,
458 Effect.map((chunk) => Array.from(chunk))
459 )
460 )
461 let ndjsonSingleIndex = 0
462 let ndjsonFragmentedIndex = 0
463 const decodeNdjsonSingle = () => runNdjson([ndjsonFrames[ndjsonSingleIndex++ % ndjsonFrames.length]!])
464 const decodeNdjsonBatch = () => runNdjson([ndjsonStream])
465 const decodeNdjsonFragmented = () => runNdjson(ndjsonFragments[ndjsonFragmentedIndex++ % ndjsonFragments.length]!)
466
467 const feedShapes = (options?: { readonly fingerprint: true }) => {
468 const frames = options === undefined ? binaryFrames : fingerprintFrames
469 const fragments = options === undefined ? binaryFragments : fingerprintFragments
470 const stream = options === undefined ? binaryStream : fingerprintStream
471 const singleParser = SchemaBinary.parser(schema, options)
472 const fragmentedParser = SchemaBinary.parser(schema, options)
473 const batchParser = SchemaBinary.parser(schema, options)
474 let singleIndex = 0
475 let fragmentedIndex = 0
476 return {
477 single: () => singleParser.feedSync(frames[singleIndex++ % frames.length]),
478 fragmented: () => {
479 const pair = fragments[fragmentedIndex++ % fragments.length]
480 const first = fragmentedParser.feedSync(pair[0])
481 const second = fragmentedParser.feedSync(pair[1])
482 return first.length === 0 ? second : [...first, ...second]
483 },
484 batch: () => batchParser.feedSync(stream)
485 }
486 }
487
488 const defaultFeeds = feedShapes()
489 const fingerprintFeeds = feedShapes({ fingerprint: true })
490 const decodeSingle = defaultFeeds.single
491 const decodeFragmented = defaultFeeds.fragmented
492 const decodeBatch = defaultFeeds.batch
493 assert.deepStrictEqual(decodeSingle(), [values[0]])
494 assert.deepStrictEqual(decodeFragmented(), [values[0]])
495 assert.deepStrictEqual(decodeBatch(), values)
496 assert.deepStrictEqual(fingerprintFeeds.single(), [values[0]])
497 assert.deepStrictEqual(fingerprintFeeds.fragmented(), [values[0]])
498 assert.deepStrictEqual(fingerprintFeeds.batch(), values)
499 assert.deepStrictEqual(decodeProtobufStream(), values)
500 assert.deepStrictEqual(await decodeNdjsonSingle(), [values[0]])
501 assert.deepStrictEqual(await decodeNdjsonFragmented(), [values[0]])
502 assert.deepStrictEqual(await decodeNdjsonBatch(), values)
503
504 return {
505 formats: [
506 { name: "SchemaBinary parser / single frame", framesPerOp: 1, decode: decodeSingle },
507 { name: "SchemaBinary parser / batch", framesPerOp: values.length, decode: decodeBatch },
508 { name: "SchemaBinary parser / fragmented", framesPerOp: 1, decode: decodeFragmented },
509 { name: "SchemaBinary fingerprint / single frame", framesPerOp: 1, decode: fingerprintFeeds.single },
510 { name: "SchemaBinary fingerprint / batch", framesPerOp: values.length, decode: fingerprintFeeds.batch },
511 { name: "SchemaBinary fingerprint / fragmented", framesPerOp: 1, decode: fingerprintFeeds.fragmented },
512 { name: "Protobuf decodeDelimited / batch", framesPerOp: values.length, decode: decodeProtobufStream },
513 { name: "NDJSON Channel / single frame", framesPerOp: 1, decode: decodeNdjsonSingle },
514 { name: "NDJSON Channel / batch", framesPerOp: values.length, decode: decodeNdjsonBatch },
515 { name: "NDJSON Channel / fragmented", framesPerOp: 1, decode: decodeNdjsonFragmented }
516 ],
517 sizes: [
518 { name: "SchemaBinary", frames: values.length, ...sizes(binaryStream) },
519 { name: "SchemaBinary fingerprint", frames: values.length, ...sizes(fingerprintStream) },
520 { name: "Protobuf", frames: values.length, ...sizes(protobufStream) },
521 { name: "NDJSON", frames: values.length, ...sizes(ndjsonStream) }
522 ]
523 }
524}
525
526const preparedStreams = await Promise.all([
527 ...cases.map((testCase) => ({
528 name: testCase.name,
529 prepared: prepareStream(
530 testCase.schema,
531 Array.from({ length: streamBatchSize }, () => testCase.value),
532 testCase.protobuf
533 )
534 })),
535 { name: "200 single-row frames", prepared: prepareStream(LargeRow, largeRows, largeRowProtobufFixture) }
536].map(async ({ name, prepared }) => ({ name, ...await prepared })))
537
538console.log(`Node ${process.version}; codec and schema construction excluded from timings.`)
539console.log("JSON sizes are UTF-8 bytes.")
540console.log("Protobuf uses prebuilt protobufjs descriptors; descriptor construction and encode adapters are excluded.")
541console.log(
542 "Compare formats within a case and direction in the same run; absolute rates vary with the machine and runtime."
543)
544
545console.table(prepared.flatMap((testCase) =>
546 testCase.formats.map((format) => ({
547 Case: testCase.name,
548 Format: format.name,
549 "Raw bytes": format.encodedSize,
550 "gzip -6 bytes": format.gzipSize,
551 "zstd bytes": format.zstdSize
552 }))
553))
554
555console.log(
556 "SchemaBinary streaming reuses one parser per feed shape; NDJSON runs its Channel per operation. Fragmented frames split after the first byte."
557)
558console.table(preparedStreams.flatMap((testCase) =>
559 testCase.sizes.map((format) => ({
560 Case: testCase.name,
561 Format: format.name,
562 Frames: format.frames,
563 "Raw bytes": format.encodedSize,
564 "Bytes / frame": format.encodedSize / format.frames,
565 "gzip -6 bytes": format.gzipSize,
566 "zstd bytes": format.zstdSize
567 }))
568))
569
570const bench = new Bench({
571 iterations: 1_000,
572 time: 0,
573 warmupIterations: 100,
574 warmupTime: 0,
575 timestampProvider: "hrtimeNow"
576})
577const tasks = new Map<string, { readonly caseName: string; readonly formatName: string; readonly direction: string }>()
578const sinkSentinel = Symbol("benchmark did not run")
579let sink: unknown = sinkSentinel
580
581for (const testCase of prepared) {
582 for (const format of testCase.formats) {
583 for (const [direction, run] of [["encode", format.encode], ["decode", format.decode]] as const) {
584 const name = `${testCase.name} / ${format.name} / ${direction}`
585 tasks.set(name, { caseName: testCase.name, formatName: format.name, direction })
586 bench.add(name, () => {
587 sink = run()
588 })
589 }
590 }
591}
592
593await bench.run()
594
595if (sink === sinkSentinel) {
596 throw new Error("Benchmark did not run")
597}
598
599console.table(bench.tasks.map((task) => {
600 const labels = tasks.get(task.name)!
601 const result = task.result
602 if (result?.state === "errored") {
603 return {
604 Case: labels.caseName,
605 Format: labels.formatName,
606 Direction: labels.direction,
607 Error: result.error.message
608 }
609 }
610 if (result?.state !== "completed") {
611 return {
612 Case: labels.caseName,
613 Format: labels.formatName,
614 Direction: labels.direction,
615 State: result?.state ?? "missing result"
616 }
617 }
618 return {
619 Case: labels.caseName,
620 Format: labels.formatName,
621 Direction: labels.direction,
622 "Throughput avg (ops/s)": Math.round(result.throughput.mean),
623 "Latency med (us/op)": (result.latency.p50 * 1_000).toFixed(2),
624 "Latency RME": `${result.latency.rme.toFixed(2)}%`,
625 Samples: result.latency.samplesCount
626 }
627}))
628
629const streamBench = new Bench({
630 iterations: 250,
631 time: 0,
632 warmupIterations: 25,
633 warmupTime: 0,
634 timestampProvider: "hrtimeNow"
635})
636const streamTasks = new Map<
637 string,
638 { readonly caseName: string; readonly formatName: string; readonly framesPerOp: number }
639>()
640
641for (const testCase of preparedStreams) {
642 for (const format of testCase.formats) {
643 const name = `${testCase.name} / ${format.name} / stream decode`
644 streamTasks.set(name, {
645 caseName: testCase.name,
646 formatName: format.name,
647 framesPerOp: format.framesPerOp
648 })
649 streamBench.add(name, () => {
650 const decoded = format.decode()
651 if (decoded instanceof Promise) {
652 return decoded.then((value) => {
653 sink = value
654 })
655 }
656 sink = decoded
657 })
658 }
659}
660
661await streamBench.run()
662
663console.table(streamBench.tasks.map((task) => {
664 const labels = streamTasks.get(task.name)!
665 const result = task.result
666 if (result?.state === "errored") {
667 return {
668 Case: labels.caseName,
669 Format: labels.formatName,
670 Error: result.error.message
671 }
672 }
673 if (result?.state !== "completed") {
674 return {
675 Case: labels.caseName,
676 Format: labels.formatName,
677 State: result?.state ?? "missing result"
678 }
679 }
680 return {
681 Case: labels.caseName,
682 Format: labels.formatName,
683 "Throughput avg (values/s)": Math.round(result.throughput.mean * labels.framesPerOp),
684 "Latency med (us/value)": (result.latency.p50 * 1_000 / labels.framesPerOp).toFixed(2),
685 "Latency RME": `${result.latency.rme.toFixed(2)}%`,
686 Samples: result.latency.samplesCount
687 }
688}))