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

@@ -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<T>` = `{ value, ops, retired }`, and `NativeWatchHandle.start(({ value }, context) => …)` with `NativeWatchChange<T>`. 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

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<number> {
+		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<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T | undefined> {
 		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<number> {
+		return this.queue.run(() => super.runCounted(sql, ...params));
+	}
+
 	override get<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T | undefined> {
 		return this.queue.run(() => super.get<T>(sql, ...params));
 	}

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<void>;
 	run(sql: string, ...params: SqliteValue[]): Promise<void>;
+	runCounted(sql: string, ...params: SqliteValue[]): Promise<number>;
 	get<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T | undefined>;
 	all<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T[]>;
 }

packages/chord/src/sqlite/node.ts

@@ -93,6 +93,10 @@ abstract class NodeSqliteExecutor implements SqliteExecutor {
 		});
 	}
 
+	runCounted(sql: string, ...params: SqliteValue[]): Promise<number> {
+		return this.runOperation(() => Number(this.statement(sql).run(...params).changes));
+	}
+
 	get<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T | undefined> {
 		return this.runOperation(() => this.statement(sql).get(...params) as T | undefined);
 	}

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(

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<Record<string, number>> {
+	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 };

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) => {

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: <T extends object>(sql: string, ...params: SqliteValue[]) => tx.get<T>(sql, ...params),
 					all: <T extends object>(sql: string, ...params: SqliteValue[]) => {
 						queries++;

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.

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);
+		});
+	});
+});

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<number> {
+		return this.delegate.runCounted(sql, ...params);
+	}
+
 	get<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T | undefined> {
 		return this.delegate.get<T>(sql, ...params);
 	}