fix: add WasmEngine.resetHeap() wrapper with safety invariant docs

The Zig WASM engine uses a bump allocator — resetHeap() invalidates ALL
prior allocations. All call sites now go through WasmEngine.resetHeap()
instead of reaching into .exports.resetHeap() directly. This creates a
single choke point for future guards and documents the 5 safety invariants
that prevent heap corruption (single-threaded DOs, no await between
reset and consumption, .slice() before next reset, per-DO ownership,
decompress methods copy results out).

Verified all 14 call sites across operators.ts, query-do.ts, and
fragment-do.ts — every path is synchronous between reset and consumption.

Author: teamchong

src/fragment-do.test.ts

@@ -17,6 +17,7 @@ vi.mock("./wasm-engine.js", () => ({
       clearTables: () => {},
     },
     reset: () => {},
+    resetHeap: () => {},
     clearTable: () => {},
     registerColumn: () => true,
     registerColumns: () => true,

src/fragment-do.ts

@@ -183,7 +183,7 @@ export class FragmentDO extends DurableObject<Env> {
       // Zero-copy WASM path: register raw page data per-column, execute SQL in WASM
       const wasmStart = Date.now();
       const fragTable = `__frag_${r2Key}`;
-      this.wasmEngine.exports.resetHeap();
+      this.wasmEngine.resetHeap();
       const fragColEntries = cols
         .filter(col => columnData.get(col.name)?.length)
         .map(col => ({

src/operators.ts

@@ -559,7 +559,7 @@ function wasmFilterNumeric(
   wasm: WasmEngine,
 ): Uint32Array | null {
   try {
-    wasm.exports.resetHeap();
+    wasm.resetHeap();
 
     if (dtype === "float64" || dtype === "float32") {
       const dataPtr = wasm.exports.alloc(rowCount * 8);
@@ -620,7 +620,7 @@ function wasmFilterRange(
   rowCount: number, wasm: WasmEngine, negate = false,
 ): Uint32Array | null {
   try {
-    wasm.exports.resetHeap();
+    wasm.resetHeap();
 
     if (dtype === "float64" || dtype === "float32") {
       const dataPtr = wasm.exports.alloc(rowCount * 8);
@@ -677,7 +677,7 @@ function wasmFilterLike(
   wasm: WasmEngine,
 ): Uint32Array | null {
   try {
-    wasm.exports.resetHeap();
+    wasm.resetHeap();
     const encoder = _textEncoder;
 
     // Build offsets array and packed string data
@@ -730,7 +730,7 @@ function wasmFilterLike(
 /** Intersect two sorted index arrays using WASM. */
 function wasmIntersect(a: Uint32Array, b: Uint32Array, wasm: WasmEngine): Uint32Array {
   try {
-    wasm.exports.resetHeap();
+    wasm.resetHeap();
     const aPtr = wasm.exports.alloc(a.byteLength);
     const bPtr = wasm.exports.alloc(b.byteLength);
     const outPtr = wasm.exports.alloc(Math.min(a.length, b.length) * 4);
@@ -758,7 +758,7 @@ function wasmIntersect(a: Uint32Array, b: Uint32Array, wasm: WasmEngine): Uint32
 /** Union two sorted index arrays using WASM. */
 function wasmUnion(a: Uint32Array, b: Uint32Array, wasm: WasmEngine): Uint32Array {
   try {
-    wasm.exports.resetHeap();
+    wasm.resetHeap();
     const aPtr = wasm.exports.alloc(a.byteLength);
     const bPtr = wasm.exports.alloc(b.byteLength);
     const outPtr = wasm.exports.alloc((a.length + b.length) * 4);
@@ -2015,7 +2015,7 @@ export class WasmAggregateOperator implements Operator {
         let matchCount = -1; // -1 = no filter, use full buffer
         let indicesPtr = 0;
         if (hasFilters) {
-          this.wasm.exports.resetHeap();
+          this.wasm.resetHeap();
 
           // Evaluate AND filters
           let currentIndices: Uint32Array | null = null;
@@ -2040,7 +2040,7 @@ export class WasmAggregateOperator implements Operator {
           if (!currentIndices || currentIndices.length === 0) continue;
 
           // Copy indices to WASM for indexed aggregates
-          this.wasm.exports.resetHeap();
+          this.wasm.resetHeap();
           indicesPtr = this.wasm.exports.alloc(currentIndices.byteLength);
           if (indicesPtr) {
             new Uint32Array(this.wasm.exports.memory.buffer, indicesPtr, currentIndices.length).set(currentIndices);
@@ -2142,7 +2142,7 @@ export class WasmAggregateOperator implements Operator {
       const elemSize = col.dtype === "int32" ? 4 : 8;
       const rowCount = buf.byteLength / elemSize;
 
-      this.wasm.exports.resetHeap();
+      this.wasm.resetHeap();
       const dataPtr = this.wasm.exports.alloc(buf.byteLength);
       if (!dataPtr) return new Uint32Array(0);
       new Uint8Array(this.wasm.exports.memory.buffer, dataPtr, buf.byteLength).set(new Uint8Array(buf));

src/query-do.ts

@@ -290,7 +290,7 @@ class EdgeScanOperator implements Operator {
     await this.fetchAllPages();
 
     const wasmStart = Date.now();
-    this.wasmEngine.exports.resetHeap();
+    this.wasmEngine.resetHeap();
     const fragTable = `__edge_${this.meta.r2Key}`;
     const colEntries = this.cols
       .filter(col => this.columnData.get(col.name)?.length)
@@ -358,7 +358,7 @@ class EdgeScanOperator implements Operator {
     for (const v of decodedColumns.values()) if (v.length > numRows) numRows = v.length;
     if (numRows === 0) return this.next(); // skip empty pages
 
-    this.wasmEngine.exports.resetHeap();
+    this.wasmEngine.resetHeap();
     const fragTable = `__edge_pq_${pi}`;
     const decodedEntries = cols
       .filter(col => decodedColumns.get(col.name)?.length)
@@ -885,7 +885,7 @@ export class QueryDO extends DurableObject<Env> {
     const r2ReadMs = Date.now() - r2Start;
 
     const wasmStart = Date.now();
-    this.wasmEngine.exports.resetHeap();
+    this.wasmEngine.resetHeap();
 
     // Load fragment and extract columns via fragment reader
     const dataPtr = this.wasmEngine.exports.alloc(fileData.byteLength);
@@ -1132,7 +1132,7 @@ export class QueryDO extends DurableObject<Env> {
 
       // Try WASM SQL path: register decoded columns → executeQuery (SIMD filter/sort/agg)
       const colNames = cols.map(c => c.name);
-      this.wasmEngine.exports.resetHeap();
+      this.wasmEngine.resetHeap();
       const decodedEntries = cols
         .filter(col => decodedColumns.get(col.name)?.length)
         .map(col => ({ name: col.name, dtype: col.dtype, values: decodedColumns.get(col.name)! }));
@@ -1176,7 +1176,7 @@ export class QueryDO extends DurableObject<Env> {
 
     // Lance path: zero-copy WASM registration + SQL execution
     const wasmStart = Date.now();
-    this.wasmEngine.exports.resetHeap();
+    this.wasmEngine.resetHeap();
     const lanceColEntries = cols
       .filter(col => columnData.get(col.name)?.length)
       .map(col => ({

src/wasm-engine.integration.test.ts

@@ -28,7 +28,7 @@ beforeAll(async () => {
 describe.skipIf(!hasWasm)("WASM integration", () => {
   describe("int64 + float64 columns → SQL → rows", () => {
     it("registers columns and executes SELECT *", () => {
-      wasm.exports.resetHeap();
+      wasm.resetHeap();
 
       const table = "test_nums";
       const rowCount = 3;
@@ -63,7 +63,7 @@ describe.skipIf(!hasWasm)("WASM integration", () => {
 
   describe("string column registration", () => {
     it("registers utf8 strings with length-prefixed encoding", () => {
-      wasm.exports.resetHeap();
+      wasm.resetHeap();
       const table = "test_strings";
 
       // Build length-prefixed string data: "hello" (5), "world" (5)
@@ -94,7 +94,7 @@ describe.skipIf(!hasWasm)("WASM integration", () => {
 
   describe("type promotion", () => {
     it("promotes int32 to int64 for WASM registration", () => {
-      wasm.exports.resetHeap();
+      wasm.resetHeap();
       const table = "test_promo";
 
       // int32 values
@@ -117,7 +117,7 @@ describe.skipIf(!hasWasm)("WASM integration", () => {
 
   describe("filter pushdown SQL", () => {
     it("filters with WHERE gt", () => {
-      wasm.exports.resetHeap();
+      wasm.resetHeap();
       const table = "test_filter";
 
       const i64Buf = new BigInt64Array([10n, 20n, 30n, 40n, 50n]);
@@ -139,7 +139,7 @@ describe.skipIf(!hasWasm)("WASM integration", () => {
     });
 
     it("filters with WHERE eq", () => {
-      wasm.exports.resetHeap();
+      wasm.resetHeap();
       const table = "test_eq";
 
       const f64Buf = new Float64Array([1.0, 2.0, 3.0, 2.0]);
@@ -160,7 +160,7 @@ describe.skipIf(!hasWasm)("WASM integration", () => {
     });
 
     it("filters with WHERE lt on float64", () => {
-      wasm.exports.resetHeap();
+      wasm.resetHeap();
       const table = "test_lt";
 
       const f64Buf = new Float64Array([1.0, 2.0, 3.0, 4.0]);
@@ -181,7 +181,7 @@ describe.skipIf(!hasWasm)("WASM integration", () => {
 
   describe("vector search", () => {
     it("finds top-K nearest vectors via vectorSearchBuffer", () => {
-      wasm.exports.resetHeap();
+      wasm.resetHeap();
       const dim = 4;
       const numVectors = 5;
 
@@ -231,7 +231,7 @@ describe.skipIf(!hasWasm)("WASM integration", () => {
 
   describe("LIMIT and ORDER BY", () => {
     it("applies LIMIT to results", () => {
-      wasm.exports.resetHeap();
+      wasm.resetHeap();
       const table = "test_limit";
 
       const i64Buf = new BigInt64Array([1n, 2n, 3n, 4n, 5n]);
@@ -252,7 +252,7 @@ describe.skipIf(!hasWasm)("WASM integration", () => {
 
   describe("multi-column SELECT with projections", () => {
     it("returns only projected columns", () => {
-      wasm.exports.resetHeap();
+      wasm.resetHeap();
       const table = "test_proj";
 
       const i64Buf = new BigInt64Array([1n, 2n, 3n]);
@@ -278,7 +278,7 @@ describe.skipIf(!hasWasm)("WASM integration", () => {
 
   describe("bool column", () => {
     it("registers and queries bool values", () => {
-      wasm.exports.resetHeap();
+      wasm.resetHeap();
       const table = "test_bool";
 
       // Bool bitmap: [true, false, true] = 0b101 = 0x05
@@ -301,7 +301,7 @@ describe.skipIf(!hasWasm)("WASM integration", () => {
 
   describe("aggregates", () => {
     it("computes SUM via WASM", () => {
-      wasm.exports.resetHeap();
+      wasm.resetHeap();
       const table = "test_sum";
 
       const f64Buf = new Float64Array([10.0, 20.0, 30.0]);

src/wasm-engine.ts

@@ -169,6 +169,22 @@ export async function instantiateWasm(wasmModule: WebAssembly.Module, opts?: { o
   return engine;
 }
 
+/**
+ * WASM Heap Safety Invariants (bump allocator):
+ *
+ * The Zig engine uses a bump allocator — resetHeap() resets the pointer to zero,
+ * invalidating ALL prior allocations. Correctness requires:
+ *
+ * 1. resetHeap() MUST be called before each batch of alloc() calls
+ * 2. Between resetHeap() and result consumption: NO await (purely synchronous)
+ * 3. Results MUST be copied to JS (.slice() or parseWasmResult) before next resetHeap()
+ * 4. Each Durable Object owns its own WasmEngine — no cross-DO sharing
+ * 5. Decompress/aggregate helpers allocate without reset (.slice() results) — safe
+ *
+ * These invariants hold because Cloudflare DOs are single-threaded:
+ * no parallel JS execution within a DO, so no interleaving between
+ * resetHeap() and result consumption.
+ */
 export class WasmEngine {
   readonly exports: WasmExports;
   constructor(exports: WasmExports) { this.exports = exports; }
@@ -179,6 +195,13 @@ export class WasmEngine {
     return this.exports.alloc(bytes);
   }
 
+  /**
+   * Reset the bump allocator — invalidates ALL prior allocations.
+   * Call before each batch of alloc()+execute operations.
+   * The caller MUST consume results synchronously (no await) before the next reset.
+   */
+  resetHeap(): void { this.exports.resetHeap(); }
+
   reset(): void { this.exports.resetResult(); }
 
   /** Decompress ZSTD data using the Zig std.compress.zstd implementation. */