diff --git a/scripts/pr-lib/review.sh b/scripts/pr-lib/review.sh index a8dbd42234cc..550ca08b06d5 100644 --- a/scripts/pr-lib/review.sh +++ b/scripts/pr-lib/review.sh @@ -160,7 +160,10 @@ review_artifacts_init() { local meta_number head_sha meta_number=$(jq -r '.number' .local/pr-meta.json) head_sha=$(jq -r '.headRefOid' .local/pr-meta.json) - if [ "$meta_number" != "$pr" ] || ! printf '%s' "$head_sha" | rg -q '^[0-9a-f]{40}$'; then + # Bash regex, not rg: this guard runs inside fork-PR CI test harnesses on + # GitHub-hosted runners without ripgrep, where a missing rg (exit 127) would + # misreport a valid head SHA as an identity mismatch. + if [ "$meta_number" != "$pr" ] || ! [[ "$head_sha" =~ ^[0-9a-f]{40}$ ]]; then echo "Review artifacts init failed: .local/pr-meta.json describes PR #$meta_number at '$head_sha', not PR #$pr. Re-run: scripts/pr review-init $pr" exit 1 fi diff --git a/src/config/sessions/session-accessor.sqlite-message-cut.test.ts b/src/config/sessions/session-accessor.sqlite-message-cut.test.ts index af3c91827652..342522a4c4ee 100644 --- a/src/config/sessions/session-accessor.sqlite-message-cut.test.ts +++ b/src/config/sessions/session-accessor.sqlite-message-cut.test.ts @@ -1,4 +1,5 @@ import { afterEach, describe, expect, it, vi } from "vitest"; +import { trackSqliteStatementExecutions } from "../../../test/helpers/sqlite-statement-execution-counter.js"; import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"; import { closeOpenClawAgentDatabasesForTest, @@ -36,13 +37,13 @@ afterEach(() => { function trackFullTranscriptLoads(env: NodeJS.ProcessEnv): () => number { const database = openOpenClawAgentDatabase({ agentId, env }); - const prepare = vi.spyOn(database.db, "prepare"); - return () => - prepare.mock.calls.filter( - ([sql]) => - sql.includes('select "event_json" from "transcript_events"') && - sql.includes('order by "seq" asc'), - ).length; + const { counts } = trackSqliteStatementExecutions(database.db, ["loads"], (sqlText) => + sqlText.includes('select "event_json" from "transcript_events"') && + sqlText.includes('order by "seq" asc') + ? "loads" + : null, + ); + return () => counts.loads; } async function createSiblingSession(params: { diff --git a/src/gateway/server.node-pairing-memo.test.ts b/src/gateway/server.node-pairing-memo.test.ts index 342813adc089..420cc8d8498b 100644 --- a/src/gateway/server.node-pairing-memo.test.ts +++ b/src/gateway/server.node-pairing-memo.test.ts @@ -1,5 +1,6 @@ import { DatabaseSync } from "node:sqlite"; -import { afterAll, beforeAll, describe, expect, test, vi } from "vitest"; +import { afterAll, beforeAll, describe, expect, test } from "vitest"; +import { trackSqliteStatementExecutions } from "../../test/helpers/sqlite-statement-execution-counter.js"; import { listDevicePairing } from "../infra/device-pairing.js"; import { requestNodePairing } from "../infra/node-pairing.js"; import { configureSqliteConnectionPragmas } from "../infra/sqlite-wal.js"; @@ -44,23 +45,25 @@ describe("gateway node pairing memoization", () => { }); await seedNodeDevice("node-list-memo-scan-count"); const database = openOpenClawStateDatabase(); - const originalPrepare = database.db.prepare.bind(database.db); - const tableSelects = { paired: 0, pending: 0 }; - const prepareSpy = vi.spyOn(database.db, "prepare").mockImplementation((sql) => { - if (sql.includes('from "device_pairing_pending"')) { - tableSelects.pending += 1; - } - if (sql.includes('from "device_pairing_paired"')) { - tableSelects.paired += 1; - } - return originalPrepare(sql); - }); + const { counts: tableSelects, restore } = trackSqliteStatementExecutions( + database.db, + ["paired", "pending"], + (sql) => { + if (sql.includes('from "device_pairing_pending"')) { + return "pending"; + } + if (sql.includes('from "device_pairing_paired"')) { + return "paired"; + } + return null; + }, + ); try { expect((await rpcReq(ws, "node.list", {})).ok).toBe(true); expect((await rpcReq(ws, "node.list", {})).ok).toBe(true); expect(tableSelects).toEqual({ paired: 1, pending: 1 }); } finally { - prepareSpy.mockRestore(); + restore(); } } finally { ws.close(); diff --git a/src/infra/kysely-sync.test.ts b/src/infra/kysely-sync.test.ts index 8e674fd285e3..20cdb88b6b53 100644 --- a/src/infra/kysely-sync.test.ts +++ b/src/infra/kysely-sync.test.ts @@ -1,10 +1,12 @@ // Covers the compile-only Kysely facade used by sync node:sqlite helpers. -import { DatabaseSync } from "node:sqlite"; -import type { Generated } from "kysely"; +import { spawnSync } from "node:child_process"; +import { constants, DatabaseSync } from "node:sqlite"; +import { sql, type Generated } from "kysely"; import { afterEach, describe, expect, it } from "vitest"; import { withTestTimeout } from "../../test/helpers/promise.js"; import { clearNodeSqliteKyselyCacheForDatabase, + enableNodeSqliteKyselyStatementCache, executeSqliteQuerySync, executeSqliteQueryTakeFirstSync, getNodeSqliteKysely, @@ -59,6 +61,336 @@ describe("kysely sync helpers", () => { ]); }); + it("returns identical results while repeated statements move from cold to warm", () => { + database = new DatabaseSync(":memory:"); + database.exec( + "create table items (id integer primary key autoincrement, name text not null unique)", + ); + const db = getNodeSqliteKysely(database); + const prepares = countPrepares(database); + + const insertResults = ["Ada", "Grace", "Lin"].map((name) => + executeSqliteQuerySync(database!, db.insertInto("items").values({ name })), + ); + expect(insertResults).toEqual([ + { insertId: 1n, numAffectedRows: 1n, rows: [] }, + { insertId: 2n, numAffectedRows: 1n, rows: [] }, + { insertId: 3n, numAffectedRows: 1n, rows: [] }, + ]); + expect(prepares.calls()).toBe(2); + + const select = db.selectFrom("items").selectAll().orderBy("id"); + const expectedRows = [ + { id: 1, name: "Ada" }, + { id: 2, name: "Grace" }, + { id: 3, name: "Lin" }, + ]; + expect(executeSqliteQuerySync(database, select).rows).toEqual(expectedRows); + expect(executeSqliteQuerySync(database, select).rows).toEqual(expectedRows); + expect(executeSqliteQuerySync(database, select).rows).toEqual(expectedRows); + expect(prepares.calls()).toBe(4); + + const conflictingInsert = db.insertInto("items").values({ id: 1, name: "Duplicate" }); + const errors = Array.from({ length: 3 }, () => { + try { + executeSqliteQuerySync(database!, conflictingInsert); + return null; + } catch (error) { + expect(error).toBeInstanceOf(Error); + return { message: (error as Error).message, name: (error as Error).name }; + } + }); + expect(errors[0]).not.toBeNull(); + expect(errors[1]).toEqual(errors[0]); + expect(errors[2]).toEqual(errors[0]); + expect(prepares.calls()).toBe(6); + }); + + it("admits only repeated SQL and bounds the prepared statement working set", () => { + database = new DatabaseSync(":memory:"); + database.exec("create table items (id integer primary key, name text not null)"); + const db = getNodeSqliteKysely(database); + const prepares = countPrepares(database); + const variableSelect = (parameterCount: number) => + db + .selectFrom("items") + .selectAll() + .where( + "id", + "not in", + Array.from({ length: parameterCount }, (_, index) => index + 1), + ); + + for (let parameterCount = 1; parameterCount <= 64; parameterCount += 1) { + expect(executeSqliteQuerySync(database, variableSelect(parameterCount)).rows).toEqual([]); + expect(executeSqliteQuerySync(database, variableSelect(parameterCount)).rows).toEqual([]); + } + expect(prepares.calls()).toBe(128); + + for (let parameterCount = 33; parameterCount <= 64; parameterCount += 1) { + expect(executeSqliteQuerySync(database, variableSelect(parameterCount)).rows).toEqual([]); + } + expect(prepares.calls()).toBe(128); + + for (let parameterCount = 1; parameterCount <= 32; parameterCount += 1) { + expect(executeSqliteQuerySync(database, variableSelect(parameterCount)).rows).toEqual([]); + } + expect(prepares.calls()).toBe(160); + }); + + it("does not retain one-shot variable-cardinality SQL statements", () => { + database = new DatabaseSync(":memory:"); + database.exec("create table items (id integer primary key, name text not null)"); + const db = getNodeSqliteKysely(database); + const prepares = countPrepares(database); + const runVariableSelects = () => { + for (let parameterCount = 1; parameterCount <= 64; parameterCount += 1) { + const ids = Array.from({ length: parameterCount }, (_, index) => index + 1); + const select = db.selectFrom("items").selectAll().where("id", "not in", ids); + expect(executeSqliteQuerySync(database!, select).rows).toEqual([]); + } + }; + + runVariableSelects(); + runVariableSelects(); + runVariableSelects(); + expect(prepares.calls()).toBe(192); + }); + + it("keeps nested lazy iterations independent", () => { + database = new DatabaseSync(":memory:"); + database.exec( + "create table items (id integer primary key autoincrement, name text not null unique)", + ); + const db = getNodeSqliteKysely(database); + for (const name of ["Ada", "Grace", "Lin"]) { + executeSqliteQuerySync(database, db.insertInto("items").values({ name })); + } + const select = db.selectFrom("items").selectAll().orderBy("id"); + const prepares = countPrepares(database); + + expect([...iterateSqliteQuerySync(database, select)]).toHaveLength(3); + expect([...iterateSqliteQuerySync(database, select)]).toHaveLength(3); + + const outer = iterateSqliteQuerySync(database, select); + expect(outer.next()).toEqual({ done: false, value: { id: 1, name: "Ada" } }); + expect([...iterateSqliteQuerySync(database, select)]).toEqual([ + { id: 1, name: "Ada" }, + { id: 2, name: "Grace" }, + { id: 3, name: "Lin" }, + ]); + expect([...outer]).toEqual([ + { id: 2, name: "Grace" }, + { id: 3, name: "Lin" }, + ]); + expect(prepares.calls()).toBe(4); + }); + + it("does not reuse an active cached statement during synchronous callback re-entry", () => { + database = new DatabaseSync(":memory:"); + let nested = false; + const db = getNodeSqliteKysely(database); + const query = db.selectNoFrom( + /* kysely-allow-raw: cache re-entry test needs a synthetic UDF call, not a store column. */ + sql`nested_value()`.as("value"), + ); + database.function("nested_value", () => { + if (nested) { + return 1; + } + nested = true; + try { + return executeSqliteQueryTakeFirstSync(database!, query)!.value + 1; + } finally { + nested = false; + } + }); + const prepares = countPrepares(database); + + expect(executeSqliteQuerySync(database, query).rows).toEqual([{ value: 2 }]); + expect(executeSqliteQuerySync(database, query).rows).toEqual([{ value: 2 }]); + expect(executeSqliteQuerySync(database, query).rows).toEqual([{ value: 2 }]); + expect(prepares.calls()).toBe(4); + }); + + it("invalidates prepared statements when the SQLite authorizer changes", () => { + database = new DatabaseSync(":memory:"); + if (typeof database.setAuthorizer !== "function") { + return; + } + database.exec("create table items (id integer primary key, name text not null)"); + database.exec("insert into items (id, name) values (1, 'Ada')"); + const db = getNodeSqliteKysely(database); + const select = db.selectFrom("items").selectAll(); + const prepares = countPrepares(database); + + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: 1, name: "Ada" }]); + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: 1, name: "Ada" }]); + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: 1, name: "Ada" }]); + expect(prepares.calls()).toBe(2); + + database.setAuthorizer((actionCode, tableName) => + actionCode === constants.SQLITE_READ && tableName === "items" + ? constants.SQLITE_IGNORE + : constants.SQLITE_OK, + ); + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: null, name: null }]); + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: null, name: null }]); + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: null, name: null }]); + expect(prepares.calls()).toBe(5); + + database.setAuthorizer(null); + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: 1, name: "Ada" }]); + expect(prepares.calls()).toBe(6); + }); + + it("does not cache while a dynamic SQLite authorizer is installed", () => { + database = new DatabaseSync(":memory:"); + if (typeof database.setAuthorizer !== "function") { + return; + } + database.exec("create table items (id integer primary key, name text not null)"); + database.exec("insert into items (id, name) values (1, 'Ada')"); + const db = getNodeSqliteKysely(database); + const select = db.selectFrom("items").selectAll(); + const prepares = countPrepares(database); + let allow = true; + database.setAuthorizer(() => (allow ? constants.SQLITE_OK : constants.SQLITE_DENY)); + + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: 1, name: "Ada" }]); + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: 1, name: "Ada" }]); + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: 1, name: "Ada" }]); + expect(prepares.calls()).toBe(3); + + allow = false; + expect(() => executeSqliteQuerySync(database!, select)).toThrow(/not authorized/iu); + expect(prepares.calls()).toBe(4); + }); + + it("does not retain oversized statement parameters", () => { + database = new DatabaseSync(":memory:"); + const db = getNodeSqliteKysely(database); + const lengthOf = (value: string) => + db.selectNoFrom( + /* kysely-allow-raw: parameter-retention test measures binding size on a scalar, schema-free query. */ + sql`length(${value})`.as("value"), + ); + const prepares = countPrepares(database); + + expect(executeSqliteQuerySync(database, lengthOf("small")).rows).toEqual([{ value: 5 }]); + expect(executeSqliteQuerySync(database, lengthOf("small")).rows).toEqual([{ value: 5 }]); + expect(executeSqliteQuerySync(database, lengthOf("small")).rows).toEqual([{ value: 5 }]); + expect(prepares.calls()).toBe(2); + + const oversized = "x".repeat(64 * 1024 + 1); + expect(executeSqliteQuerySync(database, lengthOf(oversized)).rows).toEqual([ + { value: oversized.length }, + ]); + expect(prepares.calls()).toBe(3); + expect(executeSqliteQuerySync(database, lengthOf("small")).rows).toEqual([{ value: 5 }]); + expect(prepares.calls()).toBe(3); + + const oversizedSql = db.selectNoFrom( + /* kysely-allow-raw: admission-gate test needs an oversized SQL text, only reachable via raw. */ + sql.raw(`1 /*${"x".repeat(64 * 1024)}*/`).as("value"), + ); + expect(executeSqliteQuerySync(database, oversizedSql).rows).toEqual([{ value: 1 }]); + expect(executeSqliteQuerySync(database, oversizedSql).rows).toEqual([{ value: 1 }]); + expect(prepares.calls()).toBe(5); + }); + + it("invalidates prepared statements when the database is deserialized", () => { + database = new DatabaseSync(":memory:"); + database.exec("create table items (id integer primary key, name text not null)"); + database.exec("insert into items (id, name) values (1, 'Ada')"); + let replacementBytes = new Uint8Array(); + if (typeof database.deserialize === "function") { + const replacement = new DatabaseSync(":memory:"); + replacement.exec("create table items (id integer primary key, name text not null)"); + replacement.exec("insert into items (id, name) values (2, 'Grace')"); + replacementBytes = replacement.serialize(); + replacement.close(); + } else { + Object.defineProperty(database, "deserialize", { + configurable: true, + value(this: DatabaseSync): void { + this.exec("delete from items"); + this.exec("insert into items (id, name) values (2, 'Grace')"); + }, + }); + } + const db = getNodeSqliteKysely(database); + const select = db.selectFrom("items").selectAll(); + const prepares = countPrepares(database); + + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: 1, name: "Ada" }]); + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: 1, name: "Ada" }]); + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: 1, name: "Ada" }]); + expect(prepares.calls()).toBe(2); + + database.deserialize(replacementBytes); + + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: 2, name: "Grace" }]); + expect(prepares.calls()).toBe(3); + }); + + it("refreshes cached query metadata after a schema change", () => { + database = new DatabaseSync(":memory:"); + database.exec("create table items (id integer primary key, name text not null)"); + database.exec("insert into items (id, name) values (1, 'Ada')"); + const db = getNodeSqliteKysely(database); + const select = db.selectFrom("items").selectAll(); + const prepares = countPrepares(database); + + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: 1, name: "Ada" }]); + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: 1, name: "Ada" }]); + expect(executeSqliteQuerySync(database, select).rows).toEqual([{ id: 1, name: "Ada" }]); + expect(prepares.calls()).toBe(2); + + database.exec("alter table items add column note text not null default 'new'"); + + expect(executeSqliteQuerySync(database, select).rows).toEqual([ + { id: 1, name: "Ada", note: "new" }, + ]); + expect(prepares.calls()).toBe(2); + }); + + it("resets a cached row statement after a step-time error", () => { + database = new DatabaseSync(":memory:"); + const db = getNodeSqliteKysely(database); + const jsonValue = (value: string) => + db.selectNoFrom( + /* kysely-allow-raw: step-error test needs json() to fail at execution time, not prepare time. */ + sql`json(${value})`.as("value"), + ); + const prepares = countPrepares(database); + + expect(executeSqliteQuerySync(database, jsonValue("{}")).rows).toEqual([{ value: "{}" }]); + expect(executeSqliteQuerySync(database, jsonValue("{}")).rows).toEqual([{ value: "{}" }]); + expect(executeSqliteQuerySync(database, jsonValue("{}")).rows).toEqual([{ value: "{}" }]); + expect(prepares.calls()).toBe(2); + + expect(() => executeSqliteQuerySync(database!, jsonValue("{"))).toThrow(/malformed JSON/iu); + expect(executeSqliteQuerySync(database, jsonValue("[]")).rows).toEqual([{ value: "[]" }]); + expect(prepares.calls()).toBe(2); + }); + + it("allows databases and prepared statements to collect after lifecycle clearing", () => { + expect(runRetentionScenario({ clearBeforeDrop: true })).toEqual({ + databaseCollected: true, + statementsCollected: 2, + statementCount: 2, + }); + }); + + it("retains a cached database when lifecycle clearing is skipped", () => { + expect(runRetentionScenario({ clearBeforeDrop: false })).toEqual({ + databaseCollected: false, + statementsCollected: 1, + statementCount: 2, + }); + }); + it("keeps the builder facade compile-only and fails direct execution", async () => { database = new DatabaseSync(":memory:"); database.exec("create table items (id integer primary key, name text not null)"); @@ -89,6 +421,96 @@ describe("kysely sync helpers", () => { }); }); +function countPrepares(database: DatabaseSync): { calls: () => number } { + enableNodeSqliteKyselyStatementCache(database); + const originalPrepare = database.prepare.bind(database); + let calls = 0; + database.prepare = (sqlText, options) => { + calls += 1; + return originalPrepare(sqlText, options); + }; + return { calls: () => calls }; +} + +function runRetentionScenario(options: { clearBeforeDrop: boolean }): { + databaseCollected: boolean; + statementsCollected: number; + statementCount: number; +} { + const moduleUrl = new URL("./kysely-sync.ts", import.meta.url).href; + const cleanup = options.clearBeforeDrop + ? ` + clearNodeSqliteKyselyCacheForDatabase(database); + database.close(); + ` + : ""; + const script = ` + import { DatabaseSync } from "node:sqlite"; + import { + clearNodeSqliteKyselyCacheForDatabase, + enableNodeSqliteKyselyStatementCache, + executeSqliteQuerySync, + getNodeSqliteKysely, + } from ${JSON.stringify(moduleUrl)}; + + const waitForTurn = () => new Promise((resolve) => setImmediate(resolve)); + async function runScenario() { + let database = new DatabaseSync(":memory:"); + database.exec("create table items (id integer primary key, name text not null)"); + const databaseRef = new WeakRef(database); + const statementRefs = []; + const originalPrepare = DatabaseSync.prototype.prepare; + database.prepare = function (sql, prepareOptions) { + const statement = originalPrepare.call(this, sql, prepareOptions); + statementRefs.push(new WeakRef(statement)); + return statement; + }; + const db = getNodeSqliteKysely(database); + enableNodeSqliteKyselyStatementCache(database); + const select = db.selectFrom("items").selectAll().orderBy("id"); + executeSqliteQuerySync(database, select); + executeSqliteQuerySync(database, select); + executeSqliteQuerySync(database, select); + delete database.prepare; + ${cleanup} + database = undefined; + + for (let attempt = 0; attempt < 30; attempt += 1) { + await waitForTurn(); + globalThis.gc(); + } + return { + databaseCollected: databaseRef.deref() === undefined, + statementsCollected: statementRefs.filter((ref) => ref.deref() === undefined).length, + statementCount: statementRefs.length, + }; + } + + process.stdout.write(JSON.stringify(await runScenario())); + `; + const result = spawnSync( + process.execPath, + [ + "--disable-warning=ExperimentalWarning", + "--expose-gc", + "--import", + "tsx", + "--input-type=module", + "--eval", + script, + ], + { cwd: process.cwd(), encoding: "utf8", timeout: 20_000 }, + ); + + expect(result.stderr).toBe(""); + expect(result.status).toBe(0); + return JSON.parse(result.stdout) as { + databaseCollected: boolean; + statementsCollected: number; + statementCount: number; + }; +} + async function expectCompileOnlyRejection(promise: Promise): Promise { await expect( withTestTimeout(promise, 500, "timed out waiting for compile-only rejection"), diff --git a/src/infra/kysely-sync.ts b/src/infra/kysely-sync.ts index 5d3fc064e202..e4efbfb947a9 100644 --- a/src/infra/kysely-sync.ts +++ b/src/infra/kysely-sync.ts @@ -1,5 +1,5 @@ // Adapts node:sqlite sync database calls for Kysely-style query execution. -import type { DatabaseSync, SQLInputValue } from "node:sqlite"; +import type { DatabaseSync, SQLInputValue, StatementSync } from "node:sqlite"; import type { Compilable, CompiledQuery, Kysely, QueryResult } from "kysely"; import { InsertQueryNode, Kysely as KyselyInstance, SqliteDialect } from "kysely"; @@ -7,6 +7,32 @@ import { InsertQueryNode, Kysely as KyselyInstance, SqliteDialect } from "kysely // going through Kysely's async driver path. const kyselyByDatabase = new WeakMap>(); +// Cached statements retain their database. Every enabled owner must clear before +// close or drop; this required lifecycle pairing is not enforced automatically. +const statementCacheSymbol = Symbol("openclaw.kyselySyncStatementCache"); +const statementInvalidationSymbol = Symbol("openclaw.kyselySyncStatementInvalidation"); +const statementCacheEnabledSymbol = Symbol("openclaw.kyselySyncStatementCacheEnabled"); +const authorizerActiveSymbol = Symbol("openclaw.kyselySyncAuthorizerActive"); +// Bound SQL plus variable-size bindings to about 2 MiB per enabled database. +// Process-wide retention scales with open handles; repeated variable SQL can enter. +const statementCacheCapacity = 32; +const statementCacheEntryBytes = 64 * 1024; + +type SqliteAuthorizer = Parameters[0]; + +type StatementCache = { + statements: Map; + candidates: Set; + active: WeakSet; +}; + +type StatementCacheOwner = DatabaseSync & { + [statementCacheSymbol]?: StatementCache; + [statementInvalidationSymbol]?: true; + [statementCacheEnabledSymbol]?: true; + [authorizerActiveSymbol]?: boolean; +}; + const compileOnlySqliteDialect = new SqliteDialect({ // The lazy database factory leaves compilation usable while direct execution fails fast. database: async () => { @@ -28,30 +54,177 @@ export function getNodeSqliteKysely(db: DatabaseSync): Kysely): void { + try { + deserialize(...args); + } finally { + // Node finalizes all statements before attempting deserialization, + // including failed attempts, so no cached object remains usable. + delete this[statementCacheSymbol]; + } + }, + }); + } + Object.defineProperty(owner, statementInvalidationSymbol, { + configurable: true, + value: true, + }); +} + +/** + * Enable bounded statement caching for a lifecycle-owned database that has not + * installed an authorizer before this call. + */ +export function enableNodeSqliteKyselyStatementCache(db: DatabaseSync): void { + const owner = db as StatementCacheOwner; + installStatementInvalidation(owner); + owner[statementCacheEnabledSymbol] = true; +} + +function queryFitsStatementCache(sql: string, parameters: readonly SQLInputValue[]): boolean { + let bytes = Buffer.byteLength(sql); + if (bytes > statementCacheEntryBytes) { + return false; + } + for (const parameter of parameters) { + if (typeof parameter === "string") { + bytes += Buffer.byteLength(parameter); + } else if (ArrayBuffer.isView(parameter)) { + bytes += parameter.byteLength; + } + if (bytes > statementCacheEntryBytes) { + return false; + } + } + return true; +} + +function executeWithCachedStatement( + db: DatabaseSync, + sql: string, + parameters: readonly SQLInputValue[], + execute: (statement: StatementSync) => Result, +): Result { + const owner = db as StatementCacheOwner; + installStatementInvalidation(owner); + if ( + !owner[statementCacheEnabledSymbol] || + owner[authorizerActiveSymbol] || + !queryFitsStatementCache(sql, parameters) + ) { + return execute(db.prepare(sql)); + } + let cache = owner[statementCacheSymbol]; + if (!cache) { + cache = { + statements: new Map(), + candidates: new Set(), + active: new WeakSet(), + }; + Object.defineProperty(owner, statementCacheSymbol, { + configurable: true, + value: cache, + }); + } + + const cached = cache.statements.get(sql); + let statement: StatementSync; + if (cached && !cache.active.has(cached)) { + cache.statements.delete(sql); + cache.statements.set(sql, cached); + statement = cached; + } else { + // A user-defined SQLite callback can re-enter this helper synchronously. + // Prepare a temporary statement rather than reset the active outer query. + statement = db.prepare(sql); + if (!cached && cache.candidates.delete(sql)) { + cache.statements.set(sql, statement); + if (cache.statements.size > statementCacheCapacity) { + const oldestSql = cache.statements.keys().next().value; + if (oldestSql !== undefined) { + cache.statements.delete(oldestSql); + } + } + } else if (!cached) { + // Admit only on second use so variable placeholder counts cannot fill + // the native statement cache with one-shot SQL strings. + cache.candidates.add(sql); + if (cache.candidates.size > statementCacheCapacity) { + const oldestCandidate = cache.candidates.values().next().value; + if (oldestCandidate !== undefined) { + cache.candidates.delete(oldestCandidate); + } + } + } + } + + cache.active.add(statement); + try { + return execute(statement); + } finally { + cache.active.delete(statement); + } +} + /** Execute a compiled Kysely query synchronously against node:sqlite. */ function executeCompiledSqliteQuerySync( db: DatabaseSync, compiledQuery: CompiledQuery, ): QueryResult { - const statement = db.prepare(compiledQuery.sql); const parameters = compiledQuery.parameters as SQLInputValue[]; + return executeWithCachedStatement(db, compiledQuery.sql, parameters, (statement) => { + if (statement.columns().length > 0) { + // Node's all() snapshots the column count before SQLite can reprepare + // an expired statement. Eagerly consuming iterate() reads it after step. + const iterator = statement.iterate(...parameters); + try { + return { rows: [...iterator] as Row[] }; + } catch (error) { + try { + iterator.return?.(); + } catch { + // Preserve the step error if iterator cleanup itself fails. + } + throw error; + } + } - if (statement.columns().length > 0) { - return { rows: statement.all(...parameters) as Row[] }; - } - - const { changes, lastInsertRowid } = statement.run(...parameters); - const result: QueryResult = { - numAffectedRows: BigInt(changes), - rows: [], - }; - if (InsertQueryNode.is(compiledQuery.query) && changes > 0) { - return { - ...result, - insertId: BigInt(lastInsertRowid), + const { changes, lastInsertRowid } = statement.run(...parameters); + const result: QueryResult = { + numAffectedRows: BigInt(changes), + rows: [], }; - } - return result; + if (InsertQueryNode.is(compiledQuery.query) && changes > 0) { + return { + ...result, + insertId: BigInt(lastInsertRowid), + }; + } + return result; + }); } /** Compile and execute a Kysely query synchronously. */ @@ -68,6 +241,8 @@ export function* iterateSqliteQuerySync( query: Compilable, ): IterableIterator { const compiledQuery = query.compile(); + // Iterators keep statement state across yields. A private statement prevents + // nested iteration of identical SQL from resetting an earlier iterator. const statement = db.prepare(compiledQuery.sql); if (statement.columns().length === 0) { return; @@ -84,7 +259,10 @@ export function executeSqliteQueryTakeFirstSync( return executeSqliteQuerySync(db, query).rows[0]; } -/** Drop the cached Kysely facade for a DatabaseSync after close/test reset. */ +/** Drop cached Kysely state for a DatabaseSync before close/test reset. */ export function clearNodeSqliteKyselyCacheForDatabase(db: DatabaseSync): void { + // Delete the database-owned cache before close so statements release their + // native database backreferences instead of recreating the WeakMap leak. + delete (db as StatementCacheOwner)[statementCacheSymbol]; kyselyByDatabase.delete(db); } diff --git a/src/infra/sqlite-transaction.test.ts b/src/infra/sqlite-transaction.test.ts index 316df65b1101..9ebb1c177b6e 100644 --- a/src/infra/sqlite-transaction.test.ts +++ b/src/infra/sqlite-transaction.test.ts @@ -4,6 +4,7 @@ import path from "node:path"; import type { Readable } from "node:stream"; import { afterEach, describe, expect, it, vi } from "vitest"; import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js"; +import { getNodeSqliteKysely } from "./kysely-sync.js"; import { requireNodeSqlite } from "./node-sqlite.js"; import { runSqliteDeferredTransactionSync, @@ -191,6 +192,34 @@ describe("runSqliteImmediateTransactionSync", () => { expect(execCalls).toEqual(["BEGIN IMMEDIATE", "COMMIT", "ROLLBACK"]); }); + it("clears cached state before closing after a rollback failure", () => { + const execCalls: string[] = []; + const operationError = new Error("operation failed"); + const rollbackError = new Error("rollback failed"); + let facadeAtClose: unknown; + const db = { + exec(sql: string) { + execCalls.push(sql); + if (sql === "ROLLBACK") { + throw rollbackError; + } + }, + close() { + facadeAtClose = getNodeSqliteKysely(db); + }, + } as import("node:sqlite").DatabaseSync; + const facadeBeforeClose = getNodeSqliteKysely(db); + + expect(() => + runSqliteImmediateTransactionSync(db, () => { + throw operationError; + }), + ).toThrow(operationError); + expect(execCalls).toEqual(["BEGIN IMMEDIATE", "ROLLBACK"]); + expect(facadeAtClose).toBeDefined(); + expect(facadeAtClose).not.toBe(facadeBeforeClose); + }); + it("logs one structured warning for a terminal lock failure", () => { const execCalls: string[] = []; const logger = { warn: vi.fn() }; diff --git a/src/infra/sqlite-transaction.ts b/src/infra/sqlite-transaction.ts index b1cf52ecfa25..9984a7655955 100644 --- a/src/infra/sqlite-transaction.ts +++ b/src/infra/sqlite-transaction.ts @@ -1,6 +1,7 @@ // Provides SQLite transaction helpers with nested savepoints. import type { DatabaseSync } from "node:sqlite"; import { createSubsystemLogger, type SubsystemLogger } from "../logging/subsystem.js"; +import { clearNodeSqliteKyselyCacheForDatabase } from "./kysely-sync.js"; const transactionDepthByDatabase = new WeakMap(); @@ -197,6 +198,7 @@ function abortImmediateTransaction(db: DatabaseSync): void { // If rollback itself fails, close the handle so callers cannot keep using a // connection that may still hold an abandoned write transaction. try { + clearNodeSqliteKyselyCacheForDatabase(db); db.close(); } catch { // Preserve the original transaction error; close failure is secondary. diff --git a/src/state/openclaw-agent-db.ts b/src/state/openclaw-agent-db.ts index bfd5d118f8ec..0df882a70adc 100644 --- a/src/state/openclaw-agent-db.ts +++ b/src/state/openclaw-agent-db.ts @@ -2,7 +2,10 @@ import { existsSync } from "node:fs"; import path from "node:path"; import type { DatabaseSync } from "node:sqlite"; -import { clearNodeSqliteKyselyCacheForDatabase } from "../infra/kysely-sync.js"; +import { + clearNodeSqliteKyselyCacheForDatabase, + enableNodeSqliteKyselyStatementCache, +} from "../infra/kysely-sync.js"; import { openNodeSqliteDatabase } from "../infra/node-sqlite.js"; import type { SqliteFileGeneration } from "../infra/sqlite-file-generation.js"; import { @@ -306,6 +309,7 @@ export function openOpenClawAgentDatabase( // pressure the 65th open would otherwise fail before eviction could run. evictLruAgentDatabaseHandles(); const db = openNodeSqliteDatabase(pathname); + enableNodeSqliteKyselyStatementCache(db); openedDb = db; // Eviction churn must avoid schema/registry busy waits on the event loop while // reconcile workers hold write transactions on these same agent databases. @@ -339,6 +343,7 @@ export function openOpenClawAgentDatabase( return maintenance; } catch (err) { maintenance?.close(); + clearNodeSqliteKyselyCacheForDatabase(db); db.close(); if ( err instanceof Error && diff --git a/src/state/openclaw-state-db.ts b/src/state/openclaw-state-db.ts index 96b9c6c80e61..7c2c25019ae2 100644 --- a/src/state/openclaw-state-db.ts +++ b/src/state/openclaw-state-db.ts @@ -4,6 +4,7 @@ import path from "node:path"; import type { DatabaseSync } from "node:sqlite"; import { clearNodeSqliteKyselyCacheForDatabase, + enableNodeSqliteKyselyStatementCache, executeSqliteQuerySync, getNodeSqliteKysely, } from "../infra/kysely-sync.js"; @@ -281,6 +282,7 @@ export function repairOpenClawStateDatabaseSchema(options: OpenClawStateDatabase if (db.isOpen) { db.exec("PRAGMA foreign_keys = ON;"); } + clearNodeSqliteKyselyCacheForDatabase(db); db.close(); ensureOpenClawStatePermissions(pathname, env); } @@ -402,6 +404,7 @@ export async function openExistingOpenClawStateDatabaseReadOnly( } } catch (error) { try { + clearNodeSqliteKyselyCacheForDatabase(db); db.close(); } catch { // Preserve the verification failure that explains why the database was refused. @@ -424,6 +427,7 @@ export async function openExistingOpenClawStateDatabaseReadOnly( } try { if (wasOpen) { + clearNodeSqliteKyselyCacheForDatabase(db); db.close(); } } finally { @@ -520,6 +524,7 @@ export function openOpenClawStateDatabase( } ensureOpenClawStatePermissions(pathname, env); const db = openNodeSqliteDatabase(pathname); + enableNodeSqliteKyselyStatementCache(db); const walMaintenance = (() => { let maintenance: SqliteWalMaintenance | undefined; try { @@ -540,6 +545,7 @@ export function openOpenClawStateDatabase( return maintenance; } catch (err) { maintenance?.close(); + clearNodeSqliteKyselyCacheForDatabase(db); db.close(); if ( err instanceof Error && diff --git a/test/helpers/sqlite-statement-execution-counter.ts b/test/helpers/sqlite-statement-execution-counter.ts new file mode 100644 index 000000000000..e32a57263953 --- /dev/null +++ b/test/helpers/sqlite-statement-execution-counter.ts @@ -0,0 +1,36 @@ +import type { DatabaseSync, StatementSync } from "node:sqlite"; +import { vi } from "vitest"; +import { clearNodeSqliteKyselyCacheForDatabase } from "../../src/infra/kysely-sync.js"; + +/** + * Count SQLite query executions per caller-defined bucket. Prepared-statement + * caching (src/infra/kysely-sync.ts) reuses statements across calls, so + * counting `prepare` invocations undercounts; this wraps `iterate` on matching + * statements and clears the statement cache at attach so statements cached + * before the spy cannot bypass it. + */ +export function trackSqliteStatementExecutions( + db: DatabaseSync, + keys: readonly Key[], + classify: (sql: string) => Key | null, +): { counts: Record; restore: () => void } { + clearNodeSqliteKyselyCacheForDatabase(db); + const counts = Object.fromEntries(keys.map((key) => [key, 0])) as Record; + const originalPrepare = db.prepare.bind(db); + const prepareSpy = vi.spyOn(db, "prepare").mockImplementation((sqlText: string) => { + const statement = originalPrepare(sqlText); + const key = classify(sqlText); + if (key !== null) { + const originalIterate = statement.iterate.bind(statement) as ( + ...args: unknown[] + ) => ReturnType; + // iterate is overloaded, so the wrapper forwards untyped and casts back. + statement.iterate = ((...args: unknown[]) => { + counts[key] += 1; + return originalIterate(...args); + }) as StatementSync["iterate"]; + } + return statement; + }); + return { counts, restore: () => prepareSpy.mockRestore() }; +}