diff --git a/packages/cellb/conformance/scenarios.ts b/packages/cellb/conformance/scenarios.ts index 1b8fdc3..55f109e 100644 --- a/packages/cellb/conformance/scenarios.ts +++ b/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. diff --git a/packages/cellb/docs/gaps.md b/packages/cellb/docs/gaps.md index a7bcbef..188cf64 100644 --- a/packages/cellb/docs/gaps.md +++ b/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 diff --git a/packages/cellb/e2e/p11f-celld-one-bucket.test.ts b/packages/cellb/e2e/p11f-celld-one-bucket.test.ts index ca8abe9..8886cae 100644 --- a/packages/cellb/e2e/p11f-celld-one-bucket.test.ts +++ b/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//…): 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 diff --git a/packages/cellb/src/core/async-context.ts b/packages/cellb/src/core/async-context.ts new file mode 100644 index 0000000..e710964 --- /dev/null +++ b/packages/cellb/src/core/async-context.ts @@ -0,0 +1,45 @@ +/** Runs `fn` in an async context captured earlier. */ +export type AsyncRoot = (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?(): (fn: () => R) => R }; + AsyncResource?: new (type: string) => { runInAsyncScope(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; + 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; + } + }; +} diff --git a/packages/cellb/src/core/object-host.ts b/packages/cellb/src/core/object-host.ts index 3cd1d06..dc0ba50 100644 --- a/packages/cellb/src/core/object-host.ts +++ b/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; 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; diff --git a/packages/cellb/src/core/runtime.ts b/packages/cellb/src/core/runtime.ts index c1dc2ec..b1f1eab 100644 --- a/packages/cellb/src/core/runtime.ts +++ b/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(); const background = new Set>(); + // Events start in the context the runtime was made in, not their caller's (Cloudflare). + const eventsRoot = asyncRoot(); const waitUntil = (p: Promise) => { 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, diff --git a/packages/cellb/src/core/sql-split.ts b/packages/cellb/src/core/sql-split.ts index 52f231d..5d7658b 100644 --- a/packages/cellb/src/core/sql-split.ts +++ b/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() === ""; +} diff --git a/packages/cellb/src/core/sql-storage.ts b/packages/cellb/src/core/sql-storage.ts index 22b2fc1..33ec48a 100644 --- a/packages/cellb/src/core/sql-storage.ts +++ b/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(this.internalExec(query, bindings)); } diff --git a/packages/cellb/test/async-context-worker.ts b/packages/cellb/test/async-context-worker.ts index b051766..e4d437d 100644 --- a/packages/cellb/test/async-context-worker.ts +++ b/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, diff --git a/packages/cellb/test/p15-sql-ingest.test.ts b/packages/cellb/test/p15-sql-ingest.test.ts new file mode 100644 index 0000000..dc8477c --- /dev/null +++ b/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) { + 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); + } + })); +}); diff --git a/packages/cellb/test/p16-async-context.test.ts b/packages/cellb/test/p16-async-context.test.ts new file mode 100644 index 0000000..17f7e2e --- /dev/null +++ b/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(); +const deployment = new AsyncLocalStorage(); + +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(); + } + }); +});