refactor: remove HTTP fallback handlers — RPC only

DOs (QueryDO, MasterDO, FragmentDO) no longer expose fetch() handlers.
All inter-DO calls and Worker→DO calls use RPC methods directly,
eliminating duplicate HTTP+RPC code paths and JSON serialization overhead.
Worker.ts is now the sole HTTP↔RPC translation layer.

Author: teamchong

src/fragment-do.test.ts

@@ -1,6 +1,5 @@
 import { describe, it, expect, vi } from "vitest";
 import { FragmentDO } from "./fragment-do.js";
-import { bigIntReplacer } from "./decode.js";
 import type { TableMeta, ColumnMeta } from "./types.js";
 
 // Mock WASM module import (not available in vitest)
@@ -40,41 +39,23 @@ const mockEnv = {
   QUERY_DO: null,
 } as any;
 
-function makeScanRequest(url: string, body: unknown): Request {
-  return new Request(url, {
-    method: "POST",
-    body: JSON.stringify(body, bigIntReplacer),
-    headers: { "content-type": "application/json" },
-  });
-}
-
 describe("FragmentDO", () => {
   it("is constructible", () => {
     const fdo = new FragmentDO(mockState, mockEnv);
     expect(fdo).toBeDefined();
   });
 
-  it("returns 404 for unknown path", async () => {
-    const fdo = new FragmentDO(mockState, mockEnv);
-    const res = await fdo.fetch(new Request("http://internal/unknown"));
-    expect(res.status).toBe(404);
-    expect(await res.text()).toBe("Not found");
-  });
-
-  it("returns valid JSON for /scan with empty fragments", async () => {
+  it("returns valid result for scanRpc with empty fragments", async () => {
     const fdo = new FragmentDO(mockState, mockEnv);
-    const req = makeScanRequest("http://internal/scan", {
-      fragments: [],
-      query: { table: "test", filters: [], projections: [] },
-    });
-    const res = await fdo.fetch(req);
-    expect(res.status).toBe(200);
-    const body = await res.json();
-    expect(body.rows).toEqual([]);
-    expect(body.rowCount).toBe(0);
-    expect(body.bytesRead).toBe(0);
-    expect(body.pagesSkipped).toBe(0);
-    expect(typeof body.durationMs).toBe("number");
+    const result = await fdo.scanRpc(
+      [],
+      { table: "test", filters: [], projections: [] } as any,
+    );
+    expect(result.rows).toEqual([]);
+    expect(result.rowCount).toBe(0);
+    expect(result.bytesRead).toBe(0);
+    expect(result.pagesSkipped).toBe(0);
+    expect(typeof result.durationMs).toBe("number");
   });
 
   it("handles scan with mock fragments gracefully when R2 returns null", async () => {
@@ -101,17 +82,14 @@ describe("FragmentDO", () => {
     };
 
     const fdo = new FragmentDO(mockState, mockEnv);
-    const req = makeScanRequest("http://internal/scan", {
-      fragments: [{ r2Key: "users.lance", meta }],
-      query: { table: "users", filters: [], projections: [] },
-    });
-    const res = await fdo.fetch(req);
-    expect(res.status).toBe(200);
-    const body = await res.json();
-    // R2 returns null so no data is read, but response is still valid
-    expect(body.rows).toEqual([]);
-    expect(body.rowCount).toBe(0);
-    expect(body.bytesRead).toBe(0);
-    expect(body.columns).toEqual(["id"]);
+    const result = await fdo.scanRpc(
+      [{ r2Key: "users.lance", meta }],
+      { table: "users", filters: [], projections: [] } as any,
+    );
+    // R2 returns null so no data is read, but result is still valid
+    expect(result.rows).toEqual([]);
+    expect(result.rowCount).toBe(0);
+    expect(result.bytesRead).toBe(0);
+    expect(result.columns).toEqual(["id"]);
   });
 });

src/fragment-do.ts

@@ -1,6 +1,6 @@
 import type { Env, TableMeta, QueryResult, Row } from "./types.js";
 import type { QueryDescriptor } from "./client.js";
-import { canSkipPage, bigIntReplacer } from "./decode.js";
+import { canSkipPage } from "./decode.js";
 import { instantiateWasm, type WasmEngine } from "./wasm-engine.js";
 import { coalesceRanges, fetchBounded, withRetry, withTimeout } from "./coalesce.js";
 import { R2SpillBackend } from "./r2-spill.js";
@@ -39,14 +39,6 @@ export class FragmentDO extends DurableObject<Env> {
     );
   }
 
-  async fetch(request: Request): Promise<Response> {
-    await this.ensureInitialized();
-
-    const url = new URL(request.url);
-    if (url.pathname === "/scan") return this.handleScan(request);
-    return new Response("Not found", { status: 404 });
-  }
-
   private async ensureInitialized(): Promise<void> {
     if (this.initialized) return;
     this.initialized = true;
@@ -58,27 +50,7 @@ export class FragmentDO extends DurableObject<Env> {
     this.wasmEngine = await instantiateWasm(wasmModule);
   }
 
-  private json(body: unknown, status = 200): Response {
-    return new Response(JSON.stringify(body, bigIntReplacer), {
-      status, headers: { "content-type": "application/json" },
-    });
-  }
-
-  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. */
+  /** Core scan logic used by RPC. */
   private async executeScan(fragments: ScanRequest["fragments"], query: QueryDescriptor): Promise<QueryResult> {
     const t0 = Date.now();
     let totalBytesRead = 0;

src/master-do.ts

@@ -2,7 +2,7 @@ import type { ColumnMeta, Env, Footer, TableMeta, DatasetMeta, AppendResult } fr
 import { parseFooter, parseColumnMetaFromProtobuf, FOOTER_SIZE } from "./footer.js";
 import { parseManifest } from "./manifest.js";
 import { detectFormat, getParquetFooterLength, parseParquetFooter, parquetMetaToTableMeta } from "./parquet.js";
-import { bigIntReplacer } from "./decode.js";
+import type { QueryDORpc } from "./types.js";
 import { instantiateWasm, type WasmEngine } from "./wasm-engine.js";
 import wasmModule from "./wasm-module.js";
 
@@ -15,25 +15,9 @@ export class MasterDO extends DurableObject<Env> {
     super(ctx, env);
   }
 
-  private json(body: unknown, status = 200): Response {
-    return new Response(JSON.stringify(body), {
-      status, headers: { "content-type": "application/json" },
-    });
-  }
-
-  async fetch(request: Request): Promise<Response> {
-    switch (new URL(request.url).pathname) {
-      case "/register": return this.handleRegister(request);
-      case "/write":    return this.handleWrite(request);
-      case "/append":   return this.handleAppend(request);
-      case "/refresh":  return this.handleRefresh(request);
-      case "/tables":   return this.handleListTables();
-      default:          return new Response("Not found", { status: 404 });
-    }
-  }
+  // ── RPC methods ────────────────────────────────────────────────────────
 
-  private async handleRegister(request: Request): Promise<Response> {
-    const { queryDoId, region } = (await request.json()) as { queryDoId: string; region: string };
+  async registerRpc(queryDoId: string, region: string): Promise<{ registered: boolean; region: string; tableVersions?: Record<string, { r2Key: string; updatedAt: number }> }> {
     const regions = (await this.ctx.storage.get<Record<string, string>>("regions")) ?? {};
     regions[region] = queryDoId;
     await this.ctx.storage.put("regions", regions);
@@ -45,22 +29,22 @@ export class MasterDO extends DurableObject<Env> {
       const name = key.replace("table:", "");
       tableVersions[name] = { r2Key: meta.r2Key ?? name, updatedAt: meta.updatedAt ?? 0 };
     }
-    return this.json({ registered: true, region, tableVersions });
+    return { registered: true, region, tableVersions };
   }
 
-  private async handleWrite(request: Request): Promise<Response> {
-    const { r2Key } = (await request.json()) as { r2Key: string };
+  async writeRpc(body: unknown): Promise<unknown> {
+    const { r2Key } = body as { r2Key: string };
     if (!r2Key || typeof r2Key !== "string" || r2Key.includes("..")) {
-      return this.json({ error: "Invalid r2Key" }, 400);
+      throw new Error("Invalid r2Key");
     }
 
     // Check if this is a dataset directory (ends with / or .lance/)
     if (r2Key.endsWith("/") || r2Key.endsWith(".lance/")) {
-      return this.handleDatasetWrite(r2Key);
+      return this.executeDatasetWrite(r2Key);
     }
 
     const result = await this.readFooterAndColumns(r2Key);
-    if (!result) return this.json({ error: "Failed to read footer" }, 500);
+    if (!result) throw new Error("Failed to read footer");
 
     const tableName = r2Key.replace(/\.(lance|parquet)$/, "").split("/").pop() ?? r2Key;
     const totalRows = result.columns[0]?.pages.reduce((s, p) => s + p.rowCount, 0) ?? 0;
@@ -70,26 +54,26 @@ export class MasterDO extends DurableObject<Env> {
     };
     await this.ctx.storage.put(`table:${tableName}`, meta);
     await this.broadcast(tableName, r2Key, result, { totalRows });
-    return this.json({ success: true, table: tableName });
+    return { success: true, table: tableName };
   }
 
-  /** Handle write notification for a multi-fragment dataset directory. */
-  private async handleDatasetWrite(r2Prefix: string): Promise<Response> {
+  /** Write notification for a multi-fragment dataset directory. */
+  private async executeDatasetWrite(r2Prefix: string): Promise<unknown> {
     const tableName = r2Prefix.replace(/\/$/, "").replace(/\.lance$/, "").split("/").pop() ?? r2Prefix;
 
     // Find latest manifest
     const listed = await this.env.DATA_BUCKET.list({ prefix: `${r2Prefix}_versions/`, limit: 100 });
     const manifestKeys = listed.objects
       .filter(o => o.key.endsWith(".manifest"))
       .sort((a, b) => a.key.localeCompare(b.key));
-    if (manifestKeys.length === 0) return this.json({ error: "No manifests found" }, 404);
+    if (manifestKeys.length === 0) throw new Error("No manifests found");
 
     const latestKey = manifestKeys[manifestKeys.length - 1].key;
     const manifestObj = await this.env.DATA_BUCKET.get(latestKey);
-    if (!manifestObj) return this.json({ error: "Failed to read manifest" }, 500);
+    if (!manifestObj) throw new Error("Failed to read manifest");
 
     const manifest = parseManifest(await manifestObj.arrayBuffer());
-    if (!manifest) return this.json({ error: "Failed to parse manifest" }, 500);
+    if (!manifest) throw new Error("Failed to parse manifest");
 
     // Read first fragment's footer to broadcast (Query DOs will discover the rest)
     if (manifest.fragments.length > 0) {
@@ -110,7 +94,7 @@ export class MasterDO extends DurableObject<Env> {
       name: tableName, r2Prefix, manifest, totalRows: manifest.totalRows, updatedAt: Date.now(),
     });
 
-    return this.json({ success: true, table: tableName, fragments: manifest.fragments.length, totalRows: manifest.totalRows });
+    return { success: true, table: tableName, fragments: manifest.fragments.length, totalRows: manifest.totalRows };
   }
 
   private async getWasm(): Promise<WasmEngine> {
@@ -119,29 +103,7 @@ export class MasterDO extends DurableObject<Env> {
     return this.wasmEngine;
   }
 
-  /** Append rows to a table using CAS coordination.
-   *  1. Build Lance fragment from row data via WASM
-   *  2. PUT data file to R2 (unique name, no conflict)
-   *  3. CAS loop: read _latest → build new manifest → PUT with ETag match
-   */
-  private async handleAppend(request: Request): Promise<Response> {
-    const { table, rows } = (await request.json()) as {
-      table: string;
-      rows: Record<string, unknown>[];
-    };
-
-    if (!rows?.length) return this.json({ error: "No rows provided" }, 400);
-    try {
-      const result = await this.executeAppend(table, rows);
-      return this.json(result);
-    } catch (err) {
-      const msg = err instanceof Error ? err.message : String(err);
-      const status = msg.includes("CAS failed") ? 409 : 400;
-      return this.json({ error: msg }, status);
-    }
-  }
-
-  /** Core append logic shared by HTTP fetch handler and RPC. */
+  /** Core append logic. */
   private async executeAppend(table: string, rows: Record<string, unknown>[]): Promise<AppendResult> {
     if (!rows?.length) throw new Error("No rows provided");
 
@@ -355,22 +317,22 @@ export class MasterDO extends DurableObject<Env> {
     return new Uint8Array(bytes);
   }
 
-  private async handleRefresh(request: Request): Promise<Response> {
-    const { r2Key } = (await request.json()) as { r2Key: string };
+  async refreshRpc(body: unknown): Promise<unknown> {
+    const { r2Key } = body as { r2Key: string };
     if (!r2Key || typeof r2Key !== "string" || r2Key.includes("..")) {
-      return this.json({ error: "Invalid r2Key" }, 400);
+      throw new Error("Invalid r2Key");
     }
     const result = await this.readFooterAndColumns(r2Key);
-    if (!result) return this.json({ error: "Failed to read footer" }, 500);
+    if (!result) throw new Error("Failed to read footer");
 
     const tableName = r2Key.replace(/\.(lance|parquet)$/, "").split("/").pop() ?? r2Key;
     await this.broadcast(tableName, r2Key, result);
-    return this.json({ refreshed: true, table: tableName });
+    return { refreshed: true, table: tableName };
   }
 
-  private async handleListTables(): Promise<Response> {
+  async listTablesRpc(): Promise<{ tables: string[] }> {
     const tables = await this.ctx.storage.list<TableMeta>({ prefix: "table:" });
-    return this.json({ tables: [...tables.keys()].map(k => k.replace("table:", "")) });
+    return { tables: [...tables.keys()].map(k => k.replace("table:", "")) };
   }
 
   /** Read footer + column metadata from R2 (2 range reads, done once by Master). */
@@ -427,28 +389,26 @@ export class MasterDO extends DurableObject<Env> {
     return { parsed, raw, fileSize, columns, format: "lance" };
   }
 
-  /** Broadcast invalidation with pre-parsed columns to all Query DOs. */
+  /** Broadcast invalidation with pre-parsed columns to all Query DOs via RPC. */
   private async broadcast(
     table: string, r2Key: string,
     footer: { raw: ArrayBuffer; fileSize: bigint; columns: ColumnMeta[]; format?: "lance" | "parquet" },
     opts?: { totalRows?: number; r2Prefix?: string },
   ): Promise<void> {
     const regions = (await this.ctx.storage.get<Record<string, string>>("regions")) ?? {};
-    const payload = JSON.stringify({
+    const payload = {
       table, r2Key, columns: footer.columns, format: footer.format ?? "lance",
       footerBytes: Array.from(new Uint8Array(footer.raw)),
       fileSize: footer.fileSize.toString(), timestamp: Date.now(),
       ...(opts?.totalRows != null ? { totalRows: opts.totalRows } : {}),
       ...(opts?.r2Prefix != null ? { r2Prefix: opts.r2Prefix } : {}),
-    }, bigIntReplacer);
+    };
 
     const deadRegions: string[] = [];
     await Promise.allSettled(Object.entries(regions).map(async ([region, doId]) => {
       try {
-        const queryDo = this.env.QUERY_DO.get(this.env.QUERY_DO.idFromString(doId));
-        await queryDo.fetch(new Request("http://internal/invalidate", {
-          method: "POST", body: payload, headers: { "content-type": "application/json" },
-        }));
+        const queryDo = this.env.QUERY_DO.get(this.env.QUERY_DO.idFromString(doId)) as unknown as QueryDORpc;
+        await queryDo.invalidateRpc(payload);
         this.broadcastFailures.delete(region);
       } catch {
         const count = (this.broadcastFailures.get(region) ?? 0) + 1;