feat: Worker RPC + edge page cache with VIP gating

- Refactor all DOs to extend DurableObject<Env> for zero-serialization RPC
- Add RPC methods (queryRpc, countRpc, existsRpc, firstRpc, explainRpc,
  streamRpc, appendRpc, scanRpc) — structured clone instead of JSON
- Add 3-tier page cache: L1 (WASM buffer pool) → L2 (caches.default) → L3 (R2)
- Gate L2 edge cache on VIP tables (accessCount >= 3) to prevent pollution
- Extract shared helpers: applyJsPostProcessing, assembleRowsFromColumns,
  executeCount, loadParquetR2Meta, loadParquetFragments, cacheIcebergDataset
- Fix missing "in" filter in JS post-processing path
- Fix bigint↔number coercion in Lance whole-file filter path
- Fix edgeCacheMisses inflation for non-VIP tables
- Fix error status codes (400 for parse, 500 for execution)
- Fix stack trace leak in handleQuery error response
- Add DurableObject polyfill for vitest compatibility

Author: teamchong

src/fragment-do.ts

@@ -18,17 +18,14 @@ interface ScanRequest {
  * across queries and datasets. Footer cache persists across scans,
  * and the DO hibernates when idle — next query wakes it with cache warm.
  */
-export class FragmentDO implements DurableObject {
-  private state: DurableObjectState;
-  private env: Env;
+export class FragmentDO extends DurableObject<Env> {
   private wasmEngine!: WasmEngine;
   /** Footer cache keyed by r2Key — survives across scans while the DO is alive. */
   private footerCache = new Map<string, TableMeta>();
   private initialized = false;
 
-  constructor(state: DurableObjectState, env: Env) {
-    this.state = state;
-    this.env = env;
+  constructor(ctx: DurableObjectState, env: Env) {
+    super(ctx, env);
   }
 
   private log(level: "info" | "warn" | "error", msg: string, data?: Record<string, unknown>): void {
@@ -50,7 +47,7 @@ export class FragmentDO implements DurableObject {
     this.initialized = true;
 
     // Restore footer cache from SQLite (survives hibernation)
-    const stored = await this.state.storage.list<TableMeta>({ prefix: "frag:" });
+    const stored = await this.ctx.storage.list<TableMeta>({ prefix: "frag:" });
     for (const [key, meta] of stored) this.footerCache.set(key.replace("frag:", ""), meta);
 
     this.wasmEngine = await instantiateWasm(wasmModule);
@@ -64,6 +61,20 @@ export class FragmentDO implements DurableObject {
 
   private async handleScan(request: Request): Promise<Response> {
     const { fragments, query } = (await request.json()) as ScanRequest;
+    const result = await this.executeScan(fragments, query);
+
+    this.log("info", "scan_complete", {
+      fragmentCount: fragments.length, rowCount: result.rowCount,
+      bytesRead: result.bytesRead, durationMs: result.durationMs,
+      r2ReadMs: result.r2ReadMs, wasmExecMs: result.wasmExecMs,
+      cacheHits: result.cacheHits, cacheMisses: result.cacheMisses,
+    });
+
+    return this.json(result);
+  }
+
+  /** Core scan logic shared by HTTP fetch handler and RPC. */
+  private async executeScan(fragments: ScanRequest["fragments"], query: QueryDescriptor): Promise<QueryResult> {
     const t0 = Date.now();
     let totalBytesRead = 0;
     let totalPagesSkipped = 0;
@@ -79,7 +90,7 @@ export class FragmentDO implements DurableObject {
       const effectiveMeta = (cachedMeta && cachedMeta.updatedAt >= meta.updatedAt) ? cachedMeta : meta;
       if (!cachedMeta || cachedMeta.updatedAt < meta.updatedAt) {
         this.footerCache.set(r2Key, meta);
-        this.state.storage.put(`frag:${r2Key}`, meta);
+        this.ctx.storage.put(`frag:${r2Key}`, meta);
       }
 
       let cols = query.projections.length > 0
@@ -182,7 +193,7 @@ export class FragmentDO implements DurableObject {
       totalWasmExecMs += Date.now() - wasmStart;
     }
 
-    const result: QueryResult = {
+    return {
       rows: allRows,
       rowCount: allRows.length,
       columns: query.projections.length > 0
@@ -196,14 +207,11 @@ export class FragmentDO implements DurableObject {
       cacheHits: totalCacheHits,
       cacheMisses: totalCacheMisses,
     };
+  }
 
-    this.log("info", "scan_complete", {
-      fragmentCount: fragments.length, rowCount: result.rowCount,
-      bytesRead: totalBytesRead, durationMs: result.durationMs,
-      r2ReadMs: totalR2ReadMs, wasmExecMs: totalWasmExecMs,
-      cacheHits: totalCacheHits, cacheMisses: totalCacheMisses,
-    });
-
-    return this.json(result);
+  /** RPC: Scan fragments — zero-serialization call from QueryDO. */
+  async scanRpc(fragments: ScanRequest["fragments"], query: QueryDescriptor): Promise<QueryResult> {
+    await this.ensureInitialized();
+    return this.executeScan(fragments, query);
   }
 }

src/index.ts

@@ -1,7 +1,6 @@
 import { TableQuery } from "./client.js";
 import type { QueryDescriptor, QueryExecutor } from "./client.js";
-import type { AppendResult, ExplainResult, QueryResult, Row } from "./types.js";
-import { bigIntReplacer } from "./decode.js";
+import type { AppendResult, ExplainResult, QueryResult, Row, QueryDORpc, MasterDORpc } from "./types.js";
 import { LocalExecutor } from "./local-executor.js";
 
 export { MasterDO } from "./master-do.js";
@@ -34,6 +33,8 @@ export type {
   AppendResult,
   ExplainResult,
   VectorIndexInfo,
+  QueryDORpc,
+  MasterDORpc,
 } from "./types.js";
 
 /**
@@ -95,8 +96,9 @@ export class QueryMode {
 }
 
 /**
- * Executor that sends queries to a regional Query DO.
- * The DO has cached footers — no metadata round-trip needed.
+ * Executor that calls regional Query DO via Worker RPC.
+ * Zero-serialization: structured clone instead of JSON, no HTTP overhead.
+ * RPC sessions survive DO hibernation — each session is one billable request.
  */
 class RemoteExecutor implements QueryExecutor {
   private namespace: DurableObjectNamespace;
@@ -111,76 +113,45 @@ class RemoteExecutor implements QueryExecutor {
     this.masterNamespace = masterNamespace;
   }
 
-  private getQueryDo() {
+  /** Get a typed RPC handle for the regional Query DO. */
+  private getQueryHandle(): QueryDORpc {
     const doName = `query-${this.region}`;
     const id = this.namespace.idFromName(doName);
-    return this.locationHint
+    const doRef = this.locationHint
       ? this.namespace.get(id, { locationHint: this.locationHint as DurableObjectLocationHint })
       : this.namespace.get(id);
+    return doRef as unknown as QueryDORpc;
   }
 
+  /** Execute query via RPC — zero JSON serialization overhead. */
   async execute(query: QueryDescriptor): Promise<QueryResult> {
-    const queryDo = this.getQueryDo();
-
-    const response = await queryDo.fetch(new Request("http://internal/query", {
-      method: "POST",
-      body: JSON.stringify(query, bigIntReplacer),
-      headers: { "content-type": "application/json" },
-    }));
-
-    if (!response.ok) {
-      const error = await response.text();
-      throw new Error(`QueryMode query failed: ${error}`);
-    }
-
-    return response.json() as Promise<QueryResult>;
+    const rpc = this.getQueryHandle();
+    return rpc.queryRpc(query);
   }
 
+  /** Append rows via RPC to Master DO. */
   async append(table: string, rows: Record<string, unknown>[]): Promise<AppendResult> {
     if (!this.masterNamespace) {
       throw new Error("append() requires masterDoNamespace — pass it via QueryMode.remote(queryDO, { masterDO })");
     }
-    // Append goes to Master DO (single writer)
     const id = this.masterNamespace.idFromName("master");
-    const masterDo = this.masterNamespace.get(id);
-
-    const response = await masterDo.fetch(new Request("http://internal/append", {
-      method: "POST",
-      body: JSON.stringify({ table, rows }, bigIntReplacer),
-      headers: { "content-type": "application/json" },
-    }));
-
-    if (!response.ok) {
-      const error = await response.text();
-      throw new Error(`QueryMode append failed: ${error}`);
-    }
-
-    return response.json() as Promise<AppendResult>;
+    const masterRpc = this.masterNamespace.get(id) as unknown as MasterDORpc;
+    return masterRpc.appendRpc(table, rows);
   }
 
+  /** Stream query results via RPC — native ReadableStream, no HTTP wrapper. */
   async executeStream(query: QueryDescriptor): Promise<ReadableStream<Row>> {
-    const queryDo = this.getQueryDo();
-
-    const response = await queryDo.fetch(new Request("http://internal/query/stream", {
-      method: "POST",
-      body: JSON.stringify(query, bigIntReplacer),
-      headers: { "content-type": "application/json" },
-    }));
-
-    if (!response.ok) {
-      const error = await response.text();
-      throw new Error(`QueryMode stream failed: ${error}`);
-    }
-
-    if (!response.body) throw new Error("No response body for stream");
+    const rpc = this.getQueryHandle();
+    const byteStream = await rpc.streamRpc(query);
 
+    // Parse NDJSON byte stream into Row objects
     const decoder = new TextDecoder();
-    const MAX_STREAM_BUFFER = 10 * 1024 * 1024; // 10MB
+    const MAX_STREAM_BUFFER = 10 * 1024 * 1024;
     let buffer = "";
 
     return new ReadableStream<Row>({
       async start(controller) {
-        const reader = response.body!.getReader();
+        const reader = byteStream.getReader();
         try {
           while (true) {
             const { done, value } = await reader.read();
@@ -206,36 +177,27 @@ class RemoteExecutor implements QueryExecutor {
     });
   }
 
-  private async postQuery<T>(path: string, query: QueryDescriptor): Promise<T> {
-    const queryDo = this.getQueryDo();
-    const response = await queryDo.fetch(new Request(`http://internal${path}`, {
-      method: "POST",
-      body: JSON.stringify(query, bigIntReplacer),
-      headers: { "content-type": "application/json" },
-    }));
-    if (!response.ok) {
-      const error = await response.text();
-      throw new Error(`QueryMode ${path} failed: ${error}`);
-    }
-    return response.json() as Promise<T>;
-  }
-
+  /** Count via RPC — returns number directly, no JSON wrapper. */
   async count(query: QueryDescriptor): Promise<number> {
-    const body = await this.postQuery<{ count: number }>("/query/count", query);
-    return body.count;
+    const rpc = this.getQueryHandle();
+    return rpc.countRpc(query);
   }
 
+  /** Exists via RPC — returns boolean directly. */
   async exists(query: QueryDescriptor): Promise<boolean> {
-    const body = await this.postQuery<{ exists: boolean }>("/query/exists", query);
-    return body.exists;
+    const rpc = this.getQueryHandle();
+    return rpc.existsRpc(query);
   }
 
+  /** First via RPC — returns Row directly. */
   async first(query: QueryDescriptor): Promise<Row | null> {
-    const body = await this.postQuery<{ row: Row | null }>("/query/first", query);
-    return body.row;
+    const rpc = this.getQueryHandle();
+    return rpc.firstRpc(query);
   }
 
+  /** Explain via RPC — returns ExplainResult directly. */
   async explain(query: QueryDescriptor): Promise<ExplainResult> {
-    return this.postQuery<ExplainResult>("/query/explain", query);
+    const rpc = this.getQueryHandle();
+    return rpc.explainRpc(query);
   }
 }