fix(worker-shell): bundle sqlite worker runtime

.changeset/sqlite-shared-lock-identity.md

@@ -0,0 +1,5 @@
+---
+"@cloudflare/computer": patch
+---
+
+Fix concurrent `sqlite3` writes losing updates in the Worker shell. Each execution built its own filesystem identity, so two commands writing the same database never shared a lock and the later whole-image writeback silently discarded the earlier one.

packages/computer/src/backends/worker-shell/script/build-bundle.mjs

@@ -41,13 +41,14 @@
 // so any install from this repo or downstream consumer sees a
 // fresh bundle matching the source.
 
-import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
+import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { basename, dirname, resolve } from "node:path";
 import { fileURLToPath } from "node:url";
 import { build } from "esbuild";
 
 import { buildModuleGraph, parseCommandRegistry, partitionModules } from "./partition.mjs";
+import { SQLITE_WASM_PATH, sqliteWorkerPlugin } from "./sqlite-build-plugin.mjs";
 
 const here = dirname(fileURLToPath(import.meta.url));
 // Script lives at .../backends/worker-shell/script/build-bundle.mjs;
@@ -141,6 +142,7 @@ try {
       "seek-bzip",
     ],
     plugins: [
+      sqliteWorkerPlugin(),
       // curl reaches undici only through just-bash's DNS-pinning
       // connection owner, which the Worker backend never activates:
       // ShellWorker registers curl on the plain-`fetch` path
@@ -220,6 +222,8 @@ const partition = partitionModules({ graph, registry, optionalFeatures: OPTIONAL
 // each by its @cloudflare/computer/shell/<group> subpath and
 // spreads them back together.
 const groupNames = ["core", ...Object.keys(OPTIONAL_FEATURES)];
+const sqliteWasm = await readFile(SQLITE_WASM_PATH);
+const sqliteWasmBase64 = sqliteWasm.toString("base64");
 let totalBytes = 0;
 for (const group of groupNames) {
   const names = (partition[group] ?? []).sort();
@@ -231,18 +235,32 @@ for (const group of groupNames) {
   const header =
     `// Generated by script/build-bundle.mjs — do not edit.\n` +
     `// Shell modules exclusive to the "${group}" feature group.\n`;
-  const body = `export default Object.freeze(${JSON.stringify(
-    record,
-    null,
-    2,
-  )}) as Readonly<Record<string, { js: string }>>;\n`;
+  let serializedRecord = JSON.stringify(record, null, 2);
+  let moduleType = "{ js: string }";
+  if (group === "sqlite") {
+    serializedRecord =
+      `{\n  ...${serializedRecord},\n` +
+      `  "sql-wasm.wasm": {\n` +
+      `    wasm: Uint8Array.from(atob(${JSON.stringify(
+        sqliteWasmBase64,
+      )}), (byte) => byte.charCodeAt(0)).buffer,\n` +
+      `  },\n}`;
+    moduleType = "{ js: string } | { wasm: ArrayBuffer }";
+    totalBytes += sqliteWasm.byteLength;
+  }
+  const body =
+    `export default Object.freeze(${serializedRecord}) as Readonly<Record<string, ` +
+    `${moduleType}>>;\n`;
   await writeFile(resolve(outDir, `${group}.ts`), `${header}\n${body}`);
 }
 
 const coreCount = (partition.core ?? []).length;
 const mainBytes = modules["shell.js"].length;
 const featureSummary = Object.keys(OPTIONAL_FEATURES)
-  .map((f) => `${f} ${(partition[f] ?? []).length}`)
+  .map((feature) => {
+    const extraModules = feature === "sqlite" ? 1 : 0;
+    return `${feature} ${(partition[feature] ?? []).length + extraModules}`;
+  })
   .join(", ");
 console.log(
   `Wrote ${outDir} (core ${coreCount} modules, shell.js ${mainBytes} bytes, ` +

packages/computer/src/backends/worker-shell/script/sqlite-build-plugin.mjs

@@ -0,0 +1,167 @@
+import { readFile } from "node:fs/promises";
+import { dirname, resolve } from "node:path";
+import { fileURLToPath } from "node:url";
+
+const here = dirname(fileURLToPath(import.meta.url));
+const justBashEntry = fileURLToPath(import.meta.resolve("just-bash"));
+const sqliteWorker = resolve(dirname(justBashEntry), "../commands/sqlite3/worker.js");
+const sqlJsEntry = fileURLToPath(import.meta.resolve("sql.js"));
+
+export const SQLITE_WASM_PATH = resolve(dirname(sqlJsEntry), "sql-wasm.wasm");
+
+const SQLITE_WORKER_ERROR = "sqlite3 worker not found. Run 'pnpm build' to compile the worker.";
+const FIND_WORKER =
+  'for(let r of t)if(ce(r))return r;throw new Error("' + SQLITE_WORKER_ERROR + '")';
+const CREATE_WORKER =
+  "return new he(e,{workerData:t,resourceLimits:{maxOldGenerationSizeMb:be,maxYoungGenerationSizeMb:we}})";
+
+function replaceExactlyOnce(source, search, replacement, label) {
+  const first = source.indexOf(search);
+  if (first === -1 || source.indexOf(search, first + search.length) !== -1) {
+    throw new Error(`sqlite bundle: expected exactly one ${label}`);
+  }
+  return source.slice(0, first) + replacement + source.slice(first + search.length);
+}
+
+export function sqliteWorkerPlugin() {
+  let commandAdapted = false;
+  let queryWorkerLoaded = false;
+  let sqlJsLoaded = false;
+
+  return {
+    name: "adapt-sqlite-worker",
+    setup(build) {
+      // just-bash's generated command chunk looks for a worker file with
+      // node:fs and constructs node:worker_threads.Worker. Neither mechanism
+      // can reach a Dynamic Worker Loader module, so route both through a lazy
+      // adapter that implements the same small event protocol in this isolate.
+      build.onLoad(
+        { filter: /[\\/]just-bash[\\/]dist[\\/]bundle[\\/]chunks[\\/]chunk-[^/\\]+\.js$/ },
+        async (args) => {
+          let source = await readFile(args.path, "utf8");
+          if (!source.includes(SQLITE_WORKER_ERROR)) return undefined;
+
+          source = replaceExactlyOnce(
+            source,
+            FIND_WORKER,
+            'return "inline:sqlite3"',
+            "sqlite worker lookup",
+          );
+          source = replaceExactlyOnce(
+            source,
+            CREATE_WORKER,
+            "return __createInlineSqliteWorker(t)",
+            "sqlite Worker constructor",
+          );
+          source =
+            `import { createInlineSqliteWorker as __createInlineSqliteWorker } from ${JSON.stringify(
+              resolve(here, "sqlite-command-adapter.mjs"),
+            )};\n` + source;
+          // Wrap the exported command so WorkspaceFsAdapter gets a stable
+          // database-lock identity without changing the always-on adapter.
+          source = replaceExactlyOnce(
+            source,
+            "export{$e as a,_e as b,Fe as c};",
+            "const __sqliteCommand=__adaptSqliteCommand(_e);export{$e as a,__sqliteCommand as b,Fe as c};",
+            "sqlite command export",
+          );
+          source =
+            `import { adaptSqliteCommand as __adaptSqliteCommand } from ${JSON.stringify(
+              resolve(here, "sqlite-command-adapter.mjs"),
+            )};\n` + source;
+          commandAdapted = true;
+          return { contents: source, loader: "js", resolveDir: dirname(args.path) };
+        },
+      );
+
+      // Pull the worker's query implementation into the sqlite feature graph.
+      // Its Node worker entrypoint is guarded by parentPort, which is null here;
+      // exporting executeQuery lets the adapter call the same implementation.
+      build.onResolve({ filter: /^computer:sqlite-query-worker$/ }, () => ({
+        path: "query-worker",
+        namespace: "computer-sqlite",
+      }));
+      build.onLoad({ filter: /^query-worker$/, namespace: "computer-sqlite" }, async () => {
+        let source = await readFile(sqliteWorker, "utf8");
+        // WorkerDefenseInDepth belongs around a dedicated thread. Running it
+        // in the shell's isolate would harden the shell itself after a query.
+        source = replaceExactlyOnce(
+          source,
+          "    activateDefense();\n",
+          "",
+          "worker defense activation",
+        );
+        queryWorkerLoaded = true;
+        return {
+          contents: `${source}\nexport { executeQuery };\n`,
+          loader: "js",
+          resolveDir: dirname(sqliteWorker),
+        };
+      });
+
+      // sql.js normally fetches sql-wasm.wasm from a package filesystem. The
+      // Dynamic Worker instead imports a precompiled module supplied in its
+      // Loader module table, then hands that module to Emscripten explicitly.
+      build.onResolve({ filter: /^sql\.js$/ }, () => ({
+        path: "sql.js",
+        namespace: "computer-sqlite",
+      }));
+      build.onLoad({ filter: /^sql\.js$/, namespace: "computer-sqlite" }, () => ({
+        contents: `
+          import initSqlJs from ${JSON.stringify(sqlJsEntry)};
+          import sqliteWasm from "./sql-wasm.wasm";
+          export default function init(options = {}) {
+            return initSqlJs({
+              ...options,
+              instantiateWasm(imports, receiveInstance) {
+                const instance = new WebAssembly.Instance(sqliteWasm, imports);
+                receiveInstance(instance, sqliteWasm);
+                return instance.exports;
+              },
+            });
+          }
+        `,
+        loader: "js",
+        resolveDir: dirname(sqlJsEntry),
+      }));
+      build.onResolve({ filter: /^\.\/sql-wasm\.wasm$/ }, () => ({
+        path: "./sql-wasm.wasm",
+        external: true,
+      }));
+
+      // sql.js sees WorkerGlobalScope and process shims in workerd and otherwise
+      // chooses loading branches that require self.location or node:fs. Neither
+      // branch is needed when instantiateWasm is supplied by the wrapper above.
+      build.onLoad({ filter: /[\\/]sql\.js[\\/]dist[\\/]sql-wasm\.js$/ }, async (args) => {
+        let source = await readFile(args.path, "utf8");
+        source = replaceExactlyOnce(
+          source,
+          "globalThis.WorkerGlobalScope",
+          "undefined",
+          "sql.js WorkerGlobalScope probe",
+        );
+        source = replaceExactlyOnce(
+          source,
+          "globalThis.process?.versions?.node",
+          "undefined",
+          "sql.js Node probe",
+        );
+        sqlJsLoaded = true;
+        return { contents: source, loader: "js", resolveDir: dirname(args.path) };
+      });
+
+      build.onEnd((result) => {
+        if (result.errors.length > 0) return;
+        if (!commandAdapted || !queryWorkerLoaded || !sqlJsLoaded) {
+          throw new Error(
+            `sqlite bundle: incomplete adaptation (${JSON.stringify({
+              commandAdapted,
+              queryWorkerLoaded,
+              sqlJsLoaded,
+            })})`,
+          );
+        }
+      });
+    },
+  };
+}

packages/computer/src/backends/worker-shell/script/sqlite-command-adapter.mjs

@@ -0,0 +1,103 @@
+// just-bash's sqlite3 command executes queries in a Node Worker thread. The
+// Workers runtime exposes the node:worker_threads API surface but does not
+// implement the Worker constructor. A WorkerShellBackend already runs in its
+// own Dynamic Worker isolate, so execute the query worker in that isolate while
+// preserving the small EventEmitter protocol just-bash expects.
+
+import { executeQuery } from "computer:sqlite-query-worker";
+
+class InlineSqliteWorker {
+  #listeners = new Map();
+  #terminated = false;
+
+  constructor(workerData) {
+    queueMicrotask(() => {
+      void executeQuery(workerData)
+        .then((result) => {
+          this.#emit("message", {
+            ...result,
+            protocolToken: workerData.protocolToken,
+          });
+        })
+        .catch((error) => this.#emit("error", error));
+    });
+  }
+
+  on(event, listener) {
+    let listeners = this.#listeners.get(event);
+    if (listeners === undefined) {
+      listeners = new Set();
+      this.#listeners.set(event, listeners);
+    }
+    listeners.add(listener);
+    return this;
+  }
+
+  removeListener(event, listener) {
+    this.#listeners.get(event)?.delete(listener);
+    return this;
+  }
+
+  async terminate() {
+    this.#terminated = true;
+    return 0;
+  }
+
+  #emit(event, value) {
+    if (this.#terminated) return;
+    for (const listener of this.#listeners.get(event) ?? []) listener(value);
+  }
+}
+
+// just-bash keys its sqlite3 database locks on `fsIdentity ?? fs`, then on the
+// canonical database path. ShellWorker.exec builds a fresh Bash and
+// WorkspaceFsAdapter per execution, and the wrapper below allocates a fresh
+// Proxy on top of that, so without an explicit identity every execution lands
+// in its own lock bucket and concurrent writers never contend. Because
+// writeback replaces the whole database image, the later writer would silently
+// discard the earlier one. Every execution in this isolate addresses the same
+// workspace store, so pin one shared identity and let the canonical path keep
+// distinguishing databases.
+const SQLITE_FS_IDENTITY = {};
+
+// WorkspaceFsAdapter has no Node dev/ino pair because its storage is remote.
+// It also does not support hard links, so a canonical path is a stable lock
+// identity for the database. Keep this compatibility layer in SQLite's lazy
+// chunk instead of adding bytes to every Worker shell.
+function filesystemWithStableIdentity(fs) {
+  return new Proxy(fs, {
+    get(target, property) {
+      if (property === "stat") {
+        return async (path) => {
+          const stat = await target.stat(path);
+          if (stat.identity !== undefined || (stat.dev !== undefined && stat.ino !== undefined)) {
+            return stat;
+          }
+          return { ...stat, identity: await target.realpath(path) };
+        };
+      }
+
+      // WorkspaceFsAdapter uses private fields, so methods must retain the
+      // original receiver rather than receiving this Proxy as `this`.
+      const value = Reflect.get(target, property, target);
+      return typeof value === "function" ? value.bind(target) : value;
+    },
+  });
+}
+
+export function createInlineSqliteWorker(workerData) {
+  return new InlineSqliteWorker(workerData);
+}
+
+export function adaptSqliteCommand(command) {
+  return {
+    ...command,
+    execute(args, context) {
+      return command.execute(args, {
+        ...context,
+        fs: filesystemWithStableIdentity(context.fs),
+        fsIdentity: SQLITE_FS_IDENTITY,
+      });
+    },
+  };
+}

packages/computer/src/backends/worker-shell/shell-modules.test.ts

@@ -136,6 +136,21 @@ describe("shell feature groups", () => {
     expect(curlOnDisk).toBe(true);
   });
 
+  it("ships SQLite's query worker and WebAssembly module only in the sqlite group", () => {
+    expect(sqliteModules["sql-wasm.wasm"]).toMatchObject({
+      wasm: expect.any(ArrayBuffer),
+    });
+    expect(SHELL_CORE_MODULES["sql-wasm.wasm"]).toBeUndefined();
+
+    const sqliteSource = Object.values(sqliteModules)
+      .filter((module): module is { js: string } => "js" in module)
+      .map((module) => module.js)
+      .join("\n");
+    expect(sqliteSource).toContain("function executeQuery");
+    expect(sqliteSource).toContain("InlineSqliteWorker");
+    expect(sqliteSource).not.toContain("sqlite3 worker not found");
+  });
+
   it("keeps feature groups disjoint from each other", () => {
     // A chunk owned by one feature must not also appear in another;
     // a shared chunk belongs in core.

packages/computer/src/backends/worker-shell/shell-modules.ts

@@ -19,7 +19,7 @@ import coreModules from "@cloudflare/computer/shell/core";
 // One generated feature group: module name -> source string. The
 // core group and every @cloudflare/computer/shell/<feature> import
 // share this shape.
-export type ShellModuleGroup = Readonly<Record<string, { js: string }>>;
+export type ShellModuleGroup = Readonly<Record<string, { js: string } | { wasm: ArrayBuffer }>>;
 
 // The always-on core group. Ships in every Worker shell; carries
 // the ShellWorker entry (shell.js), the base command set, and the
@@ -29,10 +29,8 @@ export const SHELL_CORE_MODULES: ShellModuleGroup = Object.freeze({ ...coreModul
 // Merge the core group with the optional groups the consumer
 // imported and passed. Later groups win on key collisions, but the
 // build keeps groups disjoint so order never matters in practice.
-export function assembleShellModules(
-  groups: readonly ShellModuleGroup[] = [],
-): Readonly<Record<string, { js: string }>> {
-  const modules: Record<string, { js: string }> = { ...coreModules };
+export function assembleShellModules(groups: readonly ShellModuleGroup[] = []): ShellModuleGroup {
+  const modules: Record<string, { js: string } | { wasm: ArrayBuffer }> = { ...coreModules };
   for (const group of groups) {
     Object.assign(modules, group);
   }