diff --git a/packages/effect/benchmark/schema/SchemaBinary.md b/packages/effect/benchmark/schema/SchemaBinary.md index 083233a5ec7..62e166a103c 100644 --- a/packages/effect/benchmark/schema/SchemaBinary.md +++ b/packages/effect/benchmark/schema/SchemaBinary.md @@ -6,11 +6,13 @@ Run from the repository root: nix develop -c pnpm --dir packages/effect exec node benchmark/schema/SchemaBinary.ts ``` -The benchmark compares arena-backed `SchemaBinary`, `SchemaBinary` followed by an ownership copy, JSON, and Effect's Msgpack schema codec with the same Schema values. The copy case calls `.slice()` on every arena result, reproducing the allocation that the arena removes. It covers a small scalar-heavy record, a nested order payload, arrays and records, and 200 repeated records where framing and field-name overhead become visible. +The benchmark compares arena-backed `SchemaBinary`, `SchemaBinary` followed by an ownership copy, JSON, and Effect's Msgpack schema codec with the same Schema values. The copy case calls `.slice()` on every arena result, reproducing the allocation that the arena removes. It covers a small scalar-heavy record, a nested order payload, collections, index-signature records below and above the parser's 256-entry cache bound, and 200 repeated records where framing and field-name overhead become visible. Codec and schema construction happen before timing. Decode uses payloads encoded before timing. Each one-shot encode and decode task gets 100 warmup samples followed by 1,000 measured samples. The output includes UTF-8 payload sizes, average operations per second, median latency, relative margin of error, and sample count. -A second table compares stateful streaming decode. Each sample contains 32 concatenated frames. `SchemaBinary.parser(schema).feedSync` processes the binary stream; a reusable msgpackr `Unpackr.unpackMultiple` processes the Msgpack stream, then the same precompiled Schema decoder validates every value. Parser construction and stream encoding happen before timing. The streaming tasks get 25 warmup samples and 250 measured samples, and report throughput and median latency per decoded value. +A second table compares stateful streaming decode with one frame per feed, 32 frames per feed, and each frame split immediately after its first byte. That fragment boundary exercises the incremental frame-header path instead of merely splitting the body. It also includes 200 distinct repeated-record frames, rather than one frame containing a 200-element array. `SchemaBinary.parser(schema).feedSync` processes the binary stream; a reusable msgpackr `Unpackr.unpackMultiple` processes the Msgpack batch, then the same precompiled Schema decoder validates every value. Parser construction, stream encoding, and fragmentation happen before timing. The streaming tasks get 25 warmup samples and 250 measured samples, and report throughput and median latency per decoded value. + +Size output includes raw, gzip level 6, and zstd bytes for each one-shot payload and concatenated stream. Compression is applied to the whole stream so repeated field ids can share the compressor dictionary. This uses the core synchronous parsing work behind Effect's Msgpack stream decoder without Channel scheduling. JSON is omitted from the streaming table because its comparable Effect API is an NDJSON Channel, whose line framing and Channel runtime would measure a different layer. SchemaBinary has no streaming encoder in v1, so encode remains a one-shot comparison. @@ -18,63 +20,83 @@ Treat the measurements as a per-machine comparison within one run. Runtime versi ## Measured comparison -One run on Node 26.7.0, Linux x64, on the wire format that encodes integral -numbers as varints. One-shot throughput is the average of 1,000 measured -samples per task; treat these as a within-run comparison, not a portable score. - -Payload sizes are exact and portable, so they are the part of this table worth -comparing across machines. - -| Case | SchemaBinary | JSON | Msgpack | SchemaBinary vs Msgpack | -| ---------------------- | -----------: | ----: | ------: | ----------------------: | -| small record | 58 | 89 | 69 | 1.19x smaller | -| nested payload | 318 | 453 | 385 | 1.21x smaller | -| collections | 1452 | 1828 | 1462 | 1.01x smaller | -| large repeated records | 27646 | 57529 | 51283 | 1.85x smaller | - -Encoding integral numbers as varints took these payloads from 72, 347, 2553 and -31486 bytes. The collections case stays close to Msgpack because it is -dominated by genuinely non-integral floats, which stay in the eight-byte f64 -form. - -The encode rows exercise both ownership models on every case. Decode does not depend on output ownership, so the arena rows below are compared with Msgpack. - -| Case | SchemaBinary arena | SchemaBinary copy | Arena vs copy | Msgpack | -| ---------------------- | -----------------: | ----------------: | ------------: | ------: | -| small record | 462,101 | 488,902 | 0.95x | 480,796 | -| nested payload | 237,505 | 223,273 | 1.06x | 210,144 | -| collections | 34,139 | 33,853 | 1.01x | 22,929 | -| large repeated records | 6,001 | 5,932 | 1.01x | 3,461 | - -| Case | SchemaBinary decode | Msgpack decode | SchemaBinary vs Msgpack | -| ---------------------- | ------------------: | -------------: | ----------------------: | -| small record | 489,273 | 483,113 | 1.01x | -| nested payload | 231,041 | 203,513 | 1.14x | -| collections | 38,306 | 22,508 | 1.70x | -| large repeated records | 6,142 | 4,116 | 1.49x | - -Both codecs run the same Schema parser, which costs roughly 140 us per -direction on the largest case and sets a floor neither format can go below. -The numbers above therefore understate the difference between the two -serialization layers. - -The small record is now the noisiest case rather than the slowest one: at 58 -bytes both encode paths are dominated by fixed per-call cost, the arena row -carried a 6.9% relative margin of error in this run, and arena and copy trade -places between runs. The other three cases put the arena 1-6% ahead of the -ownership-copy control. Results returned by the arena keep an exact byte range -but can share a larger backing buffer, as documented by -`SchemaBinary.toCodec`. - -## Streaming decode comparison - -Each sample decodes 32 concatenated frames with parser state reused between samples. Throughput is normalized to decoded values per second; latency is normalized to microseconds per value. - -| Case | SchemaBinary parser | Msgpack unpackMultiple | SchemaBinary vs Msgpack | -| ---------------------- | ------------------: | ---------------------: | ----------------------: | -| small record | 820,916 | 626,090 | 1.31x | -| nested payload | 260,545 | 196,463 | 1.33x | -| collections | 37,921 | 21,565 | 1.76x | -| large repeated records | 5,725 | 5,038 | 1.14x | - -The stream-size table printed by the benchmark matters here. A persistent Msgpack `Packr` can reuse record definitions across frames, so its large repeated-record stream averages 19,484 bytes per frame versus 27,646 bytes for SchemaBinary in this run. The throughput comparison still favors SchemaBinary, but the two formats make different size-versus-parse-cost tradeoffs. +These measurements used Node 26.7.0 on Linux x64. Payload sizes are exact; throughput is machine-local. + +| Case | SchemaBinary raw/gzip/zstd | JSON raw/gzip/zstd | Msgpack raw/gzip/zstd | +| ---------------------- | -------------------------: | ------------------: | --------------------: | +| small record | 58 / 78 / 67 | 89 / 100 / 92 | 69 / 88 / 78 | +| nested payload | 318 / 305 / 300 | 453 / 303 / 309 | 385 / 304 / 299 | +| collections | 1452 / 792 / 776 | 1828 / 671 / 660 | 1462 / 788 / 805 | +| index signatures / 128 | 2251 / 689 / 656 | 2235 / 573 / 559 | 2203 / 676 / 629 | +| index signatures / 512 | 9355 / 2373 / 2189 | 9531 / 2266 / 2074 | 9287 / 2544 / 2304 | +| large repeated records | 27646 / 3065 / 3175 | 57529 / 3230 / 3008 | 51283 / 3516 / 3320 | + +Compression narrows or reverses the raw-size advantage on several cases. Repeated field ids compress well, so raw transport size and compressed transport size should be treated as separate results. + +One representative run of the combined tree produced the following one-shot encode rates in operations per second: + +| Case | SchemaBinary arena | SchemaBinary copy | Msgpack | +| ---------------------- | -----------------: | ----------------: | ------: | +| small record | 491,619 | 500,578 | 488,432 | +| nested payload | 232,943 | 220,297 | 208,758 | +| collections | 34,078 | 33,545 | 21,706 | +| index signatures / 128 | 14,286 | 14,191 | 33,791 | +| index signatures / 512 | 2,883 | 2,887 | 6,333 | +| large repeated records | 6,156 | 6,086 | 3,521 | + +The corresponding one-shot decode rates were: + +| Case | SchemaBinary | Msgpack | +| ---------------------- | -----------: | ------: | +| small record | 496,806 | 470,364 | +| nested payload | 221,214 | 201,178 | +| collections | 37,080 | 21,081 | +| index signatures / 128 | 20,494 | 29,208 | +| index signatures / 512 | 4,979 | 4,456 | +| large repeated records | 5,876 | 4,000 | + +The small cases are dominated by fixed per-call cost, so their arena and ownership-copy rows can trade places between runs. The larger cases are more stable; all throughput numbers remain machine-local rather than portable scores. + +The per-frame repeated-record stream makes that distinction concrete: + +| Format | Frames | Raw bytes | gzip -6 | zstd | +| ------------ | -----: | --------: | ------: | ---: | +| SchemaBinary | 200 | 27840 | 3109 | 3174 | +| Msgpack | 200 | 51520 | 2956 | 2888 | + +## Parser optimization effects + +The initial investigation measured each change before they were stacked. Values below are medians of three interleaved runs on commit `265c3b0a19`; rates are decoded values per second. This isolated-change baseline is intentionally different from the later end-to-end comparison against `8e60b33c2`. + +| Variant | Small batch 32 | Small single | Small fragmented | Nested batch 32 | Collections batch 32 | +| -------------------------------------------- | -------------: | -----------: | ---------------: | --------------: | -------------------: | +| Baseline | 829k | 894k | 808k | 261k | 38.9k | +| Reader/DataView reuse | 869k | 990k | 868k | 272k | 38.9k | +| Reader reuse plus persistent signature cache | 932k | 1038k | 947k | 283k | 46.6k | + +The number-based header path was isolated separately: +9.5% small batch, +1.4% small single, +7.2% small fragmented, and +4.5% nested batch. These percentages are not added to the Reader and cache results. + +The final combined implementation was then compared with the pre-refactor `8e60b33c2` head using the same extended benchmark on both trees. The table reports medians of three full runs. Single-frame tasks have visibly higher fixed-cost noise than batch and fragmented tasks. + +| Case | Feed | Baseline | Combined | Change | +| ------------------------- | ---------- | -------: | -------: | -----: | +| small record | single | 726k | 809k | +11.5% | +| small record | batch 32 | 799k | 949k | +18.8% | +| small record | fragmented | 651k | 770k | +18.3% | +| nested payload | single | 224k | 235k | +5.0% | +| nested payload | batch 32 | 250k | 269k | +7.5% | +| nested payload | fragmented | 235k | 254k | +7.7% | +| collections | single | 36.1k | 43.1k | +19.2% | +| collections | batch 32 | 37.0k | 46.0k | +24.3% | +| collections | fragmented | 37.5k | 46.3k | +23.4% | +| index signatures / 128 | single | 20.8k | 39.5k | +90.1% | +| index signatures / 128 | batch 32 | 20.8k | 39.4k | +89.7% | +| index signatures / 128 | fragmented | 20.9k | 40.3k | +92.6% | +| index signatures / 512 | single | 4.9k | 6.5k | +32.0% | +| index signatures / 512 | batch 32 | 4.6k | 6.5k | +42.2% | +| index signatures / 512 | fragmented | 4.9k | 6.9k | +39.5% | +| per-frame repeated record | single | 581k | 649k | +11.8% | +| per-frame repeated record | batch 200 | 647k | 710k | +9.6% | +| per-frame repeated record | fragmented | 578k | 634k | +9.8% | + +The collections result matches the original roughly 20% target. The 128-key case benefits more because the parser reuses every classification on later frames. The 512-key case verifies that inputs wider than the cache bound still improve rather than thrash: the cache retains roughly half of the classifications and admits at most one replacement per frame. Those targets were diagnostic estimates, not acceptance thresholds. diff --git a/packages/effect/benchmark/schema/SchemaBinary.ts b/packages/effect/benchmark/schema/SchemaBinary.ts index eb5e4e32b21..d15b414d7f1 100644 --- a/packages/effect/benchmark/schema/SchemaBinary.ts +++ b/packages/effect/benchmark/schema/SchemaBinary.ts @@ -2,9 +2,11 @@ import { Schema } from "effect" import { Msgpack, SchemaBinary } from "effect/unstable/encoding" import { Packr, Unpackr } from "msgpackr" import assert from "node:assert/strict" +import { gzipSync, zstdCompressSync } from "node:zlib" import { Bench } from "tinybench" const streamBatchSize = 32 +const repeatedRecordStreamSize = 200 const SmallRecord = Schema.Struct({ id: Schema.Number, @@ -62,6 +64,20 @@ const LargeRow = Schema.Struct({ const LargePayload = Schema.Array(LargeRow) +const largeRows = Array.from({ length: repeatedRecordStreamSize }, (_, index) => ({ + transactionIdentifier: `transaction-${index.toString().padStart(4, "0")}`, + customerIdentifier: `customer-${index % 37}`, + productDescription: `Product ${index % 19} with a repeated descriptive field value`, + fulfillmentLocation: ["London", "New York", "Singapore", "Sydney"][index % 4]!, + quantityPurchased: index % 9 + 1, + unitPriceInCents: 500 + index % 73 * 25, + discountInBasisPoints: index % 5 * 125, + requiresManualReview: index % 17 === 0 +})) + +const metrics = (count: number) => + Object.fromEntries(Array.from({ length: count }, (_, index) => [`metric-${index}`, index * 1.25])) + const cases = [ { name: "small record", @@ -113,45 +129,48 @@ const cases = [ buckets: Array.from({ length: 8 }, (_, bucket) => Array.from({ length: 16 }, (_, index) => bucket * 100 + index)) } }, + { + name: "index signatures / 128 keys", + schema: Schema.Record(Schema.String, Schema.Number), + value: metrics(128) + }, + { + name: "index signatures / 512 keys", + schema: Schema.Record(Schema.String, Schema.Number), + value: metrics(512) + }, { name: "large repeated records", schema: LargePayload, - value: Array.from({ length: 200 }, (_, index) => ({ - transactionIdentifier: `transaction-${index.toString().padStart(4, "0")}`, - customerIdentifier: `customer-${index % 37}`, - productDescription: `Product ${index % 19} with a repeated descriptive field value`, - fulfillmentLocation: ["London", "New York", "Singapore", "Sydney"][index % 4]!, - quantityPurchased: index % 9 + 1, - unitPriceInCents: 500 + index % 73 * 25, - discountInBasisPoints: index % 5 * 125, - requiresManualReview: index % 17 === 0 - })) + value: largeRows } ] as const interface Format { readonly name: string readonly encodedSize: number + readonly gzipSize: number + readonly zstdSize: number readonly encode: () => unknown readonly decode: () => unknown } interface StreamFormat { readonly name: string - readonly encodedSize: number + readonly framesPerOp: number readonly decode: () => ReadonlyArray } -const textEncoder = new TextEncoder() - -const repeatFrame = (frame: Uint8Array, count: number): Uint8Array => { - const out = new Uint8Array(frame.length * count) - for (let i = 0; i < count; i++) { - out.set(frame, i * frame.length) - } - return out +interface StreamSize { + readonly name: string + readonly frames: number + readonly encodedSize: number + readonly gzipSize: number + readonly zstdSize: number } +const textEncoder = new TextEncoder() + const concatFrames = (frames: ReadonlyArray): Uint8Array => { const out = new Uint8Array(frames.reduce((length, frame) => length + frame.length, 0)) let offset = 0 @@ -162,12 +181,17 @@ const concatFrames = (frames: ReadonlyArray): Uint8Array => ({ + encodedSize: encoded.length, + gzipSize: gzipSync(encoded).length, + zstdSize: zstdCompressSync(encoded).length +}) + const prepare = >( schema: S, value: S["Type"] ): { readonly formats: ReadonlyArray - readonly streamFormats: ReadonlyArray } => { const jsonSchema = Schema.toCodecJson(schema) const binaryCodec = SchemaBinary.toCodec(schema) @@ -184,74 +208,107 @@ const prepare = >( const binary = binaryEncode(value) const binaryCopy = binary.slice() const json = jsonEncode(value) + const jsonBytes = textEncoder.encode(json) const msgpack = msgpackEncode(value) assert.deepStrictEqual(binaryDecode(binary), value) assert.deepStrictEqual(jsonDecode(json), value) assert.deepStrictEqual(msgpackDecode(msgpack), value) - const expectedStream = Array.from({ length: streamBatchSize }, () => value) - const binaryStream = repeatFrame(binary, streamBatchSize) - const encodeMsgpackValue = Schema.encodeUnknownSync(jsonSchema) - const msgpackPackr = new Packr() - const msgpackValue = encodeMsgpackValue(value) - const msgpackStream = concatFrames( - Array.from({ length: streamBatchSize }, () => msgpackPackr.pack(msgpackValue).slice()) - ) - const binaryParser = SchemaBinary.parser(schema) - const msgpackUnpackr = new Unpackr() - const decodeMsgpackValue = Schema.decodeUnknownSync(jsonSchema) - const decodeBinaryStream = () => binaryParser.feedSync(binaryStream) - const decodeMsgpackStream = () => - msgpackUnpackr.unpackMultiple(msgpackStream).map((value) => decodeMsgpackValue(value)) - - assert.deepStrictEqual(decodeBinaryStream(), expectedStream) - assert.deepStrictEqual(decodeMsgpackStream(), expectedStream) - return { formats: [ { name: "SchemaBinary arena", - encodedSize: binary.length, + ...sizes(binary), encode: () => binaryEncode(value), decode: () => binaryDecode(binary) }, { name: "SchemaBinary copy", - encodedSize: binaryCopy.length, + ...sizes(binaryCopy), encode: () => binaryEncode(value).slice(), decode: () => binaryDecode(binaryCopy) }, { name: "JSON", - encodedSize: textEncoder.encode(json).length, + ...sizes(jsonBytes), encode: () => jsonEncode(value), decode: () => jsonDecode(json) }, { name: "Msgpack", - encodedSize: msgpack.length, + ...sizes(msgpack), encode: () => msgpackEncode(value), decode: () => msgpackDecode(msgpack) } - ], - streamFormats: [ - { - name: "SchemaBinary parser", - encodedSize: binaryStream.length, - decode: decodeBinaryStream - }, - { - name: "Msgpack unpackMultiple", - encodedSize: msgpackStream.length, - decode: decodeMsgpackStream - } ] } } const prepared = cases.map((testCase) => ({ name: testCase.name, ...prepare(testCase.schema, testCase.value) })) +const prepareStream = >( + schema: S, + values: ReadonlyArray +): { readonly formats: ReadonlyArray; readonly sizes: ReadonlyArray } => { + const binaryCodec = SchemaBinary.toCodec(schema) + const binaryEncode = Schema.encodeUnknownSync(binaryCodec) + const binaryFrames = values.map((value) => binaryEncode(value)) + const binaryStream = concatFrames(binaryFrames) + const binaryFragments = binaryFrames.map((frame) => { + return [frame.subarray(0, 1), frame.subarray(1)] as const + }) + + const jsonSchema = Schema.toCodecJson(schema) + const encodeMsgpackValue = Schema.encodeUnknownSync(jsonSchema) + const decodeMsgpackValue = Schema.decodeUnknownSync(jsonSchema) + const msgpackPackr = new Packr() + const msgpackStream = concatFrames(values.map((value) => msgpackPackr.pack(encodeMsgpackValue(value)).slice())) + const msgpackUnpackr = new Unpackr() + + const singleParser = SchemaBinary.parser(schema) + const fragmentedParser = SchemaBinary.parser(schema) + const batchParser = SchemaBinary.parser(schema) + let singleIndex = 0 + let fragmentedIndex = 0 + const decodeSingle = () => singleParser.feedSync(binaryFrames[singleIndex++ % binaryFrames.length]) + const decodeFragmented = () => { + const fragments = binaryFragments[fragmentedIndex++ % binaryFragments.length] + const first = fragmentedParser.feedSync(fragments[0]) + const second = fragmentedParser.feedSync(fragments[1]) + return first.length === 0 ? second : [...first, ...second] + } + const decodeBatch = () => batchParser.feedSync(binaryStream) + const decodeMsgpackStream = () => + msgpackUnpackr.unpackMultiple(msgpackStream).map((value) => decodeMsgpackValue(value)) + + assert.deepStrictEqual(decodeSingle(), [values[0]]) + assert.deepStrictEqual(decodeFragmented(), [values[0]]) + assert.deepStrictEqual(decodeBatch(), values) + assert.deepStrictEqual(decodeMsgpackStream(), values) + + return { + formats: [ + { name: "SchemaBinary parser / single frame", framesPerOp: 1, decode: decodeSingle }, + { name: "SchemaBinary parser / batch", framesPerOp: values.length, decode: decodeBatch }, + { name: "SchemaBinary parser / fragmented", framesPerOp: 1, decode: decodeFragmented }, + { name: "Msgpack unpackMultiple / batch", framesPerOp: values.length, decode: decodeMsgpackStream } + ], + sizes: [ + { name: "SchemaBinary", frames: values.length, ...sizes(binaryStream) }, + { name: "Msgpack", frames: values.length, ...sizes(msgpackStream) } + ] + } +} + +const preparedStreams = [ + ...cases.map((testCase) => ({ + name: testCase.name, + ...prepareStream(testCase.schema, Array.from({ length: streamBatchSize }, () => testCase.value)) + })), + { name: "per-frame repeated records", ...prepareStream(LargeRow, largeRows) } +] + console.log(`Node ${process.version}; codec and schema construction excluded from timings.`) console.log("JSON and Msgpack use the same Schema.toCodecJson representation; JSON sizes are UTF-8 bytes.") console.log( @@ -262,20 +319,24 @@ console.table(prepared.flatMap((testCase) => testCase.formats.map((format) => ({ Case: testCase.name, Format: format.name, - "Encoded bytes": format.encodedSize + "Raw bytes": format.encodedSize, + "gzip -6 bytes": format.gzipSize, + "zstd bytes": format.zstdSize })) )) console.log( - `Streaming decode reuses one parser per format and processes ${streamBatchSize} concatenated frames per sample.` + "Streaming decode reuses one parser per feed shape. Fragmented frames split after the first byte." ) -console.table(prepared.flatMap((testCase) => - testCase.streamFormats.map((format) => ({ +console.table(preparedStreams.flatMap((testCase) => + testCase.sizes.map((format) => ({ Case: testCase.name, Format: format.name, - Frames: streamBatchSize, - "Encoded bytes": format.encodedSize, - "Bytes / frame": format.encodedSize / streamBatchSize + Frames: format.frames, + "Raw bytes": format.encodedSize, + "Bytes / frame": format.encodedSize / format.frames, + "gzip -6 bytes": format.gzipSize, + "zstd bytes": format.zstdSize })) )) @@ -345,12 +406,19 @@ const streamBench = new Bench({ warmupTime: 0, timestampProvider: "hrtimeNow" }) -const streamTasks = new Map() +const streamTasks = new Map< + string, + { readonly caseName: string; readonly formatName: string; readonly framesPerOp: number } +>() -for (const testCase of prepared) { - for (const format of testCase.streamFormats) { +for (const testCase of preparedStreams) { + for (const format of testCase.formats) { const name = `${testCase.name} / ${format.name} / stream decode` - streamTasks.set(name, { caseName: testCase.name, formatName: format.name }) + streamTasks.set(name, { + caseName: testCase.name, + formatName: format.name, + framesPerOp: format.framesPerOp + }) streamBench.add(name, () => { sink = format.decode() }) @@ -379,8 +447,8 @@ console.table(streamBench.tasks.map((task) => { return { Case: labels.caseName, Format: labels.formatName, - "Throughput avg (values/s)": Math.round(result.throughput.mean * streamBatchSize), - "Latency med (us/value)": (result.latency.p50 * 1_000 / streamBatchSize).toFixed(2), + "Throughput avg (values/s)": Math.round(result.throughput.mean * labels.framesPerOp), + "Latency med (us/value)": (result.latency.p50 * 1_000 / labels.framesPerOp).toFixed(2), "Latency RME": `${result.latency.rme.toFixed(2)}%`, Samples: result.latency.samplesCount } diff --git a/packages/effect/src/unstable/encoding/SchemaBinary.ts b/packages/effect/src/unstable/encoding/SchemaBinary.ts index cdf1f97f5b9..6209d1e6265 100644 --- a/packages/effect/src/unstable/encoding/SchemaBinary.ts +++ b/packages/effect/src/unstable/encoding/SchemaBinary.ts @@ -219,6 +219,14 @@ function decodeUtf8( const OUTPUT_ARENA_SIZE = 8 * 1024 +// Parser cache entries are derived from wire keys, so this is a hard bound on +// attacker-controlled state retained across feed calls. Once full, the cache +// admits at most one FIFO replacement per frame. Repeated frames wider than +// the cache therefore keep their existing hits instead of cycling every entry. +// One replacement still adapts to gradual key changes without making churn +// scale with either frame width or cache capacity. +const PARSER_INDEX_SIGNATURE_CACHE_SIZE = 256 + interface OutputArena { readonly buf: Uint8Array offset: number @@ -397,27 +405,110 @@ class Writer { } } -class Reader { - pos: number - readonly buf: Uint8Array - readonly view: DataView - end: number +function matchIndexSignature( + layout: StructLayout, + key: string, + options: SchemaAST.ParseOptions +): ExtraSignature | undefined { + return layout.extra.find((s) => SchemaAST.getIndexSignatureKeys({ [key]: null }, s.parameter, options).length > 0) +} + +class IndexSignatureCache { + entries = new WeakMap>() + readonly orderLayouts: Array | undefined + readonly orderKeys: Array | undefined readonly options: SchemaAST.ParseOptions - readonly indexSignatures: WeakMap> - constructor( + size = 0 + next = 0 + replace = false + constructor(options: SchemaAST.ParseOptions, capacity?: number) { + if (capacity !== undefined && capacity < 1) throw new Error("IndexSignatureCache capacity must be positive") + this.options = options + this.orderLayouts = capacity === undefined ? undefined : new Array(capacity) + this.orderKeys = capacity === undefined ? undefined : new Array(capacity) + } + beginFrame() { + this.replace = true + } + find(layout: StructLayout, key: string): ExtraSignature | undefined { + let entries = this.entries.get(layout) + if (entries === undefined) { + entries = new Map() + this.entries.set(layout, entries) + } + if (entries.has(key)) return entries.get(key) + const signature = matchIndexSignature(layout, key, this.options) + const orderLayouts = this.orderLayouts + if (orderLayouts === undefined) { + // This unbounded form only lives for one top-level encode or decode. + entries.set(key, signature) + return signature + } + const orderKeys = this.orderKeys! + if (this.size < orderLayouts.length) { + orderLayouts[this.size] = layout + orderKeys[this.size] = key + this.size++ + entries.set(key, signature) + } else if (this.replace) { + this.replace = false + const evictedLayout = orderLayouts[this.next]! + const evictedEntries = evictedLayout === layout ? entries : this.entries.get(evictedLayout) + evictedEntries?.delete(orderKeys[this.next]!) + orderLayouts[this.next] = layout + orderKeys[this.next] = key + this.next = (this.next + 1) % orderLayouts.length + entries.set(key, signature) + } + return signature + } + clear() { + this.entries = new WeakMap() + this.orderLayouts?.fill(undefined) + this.orderKeys?.fill(undefined) + this.size = this.next = 0 + this.replace = false + } +} + +const EMPTY_READER_BUFFER = new Uint8Array(0) +const EMPTY_READER_VIEW = new DataView(EMPTY_READER_BUFFER.buffer) +const EMPTY_PARSE_OPTIONS: SchemaAST.ParseOptions = {} + +class Reader { + pos = 0 + buf: Uint8Array = EMPTY_READER_BUFFER + view: DataView = EMPTY_READER_VIEW + end = 0 + options: SchemaAST.ParseOptions = EMPTY_PARSE_OPTIONS + indexSignatures: IndexSignatureCache | undefined + reset( buf: Uint8Array, start: number, end: number, options: SchemaAST.ParseOptions, - indexSignatures = new WeakMap>() + indexSignatures: IndexSignatureCache ) { + if ( + this.buf.buffer !== buf.buffer || + this.buf.byteOffset !== buf.byteOffset || + this.buf.byteLength !== buf.byteLength + ) { + this.view = new DataView(buf.buffer, buf.byteOffset, buf.byteLength) + } this.buf = buf - this.view = new DataView(buf.buffer, buf.byteOffset, buf.byteLength) this.pos = start this.end = end this.options = options this.indexSignatures = indexSignatures } + release() { + this.pos = this.end = 0 + this.buf = EMPTY_READER_BUFFER + this.view = EMPTY_READER_VIEW + this.options = EMPTY_PARSE_OPTIONS + this.indexSignatures = undefined + } get remaining(): number { return this.end - this.pos } @@ -1394,7 +1485,7 @@ function matchesLayout(layout: Layout, value: unknown): boolean { interface EncodeContext { readonly options: SchemaAST.ParseOptions - indexSignatures: WeakMap> | undefined + indexSignatures: IndexSignatureCache | undefined } function encodeFail(expected: string, input: unknown, options: SchemaAST.ParseOptions): never { @@ -1487,25 +1578,6 @@ function isCyclic(value: unknown, stack = new Set()): boolean { return false } -function findIndexSignature( - layout: StructLayout, - key: string, - options: SchemaAST.ParseOptions, - cache: WeakMap> -): ExtraSignature | undefined { - let entries = cache.get(layout) - if (entries === undefined) { - entries = new Map() - cache.set(layout, entries) - } - if (entries.has(key)) return entries.get(key) - const signature = layout.extra.find((s) => - SchemaAST.getIndexSignatureKeys({ [key]: null }, s.parameter, options).length > 0 - ) - entries.set(key, signature) - return signature -} - function encodeSized(ctx: EncodeContext, layout: Layout, value: unknown, w: Writer) { const mark = w.beginSized() encodeValue(ctx, layout, value, w) @@ -1546,12 +1618,7 @@ function encodeStructFields(ctx: EncodeContext, layout: StructLayout, value: obj const pairs: Array<[Uint8Array, string, ExtraSignature]> = [] for (const key of Object.keys(obj)) { if (named.has(key)) continue - const signature = findIndexSignature( - layout, - key, - ctx.options, - ctx.indexSignatures ??= new WeakMap() - ) + const signature = (ctx.indexSignatures ??= new IndexSignatureCache(ctx.options)).find(layout, key) if (signature === undefined) continue pairs.push([utf8Encode.encode(key), key, signature]) } @@ -1973,7 +2040,7 @@ function decodeExtraPairs(layout: StructLayout, r: Reader, out: Record( let bufferEnd = 0 let stashed: Schema.SchemaError | undefined let spent = false + const body = new Reader() + const indexSignatures = new IndexSignatureCache(parseOptions, PARSER_INDEX_SIGNATURE_CACHE_SIZE) + + const release = () => { + body.release() + indexSignatures.clear() + buffer = EMPTY_READER_BUFFER + bufferStart = bufferEnd = 0 + } const failSync = (expected: string, input?: unknown): never => { throw new Schema.SchemaError(new SchemaIssue.InvalidValue({ expected }, input, parseOptions)) @@ -2495,32 +2581,47 @@ export function parser( const fail = (expected: string, input?: unknown): Array => { spent = true const error = new Schema.SchemaError(new SchemaIssue.InvalidValue({ expected }, input, parseOptions)) + release() if (out.length === 0) throw error stashed = error - buffer = new Uint8Array(0) - bufferStart = bufferEnd = 0 return out } while (true) { // frame length varint: fewer than 10 bytes without a terminator waits, - // 10 continuation bytes is malformed immediately - let n = BIGINT_ZERO + // 10 continuation bytes is malformed immediately. Common headers that + // terminate in fewer than seven groups stay in number arithmetic; + // longer valid headers retain the exact bigint path. + let frameLen = 0 let headerLen = -1 const buffered = bufferEnd - bufferStart - for (let i = 0; i < Math.min(10, buffered); i++) { + let scale = 1 + for (let i = 0; i < Math.min(6, buffered); i++) { const b = buffer[bufferStart + i] - n |= BigInt(b & 0x7F) << BigInt(i * 7) + frameLen += (b & 0x7F) * scale if ((b & 0x80) === 0) { headerLen = i + 1 break } + scale *= 128 } if (headerLen === -1) { - if (buffered >= 10) return fail("uvarint", buffer.subarray(bufferStart, bufferStart + 10)) - return out + if (buffered < 6) return out + let n = BigInt(frameLen) + for (let i = 6; i < Math.min(10, buffered); i++) { + const b = buffer[bufferStart + i] + n |= BigInt(b & 0x7F) << BigInt(i * 7) + if ((b & 0x80) === 0) { + headerLen = i + 1 + if (n > MAX_SAFE_BIGINT) return fail("safe integer length", n) + frameLen = Number(n) + break + } + } + if (headerLen === -1) { + if (buffered >= 10) return fail("uvarint", buffer.subarray(bufferStart, bufferStart + 10)) + return out + } } - if (n > MAX_SAFE_BIGINT) return fail("safe integer length", n) - const frameLen = Number(n) if (frameLen === 0) return fail("nonzero frame length", frameLen) if (maxFrameSize !== undefined && frameLen > maxFrameSize) { return fail("frame within maxFrameSize", frameLen) @@ -2530,10 +2631,12 @@ export function parser( try { issuePathLen = 0 const bodyStart = bufferStart + headerLen - const body = new Reader(buffer, bodyStart, bodyStart + frameLen, parseOptions) + indexSignatures.beginFrame() + body.reset(buffer, bodyStart, bodyStart + frameLen, parseOptions, indexSignatures) out.push(decodeEncoded(decodeFrameBody(layout, body))) } catch (e) { spent = true + release() const error = e instanceof IssueError ? new Schema.SchemaError(e.issue) : Schema.isSchemaError(e) @@ -2543,8 +2646,6 @@ export function parser( })() if (out.length === 0) throw error stashed = error - buffer = new Uint8Array(0) - bufferStart = bufferEnd = 0 return out } finally { issuePathLen = savedPathLen @@ -2557,7 +2658,9 @@ export function parser( if (stashed !== undefined) return takeStashed() if (spent) return failSync("parser is spent") spent = true - if (bufferStart < bufferEnd) return failSync("complete value", buffer.subarray(bufferStart, bufferEnd)) + const input = bufferStart < bufferEnd ? buffer.subarray(bufferStart, bufferEnd) : undefined + release() + if (input !== undefined) return failSync("complete value", input) }, feed: (chunk) => Effect.suspend(() => { diff --git a/packages/effect/test/unstable/encoding/SchemaBinary.test.ts b/packages/effect/test/unstable/encoding/SchemaBinary.test.ts index 172f30648ee..e986e9b0067 100644 --- a/packages/effect/test/unstable/encoding/SchemaBinary.test.ts +++ b/packages/effect/test/unstable/encoding/SchemaBinary.test.ts @@ -481,6 +481,121 @@ describe("SchemaBinary", () => { assert.deepStrictEqual(values, Array.from({ length: 100 }, (_, i) => i)) }) + it("waits for a fragmented multi-byte frame header", () => { + const value = "x".repeat(300) + const bytes = encode(Schema.String, value) + const parser = SchemaBinary.parser(Schema.String) + + assert.deepStrictEqual(parser.feedSync(bytes.slice(0, 1)), []) + assert.deepStrictEqual(parser.feedSync(bytes.slice(1)), [value]) + parser.endSync() + }) + + it("uses the bigint header path for longer safe lengths", () => { + // Canonical uvarint(Number.MAX_SAFE_INTEGER): seven continuation groups + // followed by the final four bits. maxFrameSize rejects it before the + // parser waits for an impractically large body. + const bytes = Uint8Array.of(0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x0F) + const parser = SchemaBinary.parser(Schema.String, { maxFrameSize: 1 }) + + assert.deepStrictEqual(parser.feedSync(bytes.slice(0, 7)), []) + assert.match(schemaError(() => parser.feedSync(bytes.slice(7))).message, /frame within maxFrameSize/) + }) + + it("keeps parser index-signature caching bounded with FIFO eviction", () => { + let checks = 0 + const Key = Schema.String.check(Schema.makeFilter((_: string) => { + checks++ + return true + })) + const Writer = Schema.Record(Schema.String, Schema.Number) + const parser = SchemaBinary.parser(Schema.Record(Key, Schema.Number)) + const feed = (key: string) => { + const value = { [key]: 1 } + assert.deepStrictEqual(parser.feedSync(encode(Writer, value)), [value]) + } + + // The parser cache holds 256 attacker-controlled (layout, key) pairs. + // The 257th evicts key-0, while key-1 remains cached. + for (let i = 0; i < 257; i++) feed(`key-${i}`) + const beforeHit = checks + feed("key-1") + const hitChecks = checks - beforeHit + const beforeMiss = checks + feed("key-0") + const missChecks = checks - beforeMiss + + assert.strictEqual(missChecks, hitChecks + 1) + parser.endSync() + }) + + it("does not thrash the index-signature cache above its bound", () => { + let checks = 0 + const Key = Schema.String.check(Schema.makeFilter((_: string) => { + checks++ + return true + })) + const Writer = Schema.Record(Schema.String, Schema.Number) + const Reader = Schema.Record(Key, Schema.Number) + const allHitParser = SchemaBinary.parser(Reader) + const allHitValue = Object.fromEntries(Array.from({ length: 256 }, (_, index) => [`hit-${index}`, index])) + const allHitBytes = encode(Writer, allHitValue) + + assert.deepStrictEqual(allHitParser.feedSync(allHitBytes), [allHitValue]) + const beforeAllHit = checks + assert.deepStrictEqual(allHitParser.feedSync(allHitBytes), [allHitValue]) + const allHitChecksPerKey = (checks - beforeAllHit) / 256 + allHitParser.endSync() + + const parser = SchemaBinary.parser(Reader) + const value = Object.fromEntries(Array.from({ length: 300 }, (_, index) => [`key-${index}`, index])) + const bytes = encode(Writer, value) + + assert.deepStrictEqual(parser.feedSync(bytes), [value]) + const firstChecks = checks + assert.deepStrictEqual(parser.feedSync(bytes), [value]) + const secondChecks = checks - firstChecks + const misses = secondChecks - allHitChecksPerKey * 300 + + // Derive the decoder's per-key work from a fully warm parser. Each cache + // miss adds one classification check beyond that all-hit baseline. + assert.strictEqual(misses, 45) + parser.endSync() + }) + + it("isolates index-signature caches by parser options", () => { + const Key = Schema.String.check(Schema.makeFilter((key: string) => key.startsWith("allowed-"))) + const Writer = Schema.Record(Schema.String, Schema.Number) + const Reader = Schema.Record(Key, Schema.Number) + const bytes = encode(Writer, { denied: 1 }) + const strict = SchemaBinary.parser(Reader) + const unchecked = SchemaBinary.parser(Reader, { disableChecks: true }) + + assert.deepStrictEqual(strict.feedSync(bytes), [{}]) + assert.deepStrictEqual(unchecked.feedSync(bytes), [{ denied: 1 }]) + strict.endSync() + unchecked.endSync() + }) + + it("keeps nested SchemaBinary decodes independent from the outer reader", () => { + const Inner = Schema.Struct({ id: Schema.Number, label: Schema.String }) + const Outer = Schema.Struct({ id: Schema.String, inner: SchemaBinary.toCodec(Inner) }) + const first = { id: "first", inner: { id: 1, label: "one" } } + const second = { id: "second", inner: { id: 2, label: "two" } } + const parser = SchemaBinary.parser(Outer) + + assert.deepStrictEqual(parser.feedSync(concat(encode(Outer, first), encode(Outer, second))), [first, second]) + parser.endSync() + }) + + it("returns a checked-out one-shot reader after an exceptional decode", () => { + const codec = SchemaBinary.toCodec(Schema.String) + const decode = Schema.decodeUnknownSync(codec) + + assert.match(schemaError(() => decode(Uint8Array.of(2, 0x10, 0xFF))).message, /utf-8/) + assert.strictEqual(decode(encode(Schema.String, "after failure")), "after failure") + }) + it("delivers completed values before reporting a later failure", () => { const good = encode(Schema.Number, 1) const bad = encode(Schema.Number, 2).slice(0, 2) @@ -504,6 +619,14 @@ describe("SchemaBinary", () => { assert.isTrue(SchemaIssue.hasInput(error.issue)) }) + it("rejects a terminated header above the safe-integer bound", () => { + const bytes = Uint8Array.of(0x80, 0x80, 0x80, 0x80, 0x80, 0x80, 0x80, 0x10) + const parser = SchemaBinary.parser(Schema.String) + + assert.match(schemaError(() => parser.feedSync(bytes)).message, /safe integer length/) + assert.match(schemaError(() => parser.feedSync(new Uint8Array())).message, /parser is spent/) + }) + it.effect("wraps feed and end in SchemaError effects", () => Effect.gen(function*() { const parser = SchemaBinary.parser(Schema.Number) @@ -737,6 +860,16 @@ describe("SchemaBinary", () => { ) }) + it("keeps one-shot index-signature caching within each options call", () => { + const Key = Schema.String.check(Schema.makeFilter((key: string) => key.startsWith("allowed-"))) + const Writer = Schema.Record(Schema.String, Schema.Number) + const codec = SchemaBinary.toCodec(Schema.Record(Key, Schema.Number)) + const bytes = encode(Writer, { denied: 1 }) + + assert.deepStrictEqual(Schema.decodeUnknownSync(codec)(bytes), {}) + assert.deepStrictEqual(Schema.decodeUnknownSync(codec, { disableChecks: true })(bytes), { denied: 1 }) + }) + it("honors errors all for missing fields", () => { const bytes = encode(Schema.Struct({}), {}) const Reader = Schema.Struct({ a: Schema.String, b: Schema.Number })