From 8025c414d102e1658ddd61e9d79626e9a105b105 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Tue, 27 Oct 2026 08:54:53 +0000 Subject: [PATCH] feat(chord): SqliteExecutor.runCounted resolves with the rows a statement changed The Node driver takes the count from node:sqlite; the Durable Object driver reads SELECT changes() after the statement, since the cursor's rowsWritten also counts index writes (verified in workerd). run is unchanged and costs nothing more. --- packages/chord/CHANGELOG.md | 1 + packages/chord/src/sqlite/cloudflare.ts | 12 +++++++ packages/chord/src/sqlite/database.ts | 7 ++-- packages/chord/src/sqlite/node.ts | 4 +++ packages/chord/test/sqlite/cloudflare.test.ts | 11 ++++++- packages/chord/test/sqlite/helpers.ts | 32 +++++++++++++++++++ packages/chord/test/sqlite/node.test.ts | 14 ++++++++ .../chord/test/telemetry-sqlite-store.test.ts | 1 + packages/durable/CHANGELOG.md | 1 + packages/durable/test-workers/storage.test.ts | 19 +++++++++++ packages/durable/test/sqlite-facade.test.ts | 4 +++ 11 files changed, 103 insertions(+), 3 deletions(-) diff --git a/packages/chord/CHANGELOG.md b/packages/chord/CHANGELOG.md index 584aaddd0..28fd35e9d 100644 --- a/packages/chord/CHANGELOG.md +++ b/packages/chord/CHANGELOG.md @@ -6,6 +6,7 @@ - `traceIn(options, fn, context)` takes the context last (was `traceIn(context, options, fn)`), like the rest of the API. - Watch listeners take one change object before the context: `WatchHandle.start((change, context) => …)` with `change: WatchChange` = `{ value, ops, retired }`, and `NativeWatchHandle.start(({ value }, context) => …)` with `NativeWatchChange`. A change is plain data, those fields only; the frame's context is the second argument. `Watch.advance(change, context)` takes the same object, and a frame delivered alone, with nothing folded into it, reaches the listener as the object the source passed. Migration: `start(async (value, ops, ctx) => …)` becomes `start(async ({ value, ops }, ctx) => …)`, `start(async (value, ctx) => …)` on a native watch `start(async ({ value }, ctx) => …)`, and `advance(value, ops, ctx)` becomes `advance({ value, ops, retired: false }, ctx)`. +- `SqliteExecutor.runCounted(sql, ...params)` runs one statement like `run` and resolves with the rows it inserted, updated or deleted, as SQLite's `changes()` counts them (index writes and rows changed by triggers do not count). The Node driver takes it from `node:sqlite`; the Durable Object driver reads `SELECT changes()` after the statement (the cursor's `rowsWritten` also counts index writes), so `run` stays as cheap as before. Migration: custom `SqliteDatabase` and `SqliteExecutor` implementations add `runCounted`. - Retirement is a flag, not an end: `WatchEnd` loses `retired`, and `Watch.retire(context)` is removed. A source reports a retirement with `advance({ value, ops, retired: true }, context)`, whose value is what the address shows after it, and the watch goes on: the next call has `retired: true`, folded with every other pending frame, so a document created at the address afterwards arrives in the same or a later call as `[["r", value]]`. `endAfterDelivery()` tells a watch its source sends nothing more: it takes no more frames, delivers what is pending, and ends `ended` after its last call returned, at once when nothing is pending and no call runs; it stays attached until then, so the source's `end()` (such as at its close), `stop()`, cancellation, or a throwing listener still end it first. Migration: a listener that waited for `closed` to give `retired` or tested for a `null` value tests `change.retired`; a source that called `retire(ctx)` calls `advance({ value, ops, retired: true }, ctx)`, then `endAfterDelivery()` when its owner ends with the retirement. ### Added diff --git a/packages/chord/src/sqlite/cloudflare.ts b/packages/chord/src/sqlite/cloudflare.ts index 897a660b7..29d4345d0 100644 --- a/packages/chord/src/sqlite/cloudflare.ts +++ b/packages/chord/src/sqlite/cloudflare.ts @@ -103,6 +103,14 @@ class DurableObjectSqliteExecutor implements SqliteExecutor { this.storage.sql.exec(sql, ...bind(params)); } + async runCounted(sql: string, ...params: SqliteValue[]): Promise { + this.check(); + this.storage.sql.exec(sql, ...bind(params)); + // The cursor's `rowsWritten` is a billing count that includes index writes; SQLite's own count is exact. + const counted = this.storage.sql.exec("SELECT changes() AS changes").next(); + return counted.done ? 0 : Number(counted.value.changes); + } + async get(sql: string, ...params: SqliteValue[]): Promise { this.check(); const result = this.storage.sql.exec(sql, ...bind(params)).next(); @@ -151,6 +159,10 @@ export class DurableObjectSqliteDatabase extends DurableObjectSqliteExecutor imp return this.queue.run(() => super.run(sql, ...params)); } + override runCounted(sql: string, ...params: SqliteValue[]): Promise { + return this.queue.run(() => super.runCounted(sql, ...params)); + } + override get(sql: string, ...params: SqliteValue[]): Promise { return this.queue.run(() => super.get(sql, ...params)); } diff --git a/packages/chord/src/sqlite/database.ts b/packages/chord/src/sqlite/database.ts index fc5a814c6..6a8e6715b 100644 --- a/packages/chord/src/sqlite/database.ts +++ b/packages/chord/src/sqlite/database.ts @@ -5,12 +5,15 @@ export type SqliteValue = null | number | bigint | string | Uint8Array; * Asynchronous SQL operations shared by a database and its transaction handles. * * `exec` runs SQL text without bindings and may contain several statements. `run`, `get`, and `all` - * execute one statement with positional bindings. Adapters may cache prepared statements by SQL text, - * so callers pass values as bindings instead of interpolating them. + * execute one statement with positional bindings. `runCounted` runs one like `run` and resolves with the rows it + * inserted, updated or deleted, as SQLite's `changes()` counts them (index writes and rows changed by triggers do not + * count); it may cost a statement more than `run`, so callers that need no count use `run`. Adapters may cache + * prepared statements by SQL text, so callers pass values as bindings instead of interpolating them. */ export interface SqliteExecutor { exec(sql: string): Promise; run(sql: string, ...params: SqliteValue[]): Promise; + runCounted(sql: string, ...params: SqliteValue[]): Promise; get(sql: string, ...params: SqliteValue[]): Promise; all(sql: string, ...params: SqliteValue[]): Promise; } diff --git a/packages/chord/src/sqlite/node.ts b/packages/chord/src/sqlite/node.ts index b156477e8..202c37c4f 100644 --- a/packages/chord/src/sqlite/node.ts +++ b/packages/chord/src/sqlite/node.ts @@ -93,6 +93,10 @@ abstract class NodeSqliteExecutor implements SqliteExecutor { }); } + runCounted(sql: string, ...params: SqliteValue[]): Promise { + return this.runOperation(() => Number(this.statement(sql).run(...params).changes)); + } + get(sql: string, ...params: SqliteValue[]): Promise { return this.runOperation(() => this.statement(sql).get(...params) as T | undefined); } diff --git a/packages/chord/test/sqlite/cloudflare.test.ts b/packages/chord/test/sqlite/cloudflare.test.ts index 0e7521d25..41a270524 100644 --- a/packages/chord/test/sqlite/cloudflare.test.ts +++ b/packages/chord/test/sqlite/cloudflare.test.ts @@ -1,7 +1,7 @@ import { DatabaseSync } from "node:sqlite"; import { describe, expect, it } from "vitest"; import { DurableObjectSqliteDatabase } from "../../src/sqlite/cloudflare.ts"; -import { durableObjectStorage } from "./helpers.ts"; +import { durableObjectStorage, RUN_CHANGES, runChanges } from "./helpers.ts"; describe("Durable Object SQLite driver", () => { it("binds and returns BLOB values as Uint8Array", async () => { @@ -14,6 +14,15 @@ describe("Durable Object SQLite driver", () => { }); }); + it("reports the rows a statement changed, without index writes", async () => { + const database = new DurableObjectSqliteDatabase(durableObjectStorage()); + expect(await runChanges(database)).toEqual(RUN_CHANGES); + const inTransaction = await database.transaction((handle) => + handle.runCounted("INSERT INTO changes_probe VALUES (?, ?, ?)", "t", "t/1", 1), + ); + expect(inTransaction).toBe(1); + }); + it("reads only the cursor's first row for get", async () => { const operations: string[] = []; const database = new DurableObjectSqliteDatabase( diff --git a/packages/chord/test/sqlite/helpers.ts b/packages/chord/test/sqlite/helpers.ts index 042696e45..48e855eab 100644 --- a/packages/chord/test/sqlite/helpers.ts +++ b/packages/chord/test/sqlite/helpers.ts @@ -1,5 +1,6 @@ import { DatabaseSync } from "node:sqlite"; import type { DurableObjectSqliteStorage } from "../../src/sqlite/cloudflare.ts"; +import type { SqliteExecutor } from "../../src/sqlite/index.ts"; type SqlValue = ArrayBuffer | string | number | null; @@ -63,3 +64,34 @@ export function durableObjectStorage( }, }; } + +/** + * The rows each kind of statement changes, as `runCounted` reports them: on a table with two indexes, so a driver that counted + * index writes (a Durable Object cursor's `rowsWritten`) would report more. + */ +export async function runChanges(database: SqliteExecutor): Promise> { + await database.exec( + "CREATE TABLE changes_probe (id TEXT PRIMARY KEY, path TEXT NOT NULL, n INTEGER NOT NULL) STRICT, WITHOUT ROWID", + ); + await database.exec("CREATE INDEX changes_probe_path ON changes_probe (path, id)"); + await database.exec("CREATE INDEX changes_probe_n ON changes_probe (n, id)"); + for (let i = 0; i < 10; i++) await database.run("INSERT INTO changes_probe VALUES (?, ?, ?)", `k${i}`, `a/${i}`, i); + const changes = (sql: string, ...params: (string | number)[]) => database.runCounted(sql, ...params); + return { + insert: await changes("INSERT INTO changes_probe VALUES (?, ?, ?)", "x", "a/x", 99), + update: await changes("UPDATE changes_probe SET path = ? WHERE id = ?", "b/x", "x"), + none: await changes("UPDATE changes_probe SET n = 0 WHERE id = ?", "absent"), + bulk: await changes("UPDATE changes_probe SET path = 'c/' || substr(path, 3) WHERE path >= 'a/' AND path < 'a0'"), + upsert: await changes( + "INSERT INTO changes_probe VALUES (?, ?, ?) ON CONFLICT (id) DO UPDATE SET n = excluded.n", + "x", + "b/x", + 5, + ), + delete: await changes("DELETE FROM changes_probe WHERE id = ?", "x"), + bulkDelete: await changes("DELETE FROM changes_probe WHERE path >= 'c/' AND path < 'c0'"), + }; +} + +/** What `runChanges` reports on every driver. */ +export const RUN_CHANGES = { insert: 1, update: 1, none: 0, bulk: 10, upsert: 1, delete: 1, bulkDelete: 10 }; diff --git a/packages/chord/test/sqlite/node.test.ts b/packages/chord/test/sqlite/node.test.ts index 11098d1c5..c7b72b073 100644 --- a/packages/chord/test/sqlite/node.test.ts +++ b/packages/chord/test/sqlite/node.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from "vitest"; import type { SqliteExecutor } from "../../src/sqlite/index.ts"; import { openNodeSqliteDatabase } from "../../src/sqlite/node.ts"; +import { RUN_CHANGES, runChanges } from "./helpers.ts"; describe("Node SQLite driver", () => { it("configures connection defaults and accepts explicit settings", async () => { @@ -18,6 +19,19 @@ describe("Node SQLite driver", () => { } }); + it("reports the rows a statement changed, without index writes", async () => { + const database = await openNodeSqliteDatabase(":memory:"); + try { + expect(await runChanges(database)).toEqual(RUN_CHANGES); + const inTransaction = await database.transaction((handle) => + handle.runCounted("INSERT INTO changes_probe VALUES (?, ?, ?)", "t", "t/1", 1), + ); + expect(inTransaction).toBe(1); + } finally { + await database.close(); + } + }); + it("commits work done through the transaction handle and closes idempotently", async () => { const database = await openNodeSqliteDatabase(":memory:"); await database.transaction(async (transaction) => { diff --git a/packages/chord/test/telemetry-sqlite-store.test.ts b/packages/chord/test/telemetry-sqlite-store.test.ts index 4da872546..34e142c57 100644 --- a/packages/chord/test/telemetry-sqlite-store.test.ts +++ b/packages/chord/test/telemetry-sqlite-store.test.ts @@ -59,6 +59,7 @@ describe("SQLite telemetry reads", () => { const counted: SqliteExecutor = { exec: (sql) => tx.exec(sql), run: (sql, ...params) => tx.run(sql, ...params), + runCounted: (sql, ...params) => tx.runCounted(sql, ...params), get: (sql: string, ...params: SqliteValue[]) => tx.get(sql, ...params), all: (sql: string, ...params: SqliteValue[]) => { queries++; diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index ed67bba7b..0f584aaed 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -4,6 +4,7 @@ ### Breaking Changes +- `SqliteExecutor` (re-exported from Chord) gains `runCounted(sql, ...params)`, resolving with the rows the statement changed; custom `SqliteDatabase` implementations add it. - `Storage.commit(writes, adopt, context)` takes an `adopt(seq)` callback, the Session's in-memory adoption and publication of the batch: a storage calls it exactly once, synchronously, once the batch is durable and before its next commit starts, then resolves with `seq`, and never for a commit that fails before the batch is durable. Memory, SQLite, JSONL, and IndexedDB storages do; custom storages and wrappers must call or forward it, and the storage conformance suite checks it. A commit that rejects after its adoption is published and fails the Session without rolling the adopted changes back. - `Storage.mintId()`'s contract is narrowed to IDs greater than every ID minted since the storage was opened and every ID stored, also across reopen. The bundled storages already behave so. - Watches use Chord's `WatchHandle` and `WatchEnd` (Durable re-exports both, with `WatchChange` and `NativeWatchChange`). End reasons are `stopped`, `cancelled`, `ended` (was `session_closed`), `listener_error`, and `failed` (was `session_failed`); `retired` is gone (see below). An ended watch drops its listener and every reference into its Session, so a stopped watch that is still referenced keeps only its last value. diff --git a/packages/durable/test-workers/storage.test.ts b/packages/durable/test-workers/storage.test.ts index 9f45ec600..70df76dea 100644 --- a/packages/durable/test-workers/storage.test.ts +++ b/packages/durable/test-workers/storage.test.ts @@ -52,3 +52,22 @@ describe("Durable Object SQLite schema", () => { }); }); }); + +describe("Durable Object SQLite driver in workerd", () => { + // The cursor's `rowsWritten` counts index writes too (3 for this insert, 20 for the bulk update); `runCounted` must not. + it("reports the rows a statement changed, without index writes", async () => { + await runInDurableObject(fresh(), async (_instance, state) => { + const database = new DurableObjectSqliteDatabase(state.storage); + await database.exec("CREATE TABLE probe (id TEXT PRIMARY KEY, path TEXT NOT NULL, n INTEGER NOT NULL) STRICT, WITHOUT ROWID"); + await database.exec("CREATE INDEX probe_path ON probe (path, id)"); + await database.exec("CREATE INDEX probe_n ON probe (n, id)"); + for (let i = 0; i < 10; i++) await database.run("INSERT INTO probe VALUES (?, ?, ?)", `k${i}`, `a/${i}`, i); + expect(await database.runCounted("INSERT INTO probe VALUES (?, ?, ?)", "x", "a/x", 99)).toBe(1); + expect(await database.runCounted("UPDATE probe SET n = 0 WHERE id = ?", "absent")).toBe(0); + expect( + await database.runCounted("UPDATE probe SET path = 'c/' || substr(path, 3) WHERE path >= 'a/' AND path < 'a0'"), + ).toBe(10); + expect(await database.runCounted("DELETE FROM probe WHERE path >= 'c/' AND path < 'c0'")).toBe(10); + }); + }); +}); diff --git a/packages/durable/test/sqlite-facade.test.ts b/packages/durable/test/sqlite-facade.test.ts index ab8b9da8e..7ba878edb 100644 --- a/packages/durable/test/sqlite-facade.test.ts +++ b/packages/durable/test/sqlite-facade.test.ts @@ -35,6 +35,10 @@ class ControlledSettlementDatabase implements SqliteDatabase { return this.delegate.run(sql, ...params); } + runCounted(sql: string, ...params: SqliteValue[]): Promise { + return this.delegate.runCounted(sql, ...params); + } + get(sql: string, ...params: SqliteValue[]): Promise { return this.delegate.get(sql, ...params); }