diff --git a/scripts/bench-sqlite-reliability.ts b/scripts/bench-sqlite-reliability.ts new file mode 100644 index 000000000000..0e094beb4342 --- /dev/null +++ b/scripts/bench-sqlite-reliability.ts @@ -0,0 +1,135 @@ +// SQLite reliability stress proof exercises snapshots during concurrent writes. +import fs from "node:fs"; +import path from "node:path"; +import { pathToFileURL } from "node:url"; +import type { + CliOptions, + ProfileId, + ReliabilityReport, +} from "./lib/sqlite-reliability-contract.js"; +import { runReliabilityStress } from "./lib/sqlite-reliability-runner.js"; + +const BOOLEAN_FLAGS = new Set(["--help"]); +const VALUE_FLAGS = new Set(["--agent", "--output", "--profile", "--repository", "--state-dir"]); + +class CliUsageError extends Error { + override name = "CliUsageError"; +} + +function parseFlagValue(flag: string, argv: string[]): string | undefined { + const index = argv.indexOf(flag); + if (index === -1) { + return undefined; + } + const value = argv[index + 1]; + if (!value || value.startsWith("-")) { + throw new CliUsageError(`${flag} requires a value`); + } + return value; +} + +function validateArgs(argv: string[]): void { + const seenValueFlags = new Set(); + for (let index = 0; index < argv.length; index += 1) { + const arg = argv[index] ?? ""; + if (BOOLEAN_FLAGS.has(arg)) { + continue; + } + if (!VALUE_FLAGS.has(arg)) { + throw new CliUsageError(`Unknown argument: ${arg}`); + } + if (seenValueFlags.has(arg)) { + throw new CliUsageError(`${arg} was provided more than once`); + } + seenValueFlags.add(arg); + const value = argv[index + 1]; + if (!value || value.startsWith("-")) { + throw new CliUsageError(`${arg} requires a value`); + } + index += 1; + } +} + +function parseProfile(raw: string | undefined): ProfileId { + if (!raw) { + return "default"; + } + if (raw === "smoke" || raw === "default" || raw === "large") { + return raw; + } + throw new CliUsageError( + `--profile must be one of smoke, default, large; got ${JSON.stringify(raw)}`, + ); +} + +function parseOptions(argv: string[]): CliOptions { + return { + agentId: parseFlagValue("--agent", argv) ?? null, + output: parseFlagValue("--output", argv) ?? null, + profile: parseProfile(parseFlagValue("--profile", argv)), + repository: parseFlagValue("--repository", argv) ?? null, + stateDir: parseFlagValue("--state-dir", argv) ?? null, + }; +} + +function printUsage(): void { + console.log(`OpenClaw SQLite reliability stress proof + +Usage: + node --import tsx scripts/bench-sqlite-reliability.ts [options] + +Options: + --profile Stress profile (default: default) + --agent Stress one per-agent database (default: global) + --state-dir Reuse a state directory instead of a temp dir + --repository Snapshot repository path + --output Write machine-readable JSON report + --help Show this text +`); +} + +function printProofLines(report: ReliabilityReport): void { + console.log(`SQLITE_RELIABILITY_PROFILE=${report.profile}`); + console.log(`SQLITE_RELIABILITY_TARGET=${report.target}`); + console.log(`SQLITE_RELIABILITY_PLATFORM=${report.platform}`); + console.log(`SQLITE_RELIABILITY_ARCH=${report.arch}`); + console.log(`SQLITE_RELIABILITY_ITERATIONS=${report.iterations}`); + console.log(`SQLITE_RELIABILITY_RETAINED_BATCHES=${report.retainedBatches}`); + console.log(`SQLITE_RELIABILITY_RESTORES_VERIFIED=${report.restoresVerified}`); + console.log(`SQLITE_RELIABILITY_WRITER_ROWS=${report.writer.rowsCommitted}`); + console.log( + `SQLITE_RELIABILITY_WAL_SENTINEL=${report.transactionProof.committedWalSentinel ? "verified" : "missing"}`, + ); + console.log(`SQLITE_RELIABILITY_HELD_BATCH=${report.transactionProof.heldBatch}`); + console.log(`SQLITE_RELIABILITY_SNAPSHOT_P95_MS=${report.timingsMs.snapshotP95.toFixed(3)}`); + console.log(`SQLITE_RELIABILITY_RESTORE_P95_MS=${report.timingsMs.restoreP95.toFixed(3)}`); + console.log(`SQLITE_RELIABILITY_SNAPSHOT_BYTES_MAX=${report.snapshotBytes.max}`); +} + +async function main(argv: string[]): Promise { + try { + validateArgs(argv); + if (argv.includes("--help")) { + printUsage(); + return; + } + const options = parseOptions(argv); + const report = await runReliabilityStress(options); + if (options.output) { + fs.mkdirSync(path.dirname(options.output), { recursive: true }); + fs.writeFileSync(options.output, `${JSON.stringify(report, null, 2)}\n`, "utf8"); + } + printProofLines(report); + } catch (error) { + if (error instanceof CliUsageError) { + console.error(`error: ${error.message}`); + process.exitCode = 2; + return; + } + throw error; + } +} + +if (import.meta.url === pathToFileURL(process.argv[1] ?? "").href) { + await main(process.argv.slice(2)); +} diff --git a/scripts/lib/sqlite-reliability-contract.ts b/scripts/lib/sqlite-reliability-contract.ts new file mode 100644 index 000000000000..10c0529cbed4 --- /dev/null +++ b/scripts/lib/sqlite-reliability-contract.ts @@ -0,0 +1,100 @@ +export type ProfileId = "smoke" | "default" | "large"; + +export type ProfileConfig = { + iterations: number; + payloadBytes: number; + retainedBatches: number; + rowsPerBatch: number; + writerPauseMs: number; +}; + +export type CliOptions = { + agentId: string | null; + output: string | null; + profile: ProfileId; + repository: string | null; + stateDir: string | null; +}; + +export type ReliabilityReport = { + arch: string; + iterations: number; + node: string; + paths: { + repository: string; + sourceDatabase: string; + stateDir: string; + syncedRepository: string; + }; + platform: NodeJS.Platform; + profile: ProfileId; + retainedBatches: number; + restoresVerified: number; + rowsPerBatch: number; + snapshotBytes: { + max: number; + min: number; + }; + target: string; + timingsMs: { + restoreP50: number; + restoreP95: number; + snapshotP50: number; + snapshotP95: number; + total: number; + }; + transactionProof: { + committedWalSentinel: true; + heldBatch: number; + heldRows: number; + visibleAfterRestore: false; + }; + walBytes: { + after: number; + before: number; + }; + writer: { + batchesCommitted: number; + rowsCommitted: number; + }; +}; + +export const PROFILES: Record = { + smoke: { + iterations: 4, + payloadBytes: 512, + retainedBatches: 32, + rowsPerBatch: 8, + writerPauseMs: 5, + }, + default: { + iterations: 25, + payloadBytes: 4 * 1024, + retainedBatches: 128, + rowsPerBatch: 32, + writerPauseMs: 5, + }, + large: { + iterations: 100, + payloadBytes: 8 * 1024, + retainedBatches: 256, + rowsPerBatch: 64, + writerPauseMs: 1, + }, +}; + +export const STRESS_TABLE_SQL = ` + CREATE TABLE IF NOT EXISTS openclaw_reliability_sentinel ( + id INTEGER PRIMARY KEY CHECK (id = 1), + payload TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS openclaw_reliability_entries ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + batch INTEGER NOT NULL, + ordinal INTEGER NOT NULL, + payload TEXT NOT NULL, + UNIQUE(batch, ordinal) + ); +`; + +export const COMMITTED_WAL_SENTINEL = "committed-before-ready"; diff --git a/scripts/lib/sqlite-reliability-runner.ts b/scripts/lib/sqlite-reliability-runner.ts new file mode 100644 index 000000000000..1619948f1610 --- /dev/null +++ b/scripts/lib/sqlite-reliability-runner.ts @@ -0,0 +1,334 @@ +import { randomUUID } from "node:crypto"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import type { DatabaseSync } from "node:sqlite"; +import { requireNodeSqlite } from "../../src/infra/node-sqlite.js"; +import { createLocalSqliteSnapshotProvider } from "../../src/snapshot/local-repository.js"; +import type { SnapshotDatabaseIdentity } from "../../src/snapshot/snapshot-provider.js"; +import { + assertOpenClawAgentDatabaseForMaintenance, + closeOpenClawAgentDatabasesForTest, + openOpenClawAgentDatabase, +} from "../../src/state/openclaw-agent-db.js"; +import { + assertOpenClawStateDatabaseForMaintenance, + closeOpenClawStateDatabaseForTest, + openOpenClawStateDatabase, +} from "../../src/state/openclaw-state-db.js"; +import { + COMMITTED_WAL_SENTINEL, + PROFILES, + STRESS_TABLE_SQL, + type CliOptions, + type ReliabilityReport, +} from "./sqlite-reliability-contract.js"; +import { + startWriter, + stopWriter, + terminateWriter, + waitForWriterMessage, + type WriterHandle, +} from "./sqlite-reliability-writer.js"; + +type TargetDatabase = { + identity: SnapshotDatabaseIdentity; + label: string; + path: string; +}; + +type IterationMetric = { + restoreMs: number; + snapshotBytes: number; + snapshotMs: number; +}; + +function nowMs(): number { + return Number(process.hrtime.bigint()) / 1e6; +} + +function percentile(values: number[], pct: number): number { + if (values.length === 0) { + return 0; + } + const sorted = values.toSorted((left, right) => left - right); + const index = Math.min(sorted.length - 1, Math.ceil((pct / 100) * sorted.length) - 1); + return Number((sorted[index] ?? 0).toFixed(3)); +} + +function fileSize(pathname: string): number { + try { + return fs.statSync(pathname).size; + } catch { + return 0; + } +} + +function resolveTargetDatabase(options: CliOptions, env: NodeJS.ProcessEnv): TargetDatabase { + if (options.agentId) { + const database = openOpenClawAgentDatabase({ agentId: options.agentId, env }); + const target = { + identity: { role: "agent", agentId: database.agentId } as const, + label: `agent:${database.agentId}`, + path: database.path, + }; + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + return target; + } + const database = openOpenClawStateDatabase({ env }); + const target = { + identity: { role: "global" } as const, + label: "global", + path: database.path, + }; + closeOpenClawStateDatabaseForTest(); + return target; +} + +function setupStressTable(databasePath: string): void { + const { DatabaseSync } = requireNodeSqlite(); + const database = new DatabaseSync(databasePath); + try { + database.exec("PRAGMA journal_mode = WAL;"); + database.exec("PRAGMA busy_timeout = 30000;"); + database.exec(STRESS_TABLE_SQL); + database.prepare("DELETE FROM openclaw_reliability_entries").run(); + database.prepare("DELETE FROM openclaw_reliability_sentinel").run(); + } finally { + database.close(); + } +} + +function copySnapshotDirectory(sourcePath: string, syncedRepository: string): string { + fs.mkdirSync(syncedRepository, { recursive: true, mode: 0o700 }); + const destinationPath = path.join(syncedRepository, path.basename(sourcePath)); + fs.cpSync(sourcePath, destinationPath, { + errorOnExist: true, + force: false, + recursive: true, + }); + return destinationPath; +} + +function assertPragmaOk(database: DatabaseSync, pragma: "integrity_check" | "quick_check"): void { + const rows = database.prepare(`PRAGMA ${pragma};`).all() as Array>; + const messages = rows.map((row) => row[pragma]); + if (messages.length !== 1 || messages[0] !== "ok") { + throw new Error(`${pragma} failed: ${messages.map(String).join("; ")}`); + } +} + +function verifyRestoredDatabase(params: { + identity: SnapshotDatabaseIdentity; + path: string; + rowsPerBatch: number; + uncommittedBatch: number | null; +}): void { + const { DatabaseSync } = requireNodeSqlite(); + const database = new DatabaseSync(params.path, { readOnly: true }); + try { + database.exec("PRAGMA trusted_schema = OFF;"); + assertPragmaOk(database, "quick_check"); + assertPragmaOk(database, "integrity_check"); + const foreignKeys = database.prepare("PRAGMA foreign_key_check;").all(); + if (foreignKeys.length > 0) { + throw new Error(`foreign_key_check failed with ${foreignKeys.length} row(s)`); + } + if (params.identity.role === "global") { + assertOpenClawStateDatabaseForMaintenance(database, { pathname: params.path }); + } else if (params.identity.role === "agent") { + assertOpenClawAgentDatabaseForMaintenance(database, { + agentId: params.identity.agentId, + pathname: params.path, + }); + } + const sentinel = database + .prepare("SELECT payload FROM openclaw_reliability_sentinel WHERE id = 1") + .get() as { payload?: unknown } | undefined; + if (sentinel?.payload !== COMMITTED_WAL_SENTINEL) { + throw new Error("committed WAL sentinel is missing after restore"); + } + const partial = database + .prepare( + `SELECT batch, COUNT(*) AS row_count + FROM openclaw_reliability_entries + GROUP BY batch + HAVING COUNT(*) <> ? + LIMIT 1`, + ) + .get(params.rowsPerBatch) as { batch?: unknown; row_count?: unknown } | undefined; + if (partial) { + throw new Error( + `partial transaction visible after restore: batch=${String(partial.batch)} rows=${String(partial.row_count)}`, + ); + } + if (params.uncommittedBatch !== null) { + const held = database + .prepare("SELECT COUNT(*) AS rows FROM openclaw_reliability_entries WHERE batch = ?") + .get(params.uncommittedBatch) as { rows?: unknown }; + if (Number(held.rows) !== 0) { + throw new Error( + `uncommitted transaction became visible after restore: batch=${params.uncommittedBatch} rows=${String(held.rows)}`, + ); + } + } + } finally { + database.close(); + } +} + +async function runSnapshotIteration(params: { + iteration: number; + repositoryProvider: ReturnType; + restoreRoot: string; + rowsPerBatch: number; + syncedProvider: ReturnType; + syncedRepository: string; + target: TargetDatabase; + uncommittedBatch: number | null; +}): Promise { + const snapshotStarted = nowMs(); + const snapshot = await params.repositoryProvider.create({ + identity: params.target.identity, + path: params.target.path, + }); + const snapshotMs = nowMs() - snapshotStarted; + const copiedPath = copySnapshotDirectory(snapshot.ref.path, params.syncedRepository); + const copiedRef = { path: copiedPath }; + await params.syncedProvider.verify(copiedRef); + const restorePath = path.join(params.restoreRoot, `restore-${params.iteration}.sqlite`); + const restoreStarted = nowMs(); + await params.syncedProvider.restoreFresh(copiedRef, restorePath); + const restoreMs = nowMs() - restoreStarted; + verifyRestoredDatabase({ + identity: params.target.identity, + path: restorePath, + rowsPerBatch: params.rowsPerBatch, + uncommittedBatch: params.uncommittedBatch, + }); + return { + restoreMs: Number(restoreMs.toFixed(3)), + snapshotBytes: snapshot.manifest.artifact.sizeBytes, + snapshotMs: Number(snapshotMs.toFixed(3)), + }; +} + +export async function runReliabilityStress(options: CliOptions): Promise { + const profile = PROFILES[options.profile]; + const ownsStateDir = options.stateDir === null; + const stateDir = + options.stateDir ?? fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-sqlite-reliability-")); + const repository = options.repository ?? path.join(stateDir, "snapshots"); + const runScratch = path.join(stateDir, "sqlite-reliability-runs", randomUUID()); + const syncedRepository = path.join(runScratch, "synced-snapshots"); + const validationRoot = path.join(runScratch, "snapshot-validation"); + const restoreRoot = path.join(runScratch, "restored"); + const env = { ...process.env, OPENCLAW_STATE_DIR: stateDir }; + const started = nowMs(); + let writer: WriterHandle | undefined; + try { + fs.mkdirSync(validationRoot, { recursive: true, mode: 0o700 }); + const target = resolveTargetDatabase(options, env); + setupStressTable(target.path); + const repositoryProvider = createLocalSqliteSnapshotProvider({ + repositoryPath: repository, + validationRootPath: validationRoot, + }); + const syncedProvider = createLocalSqliteSnapshotProvider({ + repositoryPath: syncedRepository, + validationRootPath: validationRoot, + }); + writer = startWriter(target.path, profile); + await waitForWriterMessage(writer, "ready"); + const walBytesBefore = fileSize(`${target.path}-wal`); + const partial = await waitForWriterMessage(writer, "partial", () => { + writer?.child.send?.({ kind: "hold-partial" }); + }); + const metrics: IterationMetric[] = []; + for (let iteration = 0; iteration < profile.iterations; iteration += 1) { + metrics.push( + await runSnapshotIteration({ + iteration, + repositoryProvider, + restoreRoot, + rowsPerBatch: profile.rowsPerBatch, + syncedProvider, + syncedRepository, + target, + uncommittedBatch: iteration === 0 ? partial.batch : null, + }), + ); + if (iteration === 0) { + await waitForWriterMessage(writer, "released", () => { + writer?.child.send?.({ action: "rollback", kind: "release-partial" }); + }); + } + } + const writerResult = await stopWriter(writer); + const snapshotBytes = metrics.map((metric) => metric.snapshotBytes); + return { + arch: process.arch, + iterations: profile.iterations, + node: process.version, + paths: { + repository, + sourceDatabase: target.path, + stateDir, + syncedRepository, + }, + platform: process.platform, + profile: options.profile, + retainedBatches: profile.retainedBatches, + restoresVerified: metrics.length, + rowsPerBatch: profile.rowsPerBatch, + snapshotBytes: { + max: Math.max(...snapshotBytes), + min: Math.min(...snapshotBytes), + }, + target: target.label, + timingsMs: { + restoreP50: percentile( + metrics.map((metric) => metric.restoreMs), + 50, + ), + restoreP95: percentile( + metrics.map((metric) => metric.restoreMs), + 95, + ), + snapshotP50: percentile( + metrics.map((metric) => metric.snapshotMs), + 50, + ), + snapshotP95: percentile( + metrics.map((metric) => metric.snapshotMs), + 95, + ), + total: Number((nowMs() - started).toFixed(3)), + }, + transactionProof: { + committedWalSentinel: true, + heldBatch: partial.batch, + heldRows: partial.rows, + visibleAfterRestore: false, + }, + walBytes: { + after: fileSize(`${target.path}-wal`), + before: walBytesBefore, + }, + writer: { + batchesCommitted: writerResult.batchesCommitted, + rowsCommitted: writerResult.rowsCommitted, + }, + }; + } finally { + if (writer && !writer.stopped) { + await terminateWriter(writer); + } + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + if (ownsStateDir) { + fs.rmSync(stateDir, { force: true, recursive: true }); + } + } +} diff --git a/scripts/lib/sqlite-reliability-writer.ts b/scripts/lib/sqlite-reliability-writer.ts new file mode 100644 index 000000000000..db82bea29eb5 --- /dev/null +++ b/scripts/lib/sqlite-reliability-writer.ts @@ -0,0 +1,358 @@ +import { fork, type ChildProcess } from "node:child_process"; +import { setImmediate as delayImmediate, setTimeout as delay } from "node:timers/promises"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { requireNodeSqlite } from "../../src/infra/node-sqlite.js"; +import { + COMMITTED_WAL_SENTINEL, + STRESS_TABLE_SQL, + type ProfileConfig, +} from "./sqlite-reliability-contract.js"; + +type WriterReadyMessage = { + kind: "ready"; +}; + +export type WriterPartialMessage = { + batch: number; + kind: "partial"; + rows: number; +}; + +type WriterReleasedMessage = { + batch: number; + kind: "released"; +}; + +export type WriterResultMessage = { + batchesCommitted: number; + kind: "result"; + rowsCommitted: number; +}; + +type WriterErrorMessage = { + error: string; + kind: "error"; +}; + +type WriterMessage = + | WriterReadyMessage + | WriterPartialMessage + | WriterReleasedMessage + | WriterResultMessage + | WriterErrorMessage; + +export type WriterHandle = { + child: ChildProcess; + stderr: string[]; + stopped: boolean; +}; + +const WRITER_MESSAGE_TIMEOUT_MS = 30_000; + +export function startWriter(databasePath: string, profile: ProfileConfig): WriterHandle { + const child = fork( + fileURLToPath(import.meta.url), + [ + databasePath, + String(profile.rowsPerBatch), + String(profile.payloadBytes), + String(profile.retainedBatches), + String(profile.writerPauseMs), + ], + { + execArgv: ["--import", "tsx"], + serialization: "json", + stdio: ["ignore", "ignore", "pipe", "ipc"], + }, + ); + const stderr: string[] = []; + child.stderr?.setEncoding("utf8"); + child.stderr?.on("data", (chunk: string) => { + stderr.push(chunk); + }); + return { child, stderr, stopped: false }; +} + +export async function waitForWriterMessage( + writer: WriterHandle, + kind: T, + action?: () => void, +): Promise> { + return await new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + cleanup(); + reject( + new Error( + `SQLite reliability writer timed out waiting for ${kind}.${formatWriterStderr(writer)}`, + ), + ); + }, WRITER_MESSAGE_TIMEOUT_MS); + const onMessage = (message: WriterMessage) => { + if (message.kind === "error") { + cleanup(); + reject(new Error(`SQLite reliability writer failed: ${message.error}`)); + return; + } + if (message.kind !== kind) { + return; + } + cleanup(); + resolve(message as Extract); + }; + const onError = (error: Error) => { + cleanup(); + reject(error); + }; + const onExit = (code: number | null, signal: NodeJS.Signals | null) => { + cleanup(); + reject( + new Error( + `SQLite reliability writer exited before ${kind}: code=${String(code)} signal=${String(signal)}.${formatWriterStderr(writer)}`, + ), + ); + }; + const cleanup = () => { + clearTimeout(timeout); + writer.child.off("message", onMessage); + writer.child.off("error", onError); + writer.child.off("exit", onExit); + }; + writer.child.on("message", onMessage); + writer.child.on("error", onError); + writer.child.on("exit", onExit); + action?.(); + }); +} + +function formatWriterStderr(writer: WriterHandle): string { + const text = writer.stderr.join("").trim(); + return text ? ` stderr=${JSON.stringify(text)}` : ""; +} + +export async function stopWriter(writer: WriterHandle): Promise { + if (writer.stopped) { + throw new Error("SQLite reliability writer was already stopped."); + } + const result = await waitForWriterMessage(writer, "result", () => { + writer.child.send?.({ kind: "stop" }); + }); + await waitForChildExit(writer.child); + writer.stopped = true; + return result; +} + +async function waitForChildExit(child: ChildProcess): Promise { + if (child.exitCode !== null || child.signalCode !== null) { + return; + } + await new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + cleanup(); + reject(new Error("SQLite reliability writer did not exit after stopping.")); + }, WRITER_MESSAGE_TIMEOUT_MS); + const onExit = () => { + cleanup(); + resolve(); + }; + const onError = (error: Error) => { + cleanup(); + reject(error); + }; + const cleanup = () => { + clearTimeout(timeout); + child.off("exit", onExit); + child.off("error", onError); + }; + child.on("exit", onExit); + child.on("error", onError); + }); +} + +export async function terminateWriter(writer: WriterHandle): Promise { + if (writer.child.exitCode !== null || writer.child.signalCode !== null) { + return; + } + writer.child.kill(); + await waitForChildExit(writer.child).catch(() => undefined); +} + +function parseWriterChildArgs(argv: string[]): { + databasePath: string; + payloadBytes: number; + retainedBatches: number; + rowsPerBatch: number; + writerPauseMs: number; +} { + const [databasePath, rowsRaw, payloadRaw, retainedRaw, pauseRaw, ...extra] = argv; + const rowsPerBatch = Number(rowsRaw); + const payloadBytes = Number(payloadRaw); + const retainedBatches = Number(retainedRaw); + const writerPauseMs = Number(pauseRaw); + if ( + !databasePath || + extra.length > 0 || + !Number.isSafeInteger(rowsPerBatch) || + rowsPerBatch < 2 || + !Number.isSafeInteger(payloadBytes) || + payloadBytes < 1 || + !Number.isSafeInteger(retainedBatches) || + retainedBatches < 1 || + !Number.isSafeInteger(writerPauseMs) || + writerPauseMs < 0 + ) { + throw new Error("invalid SQLite reliability writer arguments"); + } + return { databasePath, payloadBytes, retainedBatches, rowsPerBatch, writerPauseMs }; +} + +async function runWriterChild(argv: string[]): Promise { + const options = parseWriterChildArgs(argv); + const { DatabaseSync } = requireNodeSqlite(); + const database = new DatabaseSync(options.databasePath); + let nextBatch = 0; + let batchesCommitted = 0; + let rowsCommitted = 0; + let stopping = false; + let holdPartial = false; + let releasePartial: "commit" | "rollback" | undefined; + const payload = "x".repeat(options.payloadBytes); + const requestStop = () => { + stopping = true; + releasePartial ??= "rollback"; + }; + const sendMessage = (message: WriterMessage) => { + if (process.connected) { + process.send?.(message); + } + }; + try { + database.exec("PRAGMA journal_mode = WAL;"); + database.exec("PRAGMA wal_autocheckpoint = 0;"); + database.exec("PRAGMA busy_timeout = 30000;"); + database.exec(STRESS_TABLE_SQL); + const next = database + .prepare( + "SELECT COALESCE(MAX(batch), -1) + 1 AS next_batch FROM openclaw_reliability_entries", + ) + .get() as { next_batch?: number | bigint }; + nextBatch = Number(next.next_batch ?? 0); + const insert = database.prepare( + "INSERT INTO openclaw_reliability_entries (batch, ordinal, payload) VALUES (?, ?, ?)", + ); + const insertSentinel = database.prepare( + "INSERT INTO openclaw_reliability_sentinel (id, payload) VALUES (1, ?)", + ); + const deleteExpired = database.prepare( + "DELETE FROM openclaw_reliability_entries WHERE batch < ?", + ); + process.on("message", (message: unknown) => { + if (!message || typeof message !== "object") { + return; + } + const command = message as { action?: unknown; kind?: unknown }; + if (command.kind === "stop") { + requestStop(); + } else if (command.kind === "hold-partial") { + holdPartial = true; + } else if ( + command.kind === "release-partial" && + (command.action === "commit" || command.action === "rollback") + ) { + releasePartial = command.action; + } + }); + process.on("disconnect", requestStop); + if (!process.connected) { + requestStop(); + } + const shouldStop = () => stopping; + const shouldReleasePartial = () => releasePartial !== undefined || stopping; + + const commitBatch = (includeSentinel = false) => { + database.exec("BEGIN IMMEDIATE;"); + try { + if (includeSentinel) { + insertSentinel.run(COMMITTED_WAL_SENTINEL); + } + for (let ordinal = 0; ordinal < options.rowsPerBatch; ordinal += 1) { + insert.run(nextBatch, ordinal, `${nextBatch}:${ordinal}:${payload}`); + } + deleteExpired.run(Math.max(0, nextBatch - options.retainedBatches + 1)); + database.exec("COMMIT;"); + nextBatch += 1; + batchesCommitted += 1; + rowsCommitted += options.rowsPerBatch; + } catch (error) { + database.exec("ROLLBACK;"); + throw error; + } + }; + + commitBatch(true); + sendMessage({ kind: "ready" } satisfies WriterReadyMessage); + while (!shouldStop()) { + if (holdPartial) { + holdPartial = false; + const heldBatch = nextBatch; + const heldRows = Math.max(1, Math.floor(options.rowsPerBatch / 2)); + database.exec("BEGIN IMMEDIATE;"); + try { + for (let ordinal = 0; ordinal < heldRows; ordinal += 1) { + insert.run(heldBatch, ordinal, `${heldBatch}:${ordinal}:${payload}`); + } + sendMessage({ + batch: heldBatch, + kind: "partial", + rows: heldRows, + } satisfies WriterPartialMessage); + while (!shouldReleasePartial()) { + await delay(1); + } + if (releasePartial === "commit") { + for (let ordinal = heldRows; ordinal < options.rowsPerBatch; ordinal += 1) { + insert.run(heldBatch, ordinal, `${heldBatch}:${ordinal}:${payload}`); + } + database.exec("COMMIT;"); + nextBatch += 1; + batchesCommitted += 1; + rowsCommitted += options.rowsPerBatch; + } else { + database.exec("ROLLBACK;"); + } + releasePartial = undefined; + sendMessage({ batch: heldBatch, kind: "released" } satisfies WriterReleasedMessage); + } catch (error) { + database.exec("ROLLBACK;"); + throw error; + } + } else { + commitBatch(); + } + if (options.writerPauseMs > 0) { + await delay(options.writerPauseMs); + } else { + await delayImmediate(); + } + } + sendMessage({ + batchesCommitted, + kind: "result", + rowsCommitted, + } satisfies WriterResultMessage); + } catch (error) { + sendMessage({ + error: error instanceof Error ? (error.stack ?? error.message) : String(error), + kind: "error", + } satisfies WriterErrorMessage); + process.exitCode = 1; + } finally { + database.close(); + if (process.connected) { + process.disconnect?.(); + } + } +} + +if (import.meta.url === pathToFileURL(process.argv[1] ?? "").href) { + await runWriterChild(process.argv.slice(2)); +} diff --git a/test/scripts/bench-sqlite-reliability.test.ts b/test/scripts/bench-sqlite-reliability.test.ts new file mode 100644 index 000000000000..5e77112d8d31 --- /dev/null +++ b/test/scripts/bench-sqlite-reliability.test.ts @@ -0,0 +1,207 @@ +// SQLite reliability proof tests cover CLI safety and one real snapshot round trip. +import { fork, spawnSync, type ChildProcess } from "node:child_process"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { afterEach, describe, expect, it } from "vitest"; + +const tempDirs: string[] = []; + +function makeTempDir(): string { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-sqlite-reliability-test-")); + tempDirs.push(tempDir); + return tempDir; +} + +function runProof(args: string[]) { + return spawnSync( + process.execPath, + ["--import", "tsx", "scripts/bench-sqlite-reliability.ts", ...args], + { + cwd: process.cwd(), + encoding: "utf8", + timeout: 120_000, + }, + ); +} + +async function waitForChildReady(child: ChildProcess): Promise { + await new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + cleanup(); + reject(new Error("writer child did not become ready")); + }, 10_000); + const onMessage = (message: unknown) => { + if ( + message && + typeof message === "object" && + (message as { kind?: unknown }).kind === "ready" + ) { + cleanup(); + resolve(); + } + }; + const onError = (error: Error) => { + cleanup(); + reject(error); + }; + const onExit = () => { + cleanup(); + reject(new Error("writer child exited before ready")); + }; + const cleanup = () => { + clearTimeout(timeout); + child.off("message", onMessage); + child.off("error", onError); + child.off("exit", onExit); + }; + child.on("message", onMessage); + child.on("error", onError); + child.on("exit", onExit); + }); +} + +async function waitForChildExit(child: ChildProcess): Promise<{ + code: number | null; + signal: NodeJS.Signals | null; +}> { + return await new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + cleanup(); + reject(new Error("writer child did not exit after IPC disconnect")); + }, 10_000); + const onExit = (code: number | null, signal: NodeJS.Signals | null) => { + cleanup(); + resolve({ code, signal }); + }; + const onError = (error: Error) => { + cleanup(); + reject(error); + }; + const cleanup = () => { + clearTimeout(timeout); + child.off("exit", onExit); + child.off("error", onError); + }; + child.on("exit", onExit); + child.on("error", onError); + }); +} + +afterEach(() => { + for (const tempDir of tempDirs.splice(0)) { + fs.rmSync(tempDir, { force: true, recursive: true }); + } +}); + +describe("scripts/bench-sqlite-reliability", () => { + it("rejects malformed arguments before creating state", () => { + const unknown = runProof(["--wat"]); + expect(unknown.status).toBe(2); + expect(unknown.stdout).toBe(""); + expect(unknown.stderr.trim()).toBe("error: Unknown argument: --wat"); + + const duplicate = runProof(["--profile", "smoke", "--profile", "large"]); + expect(duplicate.status).toBe(2); + expect(duplicate.stdout).toBe(""); + expect(duplicate.stderr.trim()).toBe("error: --profile was provided more than once"); + + const invalid = runProof(["--profile", "huge"]); + expect(invalid.status).toBe(2); + expect(invalid.stdout).toBe(""); + expect(invalid.stderr.trim()).toBe( + 'error: --profile must be one of smoke, default, large; got "huge"', + ); + }); + + it("reuses a state directory without stale rows or restore collisions", () => { + const stateDir = makeTempDir(); + const firstOutput = path.join(stateDir, "report-first.json"); + const firstResult = runProof([ + "--profile", + "smoke", + "--state-dir", + stateDir, + "--output", + firstOutput, + ]); + + expect(firstResult.status, firstResult.stderr).toBe(0); + expect(firstResult.stderr).toBe(""); + expect(firstResult.stdout).toContain("SQLITE_RELIABILITY_TARGET=global"); + expect(firstResult.stdout).toContain("SQLITE_RELIABILITY_RESTORES_VERIFIED=4"); + const firstReport = JSON.parse(fs.readFileSync(firstOutput, "utf8")) as { + paths: { + sourceDatabase: string; + syncedRepository: string; + }; + restoresVerified: number; + transactionProof: { + committedWalSentinel: boolean; + heldRows: number; + visibleAfterRestore: boolean; + }; + writer: { + rowsCommitted: number; + }; + }; + expect(firstReport.restoresVerified).toBe(4); + expect(firstReport.transactionProof.committedWalSentinel).toBe(true); + expect(firstReport.transactionProof.heldRows).toBeGreaterThan(0); + expect(firstReport.transactionProof.visibleAfterRestore).toBe(false); + expect(firstReport.writer.rowsCommitted).toBeGreaterThan(0); + + const database = new DatabaseSync(firstReport.paths.sourceDatabase); + try { + database + .prepare( + "INSERT INTO openclaw_reliability_entries (batch, ordinal, payload) VALUES (?, ?, ?)", + ) + .run(999_999, 0, "stale-profile-row"); + } finally { + database.close(); + } + + const secondOutput = path.join(stateDir, "report-second.json"); + const secondResult = runProof([ + "--profile", + "smoke", + "--state-dir", + stateDir, + "--output", + secondOutput, + ]); + expect(secondResult.status, secondResult.stderr).toBe(0); + const secondReport = JSON.parse(fs.readFileSync(secondOutput, "utf8")) as { + paths: { syncedRepository: string }; + restoresVerified: number; + }; + expect(secondReport.restoresVerified).toBe(4); + expect(secondReport.paths.syncedRepository).not.toBe(firstReport.paths.syncedRepository); + }); + + it("stops the writer when its parent IPC channel disconnects", async () => { + const databasePath = path.join(makeTempDir(), "writer.sqlite"); + const child = fork( + path.resolve("scripts/lib/sqlite-reliability-writer.ts"), + [databasePath, "8", "64", "4", "1"], + { + cwd: process.cwd(), + execArgv: ["--import", "tsx"], + serialization: "json", + stdio: ["ignore", "ignore", "pipe", "ipc"], + }, + ); + try { + await waitForChildReady(child); + const exitPromise = waitForChildExit(child); + child.disconnect(); + await expect(exitPromise).resolves.toEqual({ code: 0, signal: null }); + } finally { + if (child.exitCode === null && child.signalCode === null) { + child.kill(); + } + } + }); +});