fix(wire): spin-retry on stdio EAGAIN instead of 1ms sleep; flechette incremental encoder

Two fixes to the stdio (subprocess) transport:

1. writer: the non-blocking-pipe EAGAIN branch slept 1ms (Atomics.wait) per
   backpressure event. The engine drains the pipe in microseconds in lockstep,
   so this slept ~1000x too long and *dominated* large-batch write time — a
   4KB-row passthru spent ~13s of a 15s run asleep here. Spin-retry instead
   (safe: the stdio path has a single peer, nothing to starve), with a bounded
   fallback sleep so a hung reader can't pin a core. ~9-12x on wire-bound
   TS workers; per-byte wire cost 20 -> 1.6 ns/byte.

2. flechette: implement createIncrementalEncoder (previously threw), unblocking
   the flechette Arrow backend on the stdio transport. It carves per-message
   frames (schema / dictionary / record batch) out of flechette's one-shot
   stream encoder via a FlatBuffer message walk (splitIpcMessages), deferring
   the schema to the first batch so dictionary ids stay consistent. Round-trips
   verified for single-batch, multi-batch, and dictionary-encoded columns.

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

Author: rustyconover

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@query-farm/vgi-rpc",
-  "version": "0.14.0",
+  "version": "0.14.1",
   "license": "Apache-2.0",
   "homepage": "https://vgi-rpc-typescript.query.farm",
   "repository": {

src/arrow/impl-flechette/index.ts

@@ -46,7 +46,7 @@ import type {
   VgiField,
   VgiSchema,
 } from "../types.js";
-import { readFirstRecordBatchMeta } from "./message-meta.js";
+import { readFirstRecordBatchMeta, splitIpcMessages } from "./message-meta.js";
 import { toFlechetteType } from "./normalize-type.js";
 
 export const backend: VgiBackendInfo = { name: "flechette", opaquePassthrough: false };
@@ -397,18 +397,61 @@ export function conformBatchToSchema(batch: VgiBatch, schema: VgiSchema): VgiBat
   return rebuilt as unknown as VgiBatch;
 }
 
+// End-of-stream marker: continuation (0xFFFFFFFF) + zero metadata length.
+const IPC_EOS = new Uint8Array(new Int32Array([-1, 0]).buffer);
+
 /**
- * Not supported on flechette: the lockstep stdio exchange protocol needs an
- * incremental IPC writer (emit each batch's framing bytes before reading
- * the next input), and flechette only exposes a one-shot table encoder.
- * The flechette backend is selected for workerd / browser / worker, which
- * are HTTP-only (no stdio), so this factory is never reached there. The
- * stdio server requires the arrow-js backend.
+ * Incremental IPC encoder for the flechette backend.
+ *
+ * flechette only exposes a one-shot stream encoder (`tableToIPC(x, { format:
+ * "stream" })` emits a complete `[schema][dict…][recordbatch][EOS]` stream), but
+ * the lockstep stdio protocol needs the messages emitted piecemeal: the schema
+ * preamble once, then each batch's `[dict…][recordbatch]` body, then a single
+ * EOS. We reconstruct that by serializing each part as a full stream and slicing
+ * out the message frames we want (via `splitIpcMessages`, which parses the Arrow
+ * Message envelopes rather than assuming byte lengths):
+ *
+ *   start()      → nothing; the schema is carried by the first batch (below).
+ *   writeBatch() → the first call emits `[schema][dict…][recordbatch]`; later
+ *                  calls emit `[dict…][recordbatch]` (schema dropped).
+ *   finish()     → the shared EOS marker (or a schema-only stream when no batch
+ *                  was written, so an empty result still yields a valid stream).
+ *
+ * Why the schema is deferred to the first batch rather than serialized up front:
+ * flechette assigns dictionary ids per serialization, and `serializeSchema`'s
+ * empty-column path declares dictionary fields differently from a populated
+ * batch — which desyncs the reader for dictionary-encoded columns. Two populated
+ * batches of the same schema *do* share a byte-identical schema message, so
+ * taking the schema from the first real batch keeps it consistent with every
+ * batch's dictionary ids. Emitting each batch's dictionaries in full is a valid
+ * Arrow *stream* dictionary replacement (isDelta=false), which the reader
+ * accepts, so dictionary columns round-trip correctly.
  */
-export function createIncrementalEncoder(_s: VgiSchema): IncrementalEncoder {
-  throw new Error(
-    "Incremental IPC streaming (stdio producer/exchange) is not supported on the flechette " +
-      "Arrow backend. Use the arrow-js backend for the stdio transport, or the HTTP transport " +
-      "on workerd/browser builds.",
-  );
+export function createIncrementalEncoder(s: VgiSchema): IncrementalEncoder {
+  let schemaEmitted = false;
+  return {
+    start: () => new Uint8Array(0),
+    writeBatch(batch: VgiBatch): Uint8Array {
+      const full = serializeBatch(batch); // [schema][dict…][recordbatch][EOS]
+      const spans = splitIpcMessages(full); // excludes the trailing EOS
+      const bodyEnd = spans.length ? spans[spans.length - 1].frameEnd : full.length;
+      if (!schemaEmitted) {
+        schemaEmitted = true;
+        // First batch carries the schema: [schema][dict…][recordbatch].
+        return full.subarray(0, bodyEnd);
+      }
+      // Later batches: drop the leading Schema message, keep [dict…][recordbatch].
+      let start = -1;
+      for (const m of spans) {
+        if (m.headerType !== 1) {
+          start = m.frameStart;
+          break;
+        }
+      }
+      return start < 0 ? new Uint8Array(0) : full.subarray(start, bodyEnd);
+    },
+    // A result with no batches still needs a schema on the wire; serializeSchema
+    // produces a complete schema-only stream ([schema][EOS]).
+    finish: () => (schemaEmitted ? IPC_EOS : serializeSchema(s)),
+  };
 }

src/arrow/impl-flechette/message-meta.ts

@@ -172,3 +172,48 @@ export function readFirstRecordBatchMeta(
   }
   return null;
 }
