diff --git a/src/infra/sqlite-integrity.ts b/src/infra/sqlite-integrity.ts index 5e67e6ba7078..87217a02d715 100644 --- a/src/infra/sqlite-integrity.ts +++ b/src/infra/sqlite-integrity.ts @@ -18,12 +18,25 @@ export function assertSqliteIntegrity( }; } +/** Require table and associated index consistency before trusting indexed reads. */ +export function assertSqliteTableIntegrity( + database: DatabaseSync, + databaseLabel: string, + tableName: string, +): void { + runSqliteCheck(database, `${databaseLabel} table ${tableName}`, "integrity_check", tableName); +} + function runSqliteCheck( database: DatabaseSync, databaseLabel: string, pragma: SqliteCheckPragma, + tableName?: string, ): "ok" { - const rows = database.prepare(`PRAGMA ${pragma};`).all() as Array>; + const argument = tableName ? `('${tableName.replaceAll("'", "''")}')` : ""; + const rows = database.prepare(`PRAGMA ${pragma}${argument};`).all() as Array< + Record + >; const results = rows.map((row) => row[pragma] ?? Object.values(row)[0]); if (results.length === 1 && results[0] === "ok") { return "ok"; diff --git a/src/infra/sqlite-transaction.test.ts b/src/infra/sqlite-transaction.test.ts index 3990216f44b9..316df65b1101 100644 --- a/src/infra/sqlite-transaction.test.ts +++ b/src/infra/sqlite-transaction.test.ts @@ -1,18 +1,20 @@ // Covers synchronous SQLite transaction helpers. import { spawn, type ChildProcessByStdio } from "node:child_process"; -import fs from "node:fs"; -import os from "node:os"; 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 { requireNodeSqlite } from "./node-sqlite.js"; -import { runSqliteImmediateTransactionSync } from "./sqlite-transaction.js"; +import { + runSqliteDeferredTransactionSync, + runSqliteImmediateTransactionSync, +} from "./sqlite-transaction.js"; const openDatabases: Array = []; type WriterLockChild = ChildProcessByStdio; const openChildren: WriterLockChild[] = []; -const tempDirs: string[] = []; +const tempDirs = useAutoCleanupTempDirTracker(afterEach); function createDatabase(): import("node:sqlite").DatabaseSync { const { DatabaseSync } = requireNodeSqlite(); @@ -92,12 +94,38 @@ afterEach(() => { db.close(); } } - for (const tempDir of tempDirs.splice(0)) { - fs.rmSync(tempDir, { recursive: true, force: true }); - } vi.restoreAllMocks(); }); +describe("runSqliteDeferredTransactionSync", () => { + it("keeps multiple reads on one snapshot while another connection commits", () => { + const tempDir = tempDirs.make("openclaw-sqlite-read-snapshot-"); + const databasePath = path.join(tempDir, "snapshot.sqlite"); + const { DatabaseSync } = requireNodeSqlite(); + const reader = new DatabaseSync(databasePath); + const writer = new DatabaseSync(databasePath); + openDatabases.push(reader, writer); + reader.exec( + "PRAGMA journal_mode = WAL; CREATE TABLE entries (id TEXT PRIMARY KEY); INSERT INTO entries VALUES ('first');", + ); + writer.exec("PRAGMA busy_timeout = 1000;"); + + const counts = runSqliteDeferredTransactionSync(reader, () => { + const before = reader.prepare("SELECT COUNT(*) AS count FROM entries").get() as { + count: number; + }; + writer.prepare("INSERT INTO entries(id) VALUES (?)").run("second"); + const after = reader.prepare("SELECT COUNT(*) AS count FROM entries").get() as { + count: number; + }; + return [before.count, after.count]; + }); + + expect(counts).toEqual([1, 1]); + expect(writer.prepare("SELECT COUNT(*) AS count FROM entries").get()).toEqual({ count: 2 }); + }); +}); + describe("runSqliteImmediateTransactionSync", () => { it("keeps outer writes when a nested savepoint rolls back", () => { const db = createDatabase(); @@ -257,8 +285,7 @@ describe("runSqliteImmediateTransactionSync", () => { it("waits for a separate writer and exposes the synchronous event-loop cost", async () => { const holdMs = 200; - const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-sqlite-contention-")); - tempDirs.push(tempDir); + const tempDir = tempDirs.make("openclaw-sqlite-contention-"); const databasePath = path.join(tempDir, "contention.sqlite"); const { DatabaseSync } = requireNodeSqlite(); const db = new DatabaseSync(databasePath); @@ -293,8 +320,7 @@ describe("runSqliteImmediateTransactionSync", () => { }); it("fails a real separate-writer wait after the single SQLite busy timeout", async () => { - const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-sqlite-timeout-")); - tempDirs.push(tempDir); + const tempDir = tempDirs.make("openclaw-sqlite-timeout-"); const databasePath = path.join(tempDir, "contention.sqlite"); const { DatabaseSync } = requireNodeSqlite(); const db = new DatabaseSync(databasePath); diff --git a/src/infra/sqlite-transaction.ts b/src/infra/sqlite-transaction.ts index eb1e16372c83..b1cf52ecfa25 100644 --- a/src/infra/sqlite-transaction.ts +++ b/src/infra/sqlite-transaction.ts @@ -25,6 +25,7 @@ export type SqliteTransactionOptions = { }; type SqliteTransactionStep = "begin" | "commit"; +type SqliteTransactionMode = "deferred" | "immediate"; function nextSavepointName(): string { nextSavepointId += 1; @@ -164,14 +165,15 @@ function execTimedTransactionStep(params: { } } -function beginImmediateTransaction( +function beginTransaction( db: DatabaseSync, options: SqliteTransactionOptions | undefined, + mode: SqliteTransactionMode, ): void { execTimedTransactionStep({ db, options, - sql: "BEGIN IMMEDIATE", + sql: mode === "immediate" ? "BEGIN IMMEDIATE" : "BEGIN", step: "begin", }); } @@ -214,9 +216,10 @@ function setTransactionDepth(db: DatabaseSync, depth: number): void { transactionDepthByDatabase.set(db, depth); } -export function runSqliteImmediateTransactionSync( +function runSqliteTransactionSync( db: DatabaseSync, operation: () => T, + mode: SqliteTransactionMode, options?: SqliteTransactionOptions, ): T { const depth = getTransactionDepth(db); @@ -241,7 +244,7 @@ export function runSqliteImmediateTransactionSync( } } - beginImmediateTransaction(db, options); + beginTransaction(db, options, mode); setTransactionDepth(db, 1); let transactionStillActive = true; let result: T; @@ -285,3 +288,20 @@ export function runSqliteImmediateTransactionSync( } } } + +/** Run synchronous reads against one deferred SQLite snapshot. */ +export function runSqliteDeferredTransactionSync( + db: DatabaseSync, + operation: () => T, + options?: SqliteTransactionOptions, +): T { + return runSqliteTransactionSync(db, operation, "deferred", options); +} + +export function runSqliteImmediateTransactionSync( + db: DatabaseSync, + operation: () => T, + options?: SqliteTransactionOptions, +): T { + return runSqliteTransactionSync(db, operation, "immediate", options); +} diff --git a/src/tasks/task-registry.store.sqlite.ts b/src/tasks/task-registry.store.sqlite.ts index 2889ef446235..d3c0c110255b 100644 --- a/src/tasks/task-registry.store.sqlite.ts +++ b/src/tasks/task-registry.store.sqlite.ts @@ -2,7 +2,9 @@ import type { DatabaseSync } from "node:sqlite"; import type { Insertable, Selectable } from "kysely"; import { executeSqliteQuerySync, getNodeSqliteKysely } from "../infra/kysely-sync.js"; +import { assertSqliteTableIntegrity } from "../infra/sqlite-integrity.js"; import { normalizeSqliteNumber } from "../infra/sqlite-number.js"; +import { runSqliteDeferredTransactionSync } from "../infra/sqlite-transaction.js"; import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js"; import { closeOpenClawStateDatabase, @@ -216,6 +218,20 @@ function selectTaskRows(db: DatabaseSync): TaskRegistryRow[] { return executeSqliteQuerySync(db, query).rows; } +function selectTaskRowsByOwnerKey(db: DatabaseSync, ownerKey: string): TaskRegistryRow[] { + const selectColumns = TASK_RUN_SELECT_COLUMNS.join(", "); + // This lookup gates duplicate media tasks. A table scan is intentional so a + // stale secondary index cannot hide an existing task between integrity checks. + return db + .prepare( + `SELECT ${selectColumns} + FROM task_runs NOT INDEXED + WHERE owner_key = ? + ORDER BY created_at ASC, task_id ASC`, + ) + .all(ownerKey) as TaskRegistryRow[]; +} + function selectTaskDeliveryStateRows(db: DatabaseSync): TaskDeliveryStateRow[] { const query = getTaskRegistryKysely(db) .selectFrom("task_delivery_state") @@ -316,13 +332,19 @@ function withWriteTransaction(write: (database: TaskRegistryDatabase) => void) { } export function loadTaskRegistryStateFromSqlite(): TaskRegistryStoreSnapshot { - const { db } = openTaskRegistryDatabase(); - const taskRows = selectTaskRows(db); - const deliveryRows = selectTaskDeliveryStateRows(db); - return { - tasks: new Map(taskRows.map((row) => [row.task_id, rowToTaskRecord(row)])), - deliveryStates: new Map(deliveryRows.map((row) => [row.task_id, rowToTaskDeliveryState(row)])), - }; + const { db, path } = openTaskRegistryDatabase(); + return runSqliteDeferredTransactionSync(db, () => { + assertSqliteTableIntegrity(db, path, "task_runs"); + assertSqliteTableIntegrity(db, path, "task_delivery_state"); + const taskRows = selectTaskRows(db); + const deliveryRows = selectTaskDeliveryStateRows(db); + return { + tasks: new Map(taskRows.map((row) => [row.task_id, rowToTaskRecord(row)])), + deliveryStates: new Map( + deliveryRows.map((row) => [row.task_id, rowToTaskDeliveryState(row)]), + ), + }; + }); } export function listTaskRegistryRecordsByOwnerKeyFromSqlite(ownerKey: string): TaskRecord[] { @@ -331,14 +353,7 @@ export function listTaskRegistryRecordsByOwnerKeyFromSqlite(ownerKey: string): T return []; } const { db } = openTaskRegistryDatabase(); - const query = getTaskRegistryKysely(db) - .selectFrom("task_runs") - .select(TASK_RUN_SELECT_COLUMNS) - .where("owner_key", "=", key) - .orderBy("created_at", "asc") - .orderBy("task_id", "asc"); - const rows = executeSqliteQuerySync(db, query).rows; - return rows.map(rowToTaskRecord); + return selectTaskRowsByOwnerKey(db, key).map(rowToTaskRecord); } export function saveTaskRegistryStateToSqlite(snapshot: TaskRegistryStoreSnapshot) { diff --git a/src/tasks/task-registry.store.test.ts b/src/tasks/task-registry.store.test.ts index 92070c8ab554..ca04dd93dffe 100644 --- a/src/tasks/task-registry.store.test.ts +++ b/src/tasks/task-registry.store.test.ts @@ -1,6 +1,7 @@ // Covers task registry store persistence, in-memory behavior, and observer notifications. import { statSync } from "node:fs"; import path from "node:path"; +import type { DatabaseSync } from "node:sqlite"; import { afterEach, describe, expect, it, vi } from "vitest"; import { emitAgentEvent } from "../infra/agent-events.js"; import { @@ -8,6 +9,8 @@ import { executeSqliteQueryTakeFirstSync, getNodeSqliteKysely, } from "../infra/kysely-sync.js"; +import { requireNodeSqlite } from "../infra/node-sqlite.js"; +import { readSqliteNumberPragma } from "../infra/sqlite-pragma.test-support.js"; import { resetLogger, setLoggerOverride } from "../logging/logger.js"; import { loggingState } from "../logging/state.js"; import { createWarnLogCapture } from "../logging/test-helpers/warn-log-capture.js"; @@ -111,6 +114,22 @@ function createStoredTask(): TaskRecord { }; } +function createUnsafeTaskOwnerIndex(database: DatabaseSync): void { + database.exec(` + DROP INDEX idx_task_runs_owner_key; + CREATE INDEX idx_task_runs_owner_key ON task_runs(status); + `); + database.enableDefensive?.(false); + database.exec("PRAGMA writable_schema = ON;"); + database + .prepare( + "UPDATE sqlite_schema SET sql = 'CREATE INDEX idx_task_runs_owner_key ON task_runs(owner_key)' WHERE name = 'idx_task_runs_owner_key'", + ) + .run(); + const schemaVersion = readSqliteNumberPragma(database, "schema_version"); + database.exec(`PRAGMA writable_schema = OFF; PRAGMA schema_version = ${schemaVersion + 1};`); +} + describe("task-registry store runtime", () => { afterEach(() => { ORIGINAL_ENV.restore(); @@ -317,6 +336,123 @@ describe("task-registry store runtime", () => { ); }); + it("loads task and delivery rows from one sqlite read snapshot", async () => { + await withOpenClawTestState( + { layout: "state-only", prefix: "openclaw-task-store-read-snapshot-" }, + async () => { + resetTaskRegistryForTests(); + const created = createTaskRecord({ + runtime: "acp", + ownerKey: "agent:main:main", + scopeKind: "session", + childSessionKey: "agent:main:acp:snapshot", + runId: "run-read-snapshot", + task: "Read one task registry snapshot", + status: "running", + deliveryStatus: "pending", + requesterOrigin: { + channel: "test-channel", + to: "C1234567890", + }, + }); + const database = openOpenClawStateDatabase(); + database.db + .prepare("UPDATE task_delivery_state SET last_notified_event_at = ? WHERE task_id = ?") + .run(100, created.taskId); + const { DatabaseSync } = requireNodeSqlite(); + const writer = new DatabaseSync(database.path); + writer.exec("PRAGMA busy_timeout = 1000;"); + const originalPrepare = database.db.prepare.bind(database.db); + let writerCommitted = false; + const prepareSpy = vi.spyOn(database.db, "prepare").mockImplementation((sql: string) => { + if (sql.includes('from "task_delivery_state"') && !writerCommitted) { + writerCommitted = true; + writer + .prepare( + "UPDATE task_delivery_state SET last_notified_event_at = ? WHERE task_id = ?", + ) + .run(200, created.taskId); + } + return originalPrepare(sql); + }); + + try { + const restored = loadTaskRegistryStateFromSqlite(); + expect(restored.deliveryStates.get(created.taskId)?.lastNotifiedEventAt).toBe(100); + expect( + writer + .prepare("SELECT last_notified_event_at FROM task_delivery_state WHERE task_id = ?") + .get(created.taskId), + ).toEqual({ last_notified_event_at: 200 }); + } finally { + prepareSpy.mockRestore(); + writer.close(); + } + }, + ); + }); + + it("bypasses stale owner indexes for complete fresh results", async () => { + await withOpenClawTestState( + { layout: "state-only", prefix: "openclaw-task-store-owner-index-" }, + async () => { + resetTaskRegistryForTests(); + const ownerKey = "agent:main:main"; + const target = createTaskRecord({ + runtime: "cron", + ownerKey, + scopeKind: "session", + sourceId: "owner-index-target", + runId: "run-owner-index-target", + task: "Find the target owner task", + status: "running", + deliveryStatus: "not_applicable", + notifyPolicy: "silent", + }); + createTaskRecord({ + runtime: "cron", + ownerKey: "agent:other:main", + scopeKind: "session", + sourceId: "owner-index-other", + runId: "run-owner-index-other", + task: "Ignore another owner task", + status: "queued", + deliveryStatus: "not_applicable", + notifyPolicy: "silent", + }); + expect(listFreshTasksForOwnerKey(ownerKey).map((task) => task.taskId)).toContain( + target.taskId, + ); + + const database = openOpenClawStateDatabase(); + createUnsafeTaskOwnerIndex(database.db); + expect(database.db.prepare("PRAGMA quick_check").get()).toEqual({ quick_check: "ok" }); + expect(database.db.prepare("PRAGMA integrity_check('task_runs')").all()).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + integrity_check: expect.stringMatching(/idx_task_runs_owner_key/), + }), + ]), + ); + expect( + database.db + .prepare( + "SELECT task_id FROM task_runs INDEXED BY idx_task_runs_owner_key WHERE owner_key = ?", + ) + .all(ownerKey), + ).toEqual([]); + expect(() => loadTaskRegistryStateFromSqlite()).toThrow( + /integrity_check failed.*idx_task_runs_owner_key/iu, + ); + expect(listFreshTasksForOwnerKey(ownerKey).map((task) => task.taskId)).toContain( + target.taskId, + ); + + resetTaskRegistryForTests({ persist: false }); + }, + ); + }); + it("emits incremental observer events for restore, mutation, and delete", () => { const events: TaskRegistryObserverEvent[] = []; configureTaskRegistryRuntime({