Return SchemaBinary output from a shared arena

.changeset/schema-binary-codec.md

@@ -5,3 +5,5 @@
 Add `SchemaBinary`, a compact binary codec derived from the Schema AST.
 
 `SchemaBinary.toCodec(schema)` compiles a wire layout from the encoded-side AST on each side, so field names never appear on the payload. `SchemaBinary.parser(schema)` reads concatenated frames from a stream, and `SchemaBinary.fieldId(n)` pins a field's wire id so it survives a rename.
+
+Encoded results are stable views into a shared bump-allocated arena. Their byte range is exact, but their backing buffer may be larger, have a non-zero offset, and contain other encoded results. Copy a result before transferring its buffer when independent ownership is required.

packages/effect/benchmark/schema/SchemaBinary.md

@@ -6,7 +6,7 @@ Run from the repository root:
 nix develop -c pnpm --dir packages/effect exec node benchmark/schema/SchemaBinary.ts
 ```
 
-The benchmark compares `SchemaBinary`, JSON, and Effect's Msgpack schema codec with the same Schema values. 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, arrays and records, 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.
 
@@ -21,27 +21,33 @@ Treat the measurements as a per-machine comparison within one run. Runtime versi
 One run on Node 26.7.0, Linux x64. One-shot throughput is the average of 1,000 measured
 samples per task; treat these as a within-run comparison, not a portable score.
 
-| Case                   | Direction | SchemaBinary | Msgpack | SchemaBinary vs Msgpack |
-| ---------------------- | --------- | -----------: | ------: | ----------------------: |
-| small record           | encode    |      413,942 | 468,259 |                   0.88x |
-| small record           | decode    |      472,582 | 474,553 |                   1.00x |
-| nested payload         | encode    |      230,222 | 205,683 |                   1.12x |
-| nested payload         | decode    |      227,531 | 199,069 |                   1.14x |
-| collections            | encode    |       33,596 |  21,846 |                   1.54x |
-| collections            | decode    |       37,822 |  21,604 |                   1.75x |
-| large repeated records | encode    |        6,366 |   3,543 |                   1.80x |
-| large repeated records | decode    |        6,248 |   4,156 |                   1.50x |
+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,238 |           444,428 |         1.04x | 470,631 |
+| nested payload         |            235,971 |           223,358 |         1.06x | 209,940 |
+| collections            |             34,632 |            34,224 |         1.01x |  22,340 |
+| large repeated records |              6,196 |             6,094 |         1.02x |   3,514 |
+
+| Case                   | SchemaBinary decode | Msgpack decode | SchemaBinary vs Msgpack |
+| ---------------------- | ------------------: | -------------: | ----------------------: |
+| small record           |             485,041 |        471,327 |                   1.03x |
+| nested payload         |             223,300 |        203,587 |                   1.10x |
+| collections            |              37,881 |         21,442 |                   1.77x |
+| large repeated records |               6,129 |          4,066 |                   1.51x |
 
 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 row is the one case where msgpackr stays ahead. It returns a
-view into a reused 8 KiB buffer rather than allocating per call, so it skips
-the output copy that `SchemaBinary` pays on every encode. `SchemaBinary`
-returns a freshly allocated `Uint8Array` that the caller owns outright, which
-costs roughly 250 ns and dominates a 72-byte payload.
+Removing the output copy matters most for the 72-byte small record, where the
+arena was 4% faster than the ownership-copy control and nearly closed the gap
+with msgpackr. The other three cases were 1-6% faster than the copy control;
+none showed a material regression. 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
 

packages/effect/benchmark/schema/SchemaBinary.ts

@@ -182,6 +182,7 @@ const prepare = <S extends Schema.ConstraintCodec<unknown, unknown>>(
   const msgpackDecode = Schema.decodeUnknownSync(msgpackCodec)
 
   const binary = binaryEncode(value)
+  const binaryCopy = binary.slice()
   const json = jsonEncode(value)
   const msgpack = msgpackEncode(value)
 
@@ -210,11 +211,17 @@ const prepare = <S extends Schema.ConstraintCodec<unknown, unknown>>(
   return {
     formats: [
       {
-        name: "SchemaBinary",
+        name: "SchemaBinary arena",
         encodedSize: binary.length,
         encode: () => binaryEncode(value),
         decode: () => binaryDecode(binary)
       },
+      {
+        name: "SchemaBinary copy",
+        encodedSize: binaryCopy.length,
+        encode: () => binaryEncode(value).slice(),
+        decode: () => binaryDecode(binaryCopy)
+      },
       {
         name: "JSON",
         encodedSize: textEncoder.encode(json).length,

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

@@ -170,46 +170,85 @@ function decodeUtf8(
   }
 }
 
+const OUTPUT_ARENA_SIZE = 8 * 1024
+
+interface OutputArena {
+  readonly buf: Uint8Array<ArrayBuffer>
+  offset: number
+  writing: boolean
+}
+
+function makeOutputArena(size: number): OutputArena {
+  return { buf: new Uint8Array(size), offset: 0, writing: false }
+}
+
+let outputArena = makeOutputArena(OUTPUT_ARENA_SIZE)
+
 class Writer {
-  buf = new Uint8Array(1024)
+  buf: Uint8Array<ArrayBuffer> = new Uint8Array(0)
   view = new DataView(this.buf.buffer)
+  arena: OutputArena | undefined
+  start = 0
   len = 0
+  reset() {
+    let arena = outputArena
+    // A nested SchemaBinary codec can start while the outer writer still owns
+    // the current arena tail. Move the nested writer to a fresh arena rather
+    // than reserving a guessed range that the outer writer could outgrow.
+    if (arena.writing || arena.offset >= arena.buf.length) {
+      arena = outputArena = makeOutputArena(OUTPUT_ARENA_SIZE)
+    }
+    arena.writing = true
+    this.arena = arena
+    this.buf = arena.buf
+    this.view = new DataView(arena.buf.buffer)
+    this.start = arena.offset
+    this.len = 0
+  }
   ensure(n: number) {
-    if (this.len + n > this.buf.length) {
-      const next = new Uint8Array(Math.max(this.buf.length * 2, this.len + n))
-      next.set(this.buf)
-      this.buf = next
-      this.view = new DataView(next.buffer)
+    if (this.start + this.len + n > this.buf.length) {
+      const previous = this.arena!
+      const required = this.len + n
+      let size = OUTPUT_ARENA_SIZE
+      while (size < required) size *= 2
+      const next = makeOutputArena(size)
+      next.writing = true
+      next.buf.set(this.buf.subarray(this.start, this.start + this.len))
+      previous.writing = false
+      this.arena = outputArena = next
+      this.buf = next.buf
+      this.view = new DataView(next.buf.buffer)
+      this.start = 0
     }
   }
   byte(b: number) {
     this.ensure(1)
-    this.buf[this.len++] = b
+    this.buf[this.start + this.len++] = b
   }
   bytes(b: Uint8Array) {
     this.ensure(b.length)
-    this.buf.set(b, this.len)
+    this.buf.set(b, this.start + this.len)
     this.len += b.length
   }
   uvarint(n: number) {
     this.ensure(10)
     const buf = this.buf
-    let p = this.len
+    let p = this.start + this.len
     while (n > 0x7F) {
       buf[p++] = (n & 0x7F) | 0x80
       n = n < 0x80000000 ? n >>> 7 : Math.floor(n / 128)
     }
     buf[p++] = n
-    this.len = p
+    this.len = p - this.start
   }
   // Field ids are fixed by the layout, so their varint bytes are encoded once
   // at compile time and blitted here.
   raw(bytes: Uint8Array) {
     this.ensure(bytes.length)
     const buf = this.buf
-    let p = this.len
+    let p = this.start + this.len
     for (let i = 0; i < bytes.length; i++) buf[p++] = bytes[i]
-    this.len = p
+    this.len = p - this.start
   }
   uvarintBig(n: bigint) {
     while (n > BIGINT_VARINT_MASK) {
@@ -223,22 +262,22 @@ class Writer {
   }
   f64(n: number) {
     this.ensure(8)
-    this.view.setFloat64(this.len, n, true)
+    this.view.setFloat64(this.start + this.len, n, true)
     this.len += 8
   }
   i64(n: bigint) {
     this.ensure(8)
-    this.view.setBigInt64(this.len, n, true)
+    this.view.setBigInt64(this.start + this.len, n, true)
     this.len += 8
   }
   u32le(n: number) {
     this.ensure(4)
-    this.view.setUint32(this.len, n >>> 0, true)
+    this.view.setUint32(this.start + this.len, n >>> 0, true)
     this.len += 4
   }
   i32le(n: number) {
     this.ensure(4)
-    this.view.setInt32(this.len, n | 0, true)
+    this.view.setInt32(this.start + this.len, n | 0, true)
     this.len += 4
   }
   // Reserves one byte for a length prefix that is not known yet. The payload is
@@ -251,16 +290,17 @@ class Writer {
   endSized(mark: number) {
     const payload = this.len - mark - 1
     if (payload < 0x80) {
-      this.buf[mark] = payload
+      this.buf[this.start + mark] = payload
       return
     }
     const size = uvarintSize(payload)
     const extra = size - 1
     this.ensure(extra)
-    this.buf.copyWithin(mark + size, mark + 1, this.len)
+    const absoluteMark = this.start + mark
+    this.buf.copyWithin(absoluteMark + size, absoluteMark + 1, this.start + this.len)
     this.len += extra
     let n = payload
-    let p = mark
+    let p = absoluteMark
     while (n > 0x7F) {
       this.buf[p++] = (n & 0x7F) | 0x80
       n = Math.floor(n / 128)
@@ -272,19 +312,31 @@ class Writer {
     if (n === 0) return
     this.ensure(n * 3)
     const buf = this.buf
-    let p = this.len
+    let p = this.start + this.len
     for (let i = 0; i < n; i++) {
       const c = s.charCodeAt(i)
       if (c > 0x7F) {
-        this.len += utf8Encode.encodeInto(s, buf.subarray(this.len)).written
+        this.len += utf8Encode.encodeInto(s, buf.subarray(this.start + this.len)).written
         return
       }
       buf[p++] = c
     }
-    this.len = p
+    this.len = p - this.start
   }
   out(): Uint8Array<ArrayBuffer> {
-    return this.buf.slice(0, this.len) as Uint8Array<ArrayBuffer>
+    const arena = this.arena!
+    const out = new Uint8Array(arena.buf.buffer, this.start, this.len)
+    if (arena === outputArena) arena.offset = this.start + this.len
+    arena.writing = false
+    this.arena = undefined
+    return out
+  }
+  abort() {
+    const arena = this.arena
+    if (arena === undefined) return
+    if (arena === outputArena) arena.offset = this.start
+    arena.writing = false
+    this.arena = undefined
   }
 }
 
@@ -1621,7 +1673,7 @@ function encodeFrame(layout: Layout, value: unknown, options: SchemaAST.ParseOpt
   const w = pooledWriter ?? new Writer()
   const pooled = w === pooledWriter
   if (pooled) pooledWriter = undefined
-  w.len = 0
+  w.reset()
   const savedPathLen = issuePathLen
   issuePathLen = 0
   try {
@@ -1631,6 +1683,7 @@ function encodeFrame(layout: Layout, value: unknown, options: SchemaAST.ParseOpt
     w.endSized(mark)
     return w.out()
   } finally {
+    w.abort()
     issuePathLen = savedPathLen
     if (pooled) pooledWriter = w
   }
@@ -2115,6 +2168,12 @@ export interface toCodec<S extends Schema.Constraint> extends
  * One-shot encode/decode reuse the existing runners: exactly one frame,
  * leftover bytes are malformed. Use {@link parser} for concatenated frames.
  *
+ * Encoded results are stable views into a bump-allocated arena. A result's
+ * `byteLength` covers exactly one frame, but its `.buffer` may be larger,
+ * `byteOffset` may be non-zero, and unrelated encoded results may share the
+ * same buffer. Transferring or detaching that buffer affects every view into
+ * it. Use `bytes.slice()` when an independently owned buffer is required.
+ *
  * **Example**
  *
  * ```ts

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