+
+/** Byte span of one on-wire IPC message (framing + metadata + body). */
+export interface IpcMessageSpan {
+  /** Header type — 1 Schema, 2 DictionaryBatch, 3 RecordBatch. */
+  headerType: number;
+  /** Offset of the message's continuation marker in the stream. */
+  frameStart: number;
+  /** Offset just past the message's (padded) body. */
+  frameEnd: number;
+}
+
+/**
+ * Split an Arrow IPC *stream* into its constituent message frames, stopping at
+ * the end-of-stream marker. Each span covers the whole on-wire message
+ * (continuation marker + metadata length + padded metadata + padded body).
+ *
+ * Used by the flechette incremental encoder to carve a one-shot
+ * `tableToIPC(..., { format: "stream" })` output — a complete
+ * `[schema][dict…][recordbatch][EOS]` stream — into the schema preamble and the
+ * per-batch `[dict…][recordbatch]` body that the lockstep stdio protocol emits
+ * separately.
+ */
+export function splitIpcMessages(stream: Uint8Array): IpcMessageSpan[] {
+  const spans: IpcMessageSpan[] = [];
+  let pos = 0;
+  while (pos + 4 <= stream.length) {
+    const frameStart = pos;
+    let metaLen = readI32LE(stream, pos);
+    pos += 4;
+    if (metaLen === -1) {
+      // Continuation-marker form (post-Arrow 0.15): the real length follows.
+      if (pos + 4 > stream.length) break;
+      metaLen = readI32LE(stream, pos);
+      pos += 4;
+    }
+    if (metaLen === 0) break; // EOS
+    if (pos + metaLen > stream.length) break;
+    const head = stream.subarray(pos, pos + metaLen);
+    pos += metaLen;
+    const parsed = readMessageEnvelope(head);
+    pos += parsed.bodyLength; // bodyLength is the padded on-wire body length
+    spans.push({ headerType: parsed.headerType, frameStart, frameEnd: pos });
+  }
+  return spans;
+}

src/wire/writer.ts

@@ -46,17 +46,27 @@ function _loadWriteSync(): (fd: number, data: Uint8Array, offset?: number, len?:
 function writeAll(fd: number, data: Uint8Array): void {
   const writeSync = _loadWriteSync();
   let offset = 0;
+  let spins = 0;
   while (offset < data.length) {
     try {
       const written = writeSync(fd, data, offset, data.length - offset);
       if (written <= 0) throw new Error(`writeSync returned ${written}`);
       offset += written;
+      spins = 0;
     } catch (e: any) {
       if (e.code === "EAGAIN") {
-        // A non-blocking pipe — unexpected for stdio, but handle defensively.
-        // Yielding is not possible from a synchronous function; busy-wait
-        // briefly. AF_UNIX sockets must NOT use this path.
+        // Non-blocking pipe backpressure. The stdio path has a single peer (the
+        // engine), which drains the pipe within microseconds in lockstep, so
+        // spin-retry rather than sleep: a 1ms `Atomics.wait` here is ~1000x too
+        // coarse and *dominated* large-batch write time — a 4KB-row passthru
+        // spent ~13s of a 15s run asleep in this branch, i.e. the whole TS
+        // "wire tax" was this sleep. Spinning is safe here because there are no
+        // other connections to starve (unlike the AF_UNIX socketWriteAll path,
+        // which correctly awaits 'drain'). Fall back to a brief sleep only if
+        // the peer looks genuinely stalled, so a hung reader can't pin a core.
+        if (++spins < 8192) continue;
         Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 1);
+        spins = 0;
         continue;
       }
       throw e;
@@ -192,12 +202,10 @@ export class IpcStreamWriter {
  *
  * The encoder is obtained from the Arrow facade, so this file no longer
  * imports arrow-js directly — keeping arrow-js out of the flechette
- * (workerd/browser) bundle. The flechette encoder throws on construction;
- * the stdio exchange protocol is lockstep (the client reads each response
- * batch before sending the next input) which needs an incremental writer
- * flechette doesn't provide. workerd/browser deployments use HTTP (no
- * stdio), so the flechette path is never reached there; `flechette-pipe`
- * conformance is xfailed for streams.
+ * (workerd/browser) bundle. Both backends now provide an incremental encoder
+ * (the flechette one carves per-message frames out of its one-shot stream
+ * encoder), so the stdio lockstep exchange works on flechette as well as
+ * arrow-js.
  */
 export class IncrementalStream {
   private readonly encoder: IncrementalEncoder;