perf(hotpath): cut per-dispatch and per-batch overhead on the stdio path

Targets the vgi-benchmarks subprocess path (arrow-js + stdio) plus the
flechette backend. Backend-agnostic and both-backend-specific wins:

- server.ts: skip the per-dispatch request-batch serializeBatch (access-log
  IPC re-encode) unless a dispatchHook is installed — the default path saves
  ~17us (arrow-js) / ~10us (flechette) per call. Replace the per-request
  getMethods() map copy with Protocol.getMethod()/methodNames().
- arrow-js conformBatchToSchema: add a `mutated` flag + per-child identity
  skip so a already-conformant batch returns unchanged (1.81us -> 0.19us);
  drain() returns the single buffered chunk without copying.
- flechette deserializeBatch: drop the Proxy wrapper (defeated V8 inline
  caches across request parsing) for direct metadata/numRows assignment.
- flechette toFlechetteType: memoize by type-object identity (was re-deriving
  the nested type tree per batch).
- wire/writer.ts: fd target writes inline and returns a shared resolved
  promise, skipping the promise-chain + microtask hop per emitted batch.
- wire/response.ts coerceInt64: cache int64 field names per schema and skip
  the record clone when no coercion is needed.
- wire/request.ts: hoist BigInt safe-integer bounds to module constants.

Verified: 200 vgi-rpc unit tests + 235 vgi-typescript e2e tests pass on both
backends; typecheck + biome clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: rustyconover

src/arrow/impl-arrowjs/index.ts

@@ -165,6 +165,13 @@ export function createIncrementalEncoder(s: VgiSchema): IncrementalEncoder {
   writer.reset(undefined, s as unknown as A_Schema);
   const drain = (): Uint8Array => {
     const values = (writer as any)._sink._values as Uint8Array[];
+    // Common case: the encoder buffered exactly one chunk for this batch —
+    // hand it straight through instead of copying it into a fresh buffer.
+    if (values.length === 1) {
+      const only = values[0];
+      values.length = 0;
+      return only;
+    }
     const total = values.reduce((n, c) => n + c.length, 0);
     const out = new Uint8Array(total);
     let off = 0;
@@ -401,15 +408,25 @@ export function conformBatchToSchema(batch: VgiBatch, schema: VgiSchema): VgiBat
     }
   }
 
+  let mutated = false;
   const children = s.fields.map((f, i) => {
     const srcChild = a.data.children[i];
     const srcType = srcChild.type;
     const dstType = f.type;
 
+    // Already the exact schema type object (batches this process built from
+    // `schema` hit this for every column) — no clone, no rebuild needed.
+    if (srcType === dstType) return srcChild;
+
     if (!_needsValueCast(srcType, dstType)) {
+      // Same logical type but a different type object (e.g. a generic `Float`
+      // from a foreign IPC round-trip vs the schema's `Float64`). Swap the
+      // type via clone so the writer doesn't silently drop the batch.
+      mutated = true;
       return srcChild.clone(dstType);
     }
     if (_isNumeric(srcType) && _isNumeric(dstType)) {
+      mutated = true;
       const col = a.getChildAt(i)!;
       const values: number[] = [];
       for (let r = 0; r < a.numRows; r++) {
@@ -418,9 +435,14 @@ export function conformBatchToSchema(batch: VgiBatch, schema: VgiSchema): VgiBat
       }
       return a_vectorFromArray(values, dstType).data[0];
     }
+    mutated = true;
     return srcChild.clone(dstType);
   });
 
+  // Every column already carried the schema's exact type — the batch is
+  // conformant as-is, so avoid the struct/makeData/RecordBatch rebuild.
+  if (!mutated) return batch;
+
   const structType = new A_Struct(s.fields);
   const data = a_makeData({
     type: structType,

src/arrow/impl-flechette/index.ts

@@ -157,13 +157,17 @@ export function deserializeBatch(bytes: Uint8Array): VgiBatch {
   const meta = needRowsFallback ? readFirstRecordBatchMeta(bytes) : null;
   const wantRows = meta !== null && meta.numRows > 0;
   if (!wantRows && !wantMeta) return table as VgiBatch;
-  return new Proxy(table, {
-    get(target, prop, receiver) {
-      if (wantRows && prop === "numRows") return meta!.numRows;
-      if (wantMeta && prop === "metadata") return recordMd;
-      return Reflect.get(target, prop, receiver);
-    },
-  }) as unknown as VgiBatch;
+  // Assign the two overrides directly rather than wrapping in a Proxy: a Proxy
+  // routes every downstream property access (`.schema`, `.getChildAt(i)`, …) on
+  // the request batch through a trap + Reflect.get, defeating V8's inline
+  // caches across the entire request-parsing loop. `metadata` is non-shadowing
+  // on flechette's Table (same as attachBatchMetadata); `numRows` is a real
+  // property, so override it via defineProperty.
+  if (wantMeta) table.metadata = recordMd;
+  if (wantRows) {
+    Object.defineProperty(table, "numRows", { value: meta!.numRows, configurable: true, enumerable: true });
+  }
+  return table as VgiBatch;
 }
 
 // ----- Construction --------------------------------------------------------

src/arrow/impl-flechette/normalize-type.ts

@@ -44,13 +44,27 @@ function fField(f: any): any {
   return f_field(f.name, toFlechetteType(f.type), f.nullable ?? true, f.metadata ?? null);
 }
 
+// Memoize by input-type object identity. Type objects come from registered
+// schemas and are immutable, and the flechette-native result is itself an
+// immutable value type used read-only — so caching lets every batch reuse one
+// rebuilt type (and its nested children) instead of re-deriving the tree.
+const _flechetteTypeCache = new WeakMap<object, any>();
+
 /**
  * Rebuild `type` as a flechette-native DataType. Idempotent for types that are
  * already native (reconstructed from the same structural props). Unknown
  * typeIds pass through unchanged as a safety net.
  */
 export function toFlechetteType(type: any): any {
   if (type == null) return nullType();
+  const cached = _flechetteTypeCache.get(type);
+  if (cached !== undefined) return cached;
+  const built = _buildFlechetteType(type);
+  _flechetteTypeCache.set(type, built);
+  return built;
+}
+
+function _buildFlechetteType(type: any): any {
   switch (type.typeId) {
     case Type.Null:
       return nullType();

src/protocol.ts

@@ -168,4 +168,15 @@ export class Protocol {
   getMethods(): Map<string, MethodDefinition> {
     return new Map(this._methods);
   }
+
+  /** Look up a single method without copying the whole map — for the
+   *  per-request dispatch path. Callers must not mutate the returned value. */
+  getMethod(name: string): MethodDefinition | undefined {
+    return this._methods.get(name);
+  }
+
+  /** Registered method names, sorted — for diagnostics/error messages only. */
+  methodNames(): string[] {
+    return [...this._methods.keys()].sort();
+  }
 }

src/server.ts

@@ -275,10 +275,9 @@ export class VgiRpcServer {
     }
 
     // Look up method
-    const methods = this.protocol.getMethods();
-    const method = methods.get(methodName);
+    const method = this.protocol.getMethod(methodName);
     if (!method) {
-      const available = [...methods.keys()].sort();
+      const available = this.protocol.methodNames();
       const err = new MethodNotImplementedError(
         `Unknown method: '${methodName}'. Available methods: [${available.join(", ")}]`,
       );
@@ -307,11 +306,15 @@ export class VgiRpcServer {
     const methodType = method.type === MethodType.UNARY ? "unary" : "stream";
 
     // Capture self-contained IPC bytes of the request batch for the access log.
+    // Only pay the full IPC re-encode when a hook will actually read them — the
+    // default (no dispatchHook) path skips it entirely.
     let requestData: Uint8Array | undefined;
-    try {
-      requestData = serializeBatch(batch as any);
-    } catch {
-      // best-effort; observability must not fail dispatch
+    if (this.dispatchHook) {
+      try {
+        requestData = serializeBatch(batch as any);
+      } catch {
+        // best-effort; observability must not fail dispatch
+      }
     }
 
     let streamId: string | undefined;

src/wire/request.ts

@@ -6,6 +6,11 @@ import { REQUEST_ID_KEY, REQUEST_VERSION, REQUEST_VERSION_KEY, RPC_METHOD_KEY }
 import { RpcError, VersionError } from "../errors.js";
 import { isOpaquePassthroughType } from "./opaque.js";
 
+// Safe-integer bounds as BigInt, hoisted so per-field parsing doesn't
+// re-allocate them on every bigint-valued parameter of every request.
+const MIN_SAFE_BIG = BigInt(Number.MIN_SAFE_INTEGER);
+const MAX_SAFE_BIG = BigInt(Number.MAX_SAFE_INTEGER);
+
 export interface ParsedRequest {
   methodName: string;
   requestVersion: string;
@@ -108,7 +113,7 @@ export function parseRequest(schema: VgiSchema, batch: VgiBatch): ParsedRequest
     // make the builder apply a `*scale` multiplication (Decimal) or
     // `* 10^unit` (Time/Timestamp), corrupting the value.
     if (typeof value === "bigint" && !isOpaquePassthroughType(field.type)) {
-      if (value >= BigInt(Number.MIN_SAFE_INTEGER) && value <= BigInt(Number.MAX_SAFE_INTEGER)) {
+      if (value >= MIN_SAFE_BIG && value <= MAX_SAFE_BIG) {
         value = Number(value);
       }
     }

src/wire/response.ts

@@ -17,24 +17,57 @@ import {
   SERVER_ID_KEY,
 } from "../constants.js";
 
+/**
+ * Names of the Int64 fields in a schema, computed once per schema object.
+ * `VgiSchema` is treated as immutable everywhere, so identity-keying is safe.
+ */
+const _int64FieldsCache = new WeakMap<VgiSchema, readonly string[]>();
+function int64FieldNames(schema: VgiSchema): readonly string[] {
+  let names = _int64FieldsCache.get(schema);
+  if (names === undefined) {
+    const out: string[] = [];
+    for (const f of schema.fields) {
+      if (isInt(f.type) && (f.type as any).bitWidth === 64) out.push(f.name);
+    }
+    names = out;
+    _int64FieldsCache.set(schema, names);
+  }
+  return names;
+}
+
 /**
  * Coerce values for Int64 schema fields from Number to BigInt.
- * Handles both single values and arrays. Returns a new record with coerced values.
+ * Handles both single values and arrays. Returns a new record with coerced
+ * values, or the original record untouched when no coercion is needed.
  */
 export function coerceInt64(schema: VgiSchema, values: Record<string, any>): Record<string, any> {
-  const result: Record<string, any> = { ...values };
-  for (const f of schema.fields) {
-    const val = result[f.name];
+  const int64Fields = int64FieldNames(schema);
+  if (int64Fields.length === 0) return values;
+
+  let result: Record<string, any> | null = null;
+  for (const name of int64Fields) {
+    const val = values[name];
     if (val === undefined) continue;
-    if (!isInt(f.type) || (f.type as any).bitWidth !== 64) continue;
 
     if (Array.isArray(val)) {
-      result[f.name] = val.map((v: any) => (typeof v === "number" ? BigInt(v) : v));
+      // Clone lazily and only map when a Number element is actually present.
+      let mapped: any[] | null = null;
+      for (let i = 0; i < val.length; i++) {
+        if (typeof val[i] === "number") {
+          if (mapped === null) mapped = val.slice();
+          mapped[i] = BigInt(val[i]);
+        }
+      }
+      if (mapped !== null) {
+        result ??= { ...values };
+        result[name] = mapped;
+      }
     } else if (typeof val === "number") {
-      result[f.name] = BigInt(val);
+      result ??= { ...values };
+      result[name] = BigInt(val);
     }
   }
-  return result;
+  return result ?? values;
 }
 
 /**

src/wire/writer.ts

@@ -7,6 +7,10 @@ import { createIncrementalEncoder, serializeBatches } from "../arrow/index.js";
 
 const STDOUT_FD = 1;
 
+// Shared resolved promise returned by the synchronous fd write path so we
+// don't allocate a fresh Promise per emitted batch.
+const RESOLVED: Promise<void> = Promise.resolve();
+
 // Resolve node:fs via indirect-string require so esbuild/wrangler don't
 // statically pull node:fs into the bundle. Workers (Cloudflare workerd) never
 // instantiate IpcStreamWriter (no stdio transport on workers), so the
@@ -238,15 +242,20 @@ export class IncrementalStream {
   }
 
   private enqueue(bytes: Uint8Array): Promise<void> {
+    // Fast path: the fd target's write is fully synchronous and a stream's
+    // target kind is fixed for its lifetime, so nothing is ever async-in-flight
+    // ahead of us. Write inline and skip the promise-chain + microtask hop —
+    // this runs once per emitted batch on the stdio producer path.
+    const target = this.target;
+    if (target.kind === "fd") {
+      writeAll(target.fd, bytes);
+      return RESOLVED;
+    }
     const next = this.writeChain.then(() => {
-      if (this.target.kind === "fd") {
-        writeAll(this.target.fd, bytes);
-        return;
-      }
-      if (this.target.kind === "sink") {
-        return this.target.sink.write(bytes);
+      if (target.kind === "sink") {
+        return target.sink.write(bytes);
       }
-      return socketWriteAll(this.target.socket, bytes);
+      return socketWriteAll(target.socket, bytes);
     });
     // Swallow rejections on the chain so a single failure doesn't poison
     // every subsequent enqueue with an unhandled rejection. The returned