@@ -120,6 +120,65 @@ describe("SchemaBinary", () => {
     })
   })
 
+  describe("output arena", () => {
+    it("keeps an earlier result stable through later encodes and arena rollover", () => {
+      const codec = SchemaBinary.toCodec(Schema.String)
+      const encode = Schema.encodeUnknownSync(codec)
+      const first = encode("first")
+      const expected = first.slice()
+      let rolledOver = false
+
+      for (let i = 0; i < 500; i++) {
+        const later = encode(`later-${i}-${"x".repeat(32)}`)
+        if (later.buffer !== first.buffer) rolledOver = true
+        assert.deepStrictEqual(first, expected)
+      }
+
+      assert.isTrue(rolledOver)
+      assert.strictEqual(Schema.decodeUnknownSync(codec)(first), "first")
+    })
+
+    it("keeps different results independent inside a shared arena", () => {
+      const codec = SchemaBinary.toCodec(Schema.String)
+      const encode = Schema.encodeUnknownSync(codec)
+      const decode = Schema.decodeUnknownSync(codec)
+      const results = [encode("alpha"), encode("bravo"), encode("charlie")]
+
+      assert.deepStrictEqual(results.map((bytes) => decode(bytes)), ["alpha", "bravo", "charlie"])
+      const shared = results.flatMap((left, index) => results.slice(index + 1).map((right) => [left, right] as const))
+        .find(([left, right]) => left.buffer === right.buffer)
+      assert.isDefined(shared)
+      assert.notStrictEqual(shared![0].byteOffset, shared![1].byteOffset)
+    })
+
+    it("preserves nested two-phase codec composition", () => {
+      const Inner = Schema.Struct({ id: Schema.Number, label: Schema.String })
+      const Outer = Schema.Struct({ id: Schema.String, inner: SchemaBinary.toCodec(Inner) })
+      const codec = SchemaBinary.toCodec(Outer)
+      const value = { id: "outer", inner: { id: 1, label: "inner" } }
+
+      assert.deepStrictEqual(Schema.decodeUnknownSync(codec)(Schema.encodeUnknownSync(codec)(value)), value)
+    })
+
+    it.effect("keeps concurrent fiber results readable after their producing turn", () => {
+      const Value = Schema.Struct({ id: Schema.Number, value: Schema.String })
+      const codec = SchemaBinary.toCodec(Value)
+      const encode = Schema.encodeUnknownEffect(codec)
+      const decode = Schema.decodeUnknownSync(codec)
+      const values = Array.from({ length: 100 }, (_, id) => ({ id, value: `value-${id}` }))
+
+      return Effect.gen(function*() {
+        const encoded = yield* Effect.forEach(
+          values,
+          (value) => encode(value).pipe(Effect.flatMap((bytes) => Effect.yieldNow.pipe(Effect.as(bytes)))),
+          { concurrency: "unbounded" }
+        )
+        yield* Effect.yieldNow
+        assert.deepStrictEqual(encoded.map((bytes) => decode(bytes)), values)
+      })
+    })
+  })
+
   describe("schema evolution", () => {
     it("skips unknown fields, accepts field reorder, and leaves missing optionals absent", () => {
       const Writer = Schema.Struct({ a: Schema.Number, extra: Schema.String, b: Schema.String })