exec() refuses SQL without a statement; new events in the runtime's async context

packages/cellb/conformance/scenarios.ts

@@ -499,7 +499,7 @@ scenarios["wd-sql-default"] = {
     },
   ],
   knownGap:
-    "cellb's exec() runs a script that ends in a comment where celld refuses it, an assertion before the one celld stops at (an ingest syntax error, reported with celld's own stack frames)",
+    "both stop at the same assertion (an ingest syntax error); celld reports it with its own stack frames",
 };
 
 // celld's deviations from workerd (conformance/record-workerd.ts); cellb follows celld.

packages/cellb/docs/gaps.md

@@ -10,7 +10,7 @@ from the celld golden, and turns red when it stops differing ([conformance.md](c
 
 | Scenario | What differs |
 |---|---|
-| `wd-sql-default` | `sql.exec('SELECT 1; -- comment')` runs; celld refuses it (workerd: `SQL code did not contain a statement`), so cellb fails an earlier assertion than celld. celld stops at `sql.ingest()`'s syntax error, which cellb words the same (`sql-ingest` agrees), but celld's report of it carries celld's own stack frames. |
+| `wd-sql-default` | cellb and celld stop at the same assertion, `sql.ingest()`'s syntax error, which cellb words as celld does (`sql-ingest` agrees), but celld's report of it carries celld's own stack frames. |
 | `wd-actor-fetch-rejection` | No global `addEventListener('unhandledrejection')` in Node; celld's own `assert` rethrows its `rejected: … [at …]` error. |
 | `wd-websocket-hibernation` | The other way round: celld fails this workerd test (`got `); cellb passes it, as workerd does. |
 | `wd-websocket-constructor-gaps` | The host's own classes: Node's `WebSocket` takes `'BLOB'` as `'blob'`, and Chromium's `CloseEvent` keeps lone surrogates in a close reason. |
@@ -95,12 +95,12 @@ Dedicated Workers all work in Firefox 157, and the whole suite passes there. Wha
   stores: a `run()`'s store stays set until its promise settles, which is right for one run at a
   time and shared between concurrent ones (two cells answering at once can read each other's).
   Event listeners (`addEventListener`, MessagePort `onmessage`) do not carry stores, on Node either.
-- **A called object starts in its caller's async context.** Its `AsyncLocalStorage` stores are the
-  caller's (on Node and in a browser), where on Cloudflare a call to another object is a new event
-  with none. The Agents SDK and Flue set their own stores at each entry, so they are not affected;
-  code that reads a store it did not set in the callee sees the caller's. Changing it means
-  starting each event in the root context, which the celld conformance harness (test/p09) relies
-  on not doing: it attributes Durable Object logs to a scenario through such a store.
+- **An object's events start in the async context the runtime was made in.** A call to an object
+  is a new event, as on Cloudflare: the callee (and its constructor) does not see the caller's
+  `AsyncLocalStorage` stores. Where Cloudflare has none, cellb has the stores that were current
+  when `createCellRuntime` ran, so a host can scope a whole runtime (test/p09 attributes each
+  scenario's logs this way). On Node this needs `process.getBuiltinModule` (Node 22.3 and later);
+  in a browser it goes through cellb's `node:async_hooks` polyfill.
 - **A browser runtime is one thread, and OPFS writes are its cost.** Its cells interleave, and
   each commit's SQLite I/O is synchronous on that thread. A Flue conversation turn (a prompt, a tool
   call, two model turns) makes 21 commits (155 when commits ended with each microtask); in this

packages/cellb/e2e/p11f-celld-one-bucket.test.ts

@@ -529,8 +529,11 @@ describe.skipIf(!CELLD)(
         await w.close();
         celld = await celldDev(dir);
         try {
+          // celld wakes cells from its wake index by minute (wake/entries/<UTC minute>/…): a
+          // message due in a later minute than celld started in is woken at its next scan, up
+          // to a minute on. So up to 80 s here; within one minute it is a few seconds.
           let got: unknown[] = [];
-          for (let i = 0; i < 40 && got.length < 2; i++) {
+          for (let i = 0; i < 160 && got.length < 2; i++) {
             await new Promise((r) => setTimeout(r, 500));
             got = await (await fetch(`${celld.base}/`)).json();
           }
@@ -559,7 +562,7 @@ describe.skipIf(!CELLD)(
     it(
       "Queues: a delayed message sent on cellb is delivered by celld",
       () => handoff("cellb→celld"),
-      120_000,
+      150_000, // celld may take up to a minute to wake the queue (above)
     );
 
     // A workflow instance is a cell (celld's own workflow code on both sides): a sleeping or

packages/cellb/src/core/async-context.ts

@@ -0,0 +1,45 @@
+/** Runs `fn` in an async context captured earlier. */
+export type AsyncRoot = <R>(fn: () => R) => R;
+
+const POLYFILL_STATE = Symbol.for("cellb.asyncContext.state");
+
+/**
+ * The async context current now, to start events in: on Cloudflare a call to a Durable Object is
+ * a new event, which sees no `AsyncLocalStorage` store its caller set. On Node it is every store
+ * (`AsyncLocalStorage.snapshot()`); in a browser, the `node:async_hooks` polyfill's frame (a
+ * realm's polyfill copies share it), or its root frame if the polyfill loads later. Elsewhere
+ * events run in their caller's context, as before.
+ */
+export function asyncRoot(): AsyncRoot {
+  const hooks = (
+    globalThis as {
+      process?: { getBuiltinModule?(id: string): unknown };
+    }
+  ).process?.getBuiltinModule?.("node:async_hooks") as
+    | {
+        AsyncLocalStorage?: { snapshot?(): <R>(fn: () => R) => R };
+        AsyncResource?: new (type: string) => { runInAsyncScope<R>(fn: () => R): R };
+      }
+    | undefined;
+  const snapshot = hooks?.AsyncLocalStorage?.snapshot?.();
+  const Resource = hooks?.AsyncResource;
+  // Each event in a resource of its own, made in the snapshot: it inherits the snapshot's stores,
+  // and a store an event sets (`run` writes it on the current resource, before Node's
+  // AsyncContextFrame) stays its own instead of reaching the next event started in the snapshot.
+  if (snapshot && Resource)
+    return (fn) => snapshot(() => new Resource("cellb.event").runInAsyncScope(fn));
+  type State = { root: unknown; current: unknown };
+  const realm = globalThis as Record<symbol, State | undefined>;
+  const captured = realm[POLYFILL_STATE]?.current;
+  return (fn) => {
+    const state = realm[POLYFILL_STATE];
+    if (!state) return fn();
+    const prev = state.current;
+    state.current = captured ?? state.root;
+    try {
+      return fn();
+    } finally {
+      state.current = prev;
+    }
+  };
+}

packages/cellb/src/core/object-host.ts

@@ -9,6 +9,7 @@ import type {
   SyncSqlDriver,
   ValueCodec,
 } from "../ports.js";
+import type { AsyncRoot } from "./async-context.js";
 import { CELLD_SCHEMA, LOG_SEQ_READ, LOG_SEQ_WRITE, LOG_TABLE, scopeOf } from "./celld-schema.js";
 import { InputGate, OutputGate } from "./gates.js";
 import {
@@ -93,6 +94,8 @@ export class ObjectHost {
     /** A facet: on open its rows move to this run's scope, as celld does. */
     facet?: boolean;
     acquire?: Lease["tryAcquire"];
+    /** The async context events start in (`asyncRoot()`): not the caller's, as on Cloudflare. */
+    asyncRoot?: AsyncRoot;
   }) {
     this.#backup = opts.backup;
     this.cellKey = opts.cellKey;
@@ -174,7 +177,9 @@ export class ObjectHost {
         return facet;
       },
     );
-    const construct = () => (this.#instance = new opts.cls(state, opts.env));
+    const root = opts.asyncRoot ?? ((fn) => fn());
+    this.#root = root;
+    const construct = () => root(() => (this.#instance = new opts.cls(state, opts.env)));
     this.#construct = construct;
     const load = opts.backup?.load;
     const replay = opts.backup?.replay;
@@ -265,7 +270,7 @@ export class ObjectHost {
         let started: Promise<T>;
         this.#active++;
         try {
-          started = Promise.resolve(fn(this.#instance, this.state));
+          started = Promise.resolve(this.#root(() => fn(this.#instance, this.state)));
         } catch (e) {
           started = Promise.reject(e);
         }
@@ -410,6 +415,7 @@ export class ObjectHost {
   }
 
   #construct?: () => unknown;
+  #root: AsyncRoot = (fn) => fn();
   readonly #open: () => void;
   #epoch: number | undefined;
 

packages/cellb/src/core/runtime.ts

@@ -11,6 +11,7 @@ import { v8Codec, v8CodecWith } from "../v8/codec.js";
 import { type CfWebSocket, takeClientPort } from "../ws/server.js";
 import { withUpgrade } from "../ws/upgrade.js";
 import { type AlarmEntry, AlarmScheduler } from "./alarms.js";
+import { asyncRoot } from "./async-context.js";
 import type { ExecutionContext } from "./durable-object.js";
 import { ctxExports, type DurableObjectClass, storedExports } from "./handles.js";
 import { DurableObjectId, namespaceKey } from "./id.js";
@@ -127,6 +128,8 @@ export function createCellRuntime(opts: CellRuntimeOptions): CellRuntime {
   const codec = opts.codec ?? v8Codec;
   const hosts = new Map<string, ObjectHost>();
   const background = new Set<Promise<unknown>>();
+  // Events start in the context the runtime was made in, not their caller's (Cloudflare).
+  const eventsRoot = asyncRoot();
   const waitUntil = (p: Promise<unknown>) => {
     const t = p.catch(() => {}).finally(() => background.delete(t));
     background.add(t);
@@ -232,6 +235,7 @@ export function createCellRuntime(opts: CellRuntimeOptions): CellRuntime {
           onAlarmChange: (at) => alarms.set({ cellKey, className, idHex: id.toString() }, at),
           exports: exportsObj,
           facet: opts.facets,
+          asyncRoot: eventsRoot,
           acquire: ownership ? (key) => ownership.tryAcquire(key) : undefined,
           sqlExtensions: opts.sqlExtensions,
           deleteAllPreservesAlarm: opts.deleteAllPreservesAlarm,

packages/cellb/src/core/sql-split.ts

@@ -55,3 +55,15 @@ export function completeSqlStatements(sql: string): { statements: string[]; rema
   }
   return { statements, remainder: sql.slice(rest) };
 }
+
+const COMMENTS = /--[^\n]*(\n|$)|\/\*[\s\S]*?(\*\/|$)/g;
+
+/**
+ * Whether `sql` holds no statement to run (workerd and celld: "SQL code did not contain a
+ * statement"): nothing but whitespace, comments and `;`, or a last statement of comments alone.
+ */
+export function lacksStatement(sql: string): boolean {
+  if (sql.replace(COMMENTS, " ").replace(/;/g, " ").trim() === "") return true;
+  const { remainder } = completeSqlStatements(sql);
+  return remainder.trim() !== "" && remainder.replace(COMMENTS, " ").trim() === "";
+}

packages/cellb/src/core/sql-storage.ts

@@ -1,6 +1,6 @@
 import type { CommitChanges, SqlResult, SyncSqlDriver } from "../ports.js";
 import { LOG_SEQ_READ, LOG_SEQ_WRITE, LOG_TABLE } from "./celld-schema.js";
-import { completeSqlStatements } from "./sql-split.js";
+import { completeSqlStatements, lacksStatement } from "./sql-split.js";
 
 const FORBIDDEN_TXN = /^\s*(BEGIN|COMMIT|END|ROLLBACK|SAVEPOINT|RELEASE)\b/i;
 const FORBIDDEN_INTERNAL = /\b_cf_\w*/i;
@@ -168,6 +168,9 @@ export class SqlStorage {
     if (refused) throw new Error(`SQL error: prepare SQL cursor: ${refused}`);
     if (bindings.length && countParameters(query) !== bindings.length)
       throw new Error("SQL error: Wrong number of parameter bindings for SQL query.");
+    // Statement after statement, as workerd and celld prepare them: none at all, or a last one
+    // of comments alone, is refused.
+    if (lacksStatement(query)) throw new Error("SQL code did not contain a statement.");
     return new SqlStorageCursor<T>(this.internalExec(query, bindings));
   }
 

packages/cellb/test/async-context-worker.ts

@@ -2,9 +2,8 @@
  * A Worker that reads one AsyncLocalStorage store across every kind of continuation, in many
  * Durable Objects at once. Run on Node (the real node:async_hooks) and in a browser (cellb's
  * polyfill), it must answer the same: each object sees its own store everywhere, and the Worker's
- * fetch, outside any run(), sees none. The object it calls sees the CALLER's store: cellb starts a
- * called object's event in the caller's async context (both run in one realm). On Cloudflare that
- * event starts with no store; docs/gaps.md has the difference.
+ * fetch, outside any run(), sees none. The object it calls sees no store either: a call to an object
+ * is a new event, which starts in the runtime's own async context, as on Cloudflare.
  */
 export const ASYNC_CONTEXT_WORKER = `
 import { DurableObject, env } from "cloudflare:workers";
@@ -76,7 +75,7 @@ export function expectedSeen(name: string) {
     inTimeout: name,
     inMicrotask: name,
     inThen: name,
-    calleeSees: name, // Cloudflare: null (docs/gaps.md)
+    calleeSees: null, // a new event: not the caller's store
     afterRpc: name,
     inForAwait: name,
     inAll: name,

packages/cellb/test/p15-sql-ingest.test.ts

@@ -0,0 +1,90 @@
+import { describe, expect, it } from "vitest";
+import {
+  createCellRuntime,
+  DurableObject,
+  type DurableObjectNamespace,
+  runInDurableObject,
+} from "../src/index.js";
+import { nodeSqliteOpener } from "../src/node.js";
+
+// workerd's `sql.ingest()` (experimental): an SQL dump read in chunks, with errors worded as
+// celld words them (conformance project sql-ingest). The rest of the assertions are workerd's
+// own (conformance/workerd-tests/wd-sql/sql-test.js), run on cellb directly.
+class Box extends DurableObject {}
+
+async function withSql(fn: (sql: any) => void | Promise<void>) {
+  const runtime = createCellRuntime({
+    worker: { Box, default: {} },
+    durableObjects: { BOX: { className: "Box" } },
+    openSql: nodeSqliteOpener(),
+  });
+  try {
+    const stub = (runtime.env.BOX as DurableObjectNamespace).getByName("a");
+    await runInDurableObject(stub as never, async (_i, ctx) => fn(ctx.storage.sql));
+  } finally {
+    await runtime.close();
+  }
+}
+
+describe("P15 sql.ingest", () => {
+  it("runs the complete statements and returns the rest", () =>
+    withSql((sql) => {
+      expect(sql.ingest("SELECT 123; SELECT 456;    ").remainder).toBe("    ");
+      expect(sql.ingest("SELECT 123; SELECT 456;").remainder).toBe("");
+      for (const tail of [" SELECT 456", " SELECT 45", " SELECT ", " SEL", " "])
+        expect(sql.ingest(`SELECT 123;${tail}`).remainder).toBe(tail);
+      expect(sql.ingest("SELECT 123").remainder).toBe("SELECT 123");
+      // A trailing comment is a remainder for ingest, and no statement for exec.
+      expect(sql.ingest("SELECT 123; SELECT 456; -- trailing comment").remainder).toBe(
+        " -- trailing comment",
+      );
+      expect(() => sql.exec("SELECT 123; SELECT 456; -- trailing comment")).toThrow(
+        /SQL code did not contain a statement/,
+      );
+      expect(() => sql.exec("")).toThrow(/SQL code did not contain a statement/);
+      expect(() => sql.exec(";")).toThrow(/SQL code did not contain a statement/);
+    }));
+
+  it("throws on a complete statement that does not parse, as celld words it", () =>
+    withSql((sql) => {
+      expect(() => sql.ingest("SELECT * bunk;")).toThrow(
+        /^SQL error: prepare ingested SQL: near "bunk": syntax error/,
+      );
+      expect(() => sql.ingest("INSER INTO xyz VALUES ('a'),('b');")).toThrow(
+        /near "INSER": syntax error/,
+      );
+      expect(() => sql.ingest("INSERT INTO xyz VALUES ('a')('b');")).toThrow(
+        /near "\(": syntax error/,
+      );
+    }));
+
+  it("ingests a dump split into chunks of any size, strings and triggers included", () =>
+    withSql((sql) => {
+      sql.exec("CREATE TABLE streaming(val TEXT)");
+      sql.exec("CREATE TABLE log(n INTEGER)");
+      const dump = [
+        "CREATE TRIGGER IF NOT EXISTS t AFTER INSERT ON streaming BEGIN INSERT INTO log VALUES (1); END;",
+        ...Array.from(
+          { length: 6 },
+          (_, i) => `INSERT INTO streaming VALUES ('a;${i}'), ("b -- ${i}"), ('ü${i}');`,
+        ),
+        "/* done */",
+      ].join("\n");
+      for (const size of [1, 3, 9, 27, 81]) {
+        sql.exec("DELETE FROM streaming");
+        let buffer = "";
+        let rows = 0;
+        let statements = 0;
+        for (let i = 0; i < dump.length; i += size) {
+          const r = sql.ingest(buffer + dump.slice(i, i + size));
+          buffer = r.remainder;
+          rows += r.rowsWritten;
+          statements += r.statementCount;
+        }
+        expect(buffer.trim()).toBe("/* done */");
+        expect(sql.exec("SELECT count(*) AS n FROM streaming").one().n).toBe(18);
+        expect(statements).toBe(7); // the trigger and the six inserts
+        expect(rows).toBeGreaterThanOrEqual(18);
+      }
+    }));
+});

packages/cellb/test/p16-async-context.test.ts

@@ -0,0 +1,53 @@
+import { AsyncLocalStorage } from "node:async_hooks";
+import { describe, expect, it } from "vitest";
+import { createCellRuntime, DurableObject, type DurableObjectNamespace } from "../src/index.js";
+import { nodeSqliteOpener } from "../src/node.js";
+
+// On Cloudflare a call to a Durable Object is a new event: the callee sees no AsyncLocalStorage
+// store its caller set. cellb starts every event (and the constructor) in the context the
+// runtime was made in.
+const request = new AsyncLocalStorage<string>();
+const deployment = new AsyncLocalStorage<string>();
+
+class Probe extends DurableObject {
+  constructedWith = { request: request.getStore(), deployment: deployment.getStore() };
+  seen() {
+    return {
+      request: request.getStore() ?? null,
+      deployment: deployment.getStore() ?? null,
+      constructedWith: this.constructedWith,
+    };
+  }
+  async callOther(name: string) {
+    // A store set in this object does not reach the object it calls either.
+    return request.run("set-in-caller-object", () =>
+      (this.env as { PROBE: DurableObjectNamespace }).PROBE.getByName(name).seen(),
+    );
+  }
+}
+
+describe("P16 every event starts in the runtime's own async context", () => {
+  it("a called object sees the runtime's stores, not its caller's", async () => {
+    const runtime = deployment.run("deployment-1", () =>
+      createCellRuntime({
+        worker: { Probe, default: {} },
+        durableObjects: { PROBE: { className: "Probe" } },
+        openSql: nodeSqliteOpener(),
+      }),
+    );
+    try {
+      const ns = runtime.env.PROBE as DurableObjectNamespace;
+      const direct = await request.run("set-by-caller", () => (ns.getByName("a") as any).seen());
+      expect(direct).toEqual({
+        request: null,
+        deployment: "deployment-1",
+        constructedWith: { request: undefined, deployment: "deployment-1" },
+      });
+      const viaObject = await (ns.getByName("b") as any).callOther("c");
+      expect(viaObject.request).toBeNull();
+      expect(viaObject.deployment).toBe("deployment-1");
+    } finally {
+      await runtime.close();
+    }
+  });
+});