diff --git a/.github/workflows/openclaw-performance.yml b/.github/workflows/openclaw-performance.yml index ac78dfae36d9..e6eb81e0b123 100644 --- a/.github/workflows/openclaw-performance.yml +++ b/.github/workflows/openclaw-performance.yml @@ -527,6 +527,13 @@ jobs: cleanup_gateway trap - EXIT + if node -e "const fs=require('node:fs'); const scripts=require('./package.json').scripts||{}; process.exit(scripts['test:sqlite:perf:smoke'] && fs.existsSync('scripts/bench-sqlite-state.ts') ? 0 : 1)"; then + pnpm test:sqlite:perf:smoke + cp .artifacts/sqlite-perf/smoke.json "$SOURCE_PERF_DIR/sqlite-perf-smoke.json" + else + echo "SQLite state smoke probe is not available in ${TESTED_REF}; continuing with the remaining source probes." >> "$GITHUB_STEP_SUMMARY" + fi + summary_args=(node "$PERFORMANCE_HELPER_DIR/scripts/openclaw-performance-source-summary.mjs" \ --source-dir "$SOURCE_PERF_DIR" \ --output "$SOURCE_PERF_DIR/index.md") @@ -604,7 +611,7 @@ jobs: ## Source probes - Additional gateway boot, memory, plugin pressure, mock hello-loop, and CLI startup numbers are in [source/index.md](source/index.md). + Additional gateway boot, memory, plugin pressure, mock hello-loop, CLI startup, and SQLite state smoke numbers are in [source/index.md](source/index.md). EOF fi fi diff --git a/docs/ci.md b/docs/ci.md index ca13534166d5..db9067bc9bed 100644 --- a/docs/ci.md +++ b/docs/ci.md @@ -183,7 +183,7 @@ The workflow installs OCM from a pinned release and Kova from `openclaw/Kova` at - `mock-deep-profile`: CPU/heap/trace profiling for startup, gateway, and agent-turn hotspots. - `live-openai-candidate`: a real OpenAI `openai/gpt-5.5` agent turn, skipped when `OPENAI_API_KEY` is unavailable. -The mock-provider lane also runs OpenClaw-native source probes after the Kova pass: gateway boot timing and memory across default, hook, and 50-plugin startup cases; bundled plugin import RSS, repeated mock-OpenAI `channel-chat-baseline` hello loops, and CLI startup commands against the booted gateway. When the previous published mock-provider source report is available for the tested ref, the source summary compares current RSS and heap values against that baseline and marks large RSS increases as `watch`. The source probe Markdown summary lives at `source/index.md` in the report bundle, with raw JSON beside it. +The mock-provider lane also runs OpenClaw-native source probes after the Kova pass: gateway boot timing and memory across default, hook, and 50-plugin startup cases; bundled plugin import RSS, repeated mock-OpenAI `channel-chat-baseline` hello loops, CLI startup commands against the booted gateway, and the SQLite state smoke performance probe. When the previous published mock-provider source report is available for the tested ref, the source summary compares current RSS and heap values against that baseline and marks large RSS increases as `watch`. The source probe Markdown summary lives at `source/index.md` in the report bundle, with raw JSON beside it. Every lane uploads GitHub artifacts. When `CLAWGRIT_REPORTS_TOKEN` is configured, the workflow also commits `report.json`, `report.md`, bundles, `index.md`, and source-probe artifacts into `openclaw/clawgrit-reports` under `openclaw-performance//-//`. The current tested-ref pointer is written as `openclaw-performance//latest-.json`. diff --git a/docs/docs.json b/docs/docs.json index 13669bc5890c..32b495cc934f 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -1376,7 +1376,6 @@ "clawhub/publishing", "clawhub/plugin-validation-fixes", "clawhub/skill-format", - "clawhub/soul-format", "clawhub/auth", "clawhub/telemetry", "clawhub/troubleshooting" diff --git a/package.json b/package.json index f19f7616673c..f396fbe4b955 100644 --- a/package.json +++ b/package.json @@ -1813,6 +1813,9 @@ "test:perf:imports:changed": "node scripts/test-projects-imports.mjs --changed origin/main", "test:perf:profile:main": "node scripts/run-vitest-profile.mjs main", "test:perf:profile:runner": "node scripts/run-vitest-profile.mjs runner", + "test:sqlite:perf": "node --import tsx scripts/bench-sqlite-state.ts --profile default --output .artifacts/sqlite-perf/default.json", + "test:sqlite:perf:large": "node --import tsx scripts/bench-sqlite-state.ts --profile large --output .artifacts/sqlite-perf/large.json", + "test:sqlite:perf:smoke": "node --import tsx scripts/bench-sqlite-state.ts --profile smoke --output .artifacts/sqlite-perf/smoke.json", "test:plugins:gateway-gauntlet": "node scripts/check-plugin-gateway-gauntlet.mjs", "test:plugins:kitchen-sink-live": "bash -lc 'if [ -x \"$HOME/.local/bin/openclaw-testbox-env\" ]; then exec \"$HOME/.local/bin/openclaw-testbox-env\" pnpm openclaw qa suite --provider-mode live-frontier --scenario kitchen-sink-live-openai; fi; exec pnpm openclaw qa suite --provider-mode live-frontier --scenario kitchen-sink-live-openai'", "test:plugins:kitchen-sink-rpc": "node --import tsx scripts/e2e/kitchen-sink-rpc-walk.mjs", diff --git a/scripts/bench-sqlite-state.ts b/scripts/bench-sqlite-state.ts new file mode 100644 index 000000000000..43d678fb2c75 --- /dev/null +++ b/scripts/bench-sqlite-state.ts @@ -0,0 +1,630 @@ +// SQLite state benchmark seeds OpenClaw DBs and reports hot-query proof lines. +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import type { DatabaseSync } from "node:sqlite"; +import { pathToFileURL } from "node:url"; +import { + openOpenClawAgentDatabase, + closeOpenClawAgentDatabasesForTest, +} from "../src/state/openclaw-agent-db.js"; +import { + closeOpenClawStateDatabaseForTest, + openOpenClawStateDatabase, +} from "../src/state/openclaw-state-db.js"; +import { parseStrictIntegerOption } from "./lib/dev-tooling-safety.ts"; + +type ProfileId = "smoke" | "default" | "large"; + +type ProfileConfig = { + agentCacheEntries: number; + agentCount: number; + channelIngressEvents: number; + cronJobs: number; + cronRunLogs: number; + deliveryQueueEntries: number; + pluginStateEntries: number; + queryRuns: number; +}; + +type TimedQuery = { + p50Ms: number; + p95Ms: number; + query: string; + rows: number; +}; + +type BenchmarkReport = { + integrity: { + agent: string[]; + state: string; + }; + node: string; + paths: { + agentDatabases: string[]; + artifact: string | null; + stateDatabase: string; + stateDir: string; + }; + profile: ProfileId; + queries: TimedQuery[]; + rows: { + agentCacheEntries: number; + agentDatabases: number; + channelIngressEvents: number; + cronJobs: number; + cronRunLogs: number; + deliveryQueueEntries: number; + pluginStateEntries: number; + stateRows: number; + }; + timingsMs: { + checkpoint: number; + seed: number; + total: number; + }; + walBytes: { + agentAfter: number[]; + agentBefore: number[]; + stateAfter: number; + stateBefore: number; + }; +}; + +const PROFILES: Record = { + smoke: { + agentCacheEntries: 1_000, + agentCount: 2, + channelIngressEvents: 1_000, + cronJobs: 100, + cronRunLogs: 1_000, + deliveryQueueEntries: 1_000, + pluginStateEntries: 1_000, + queryRuns: 12, + }, + default: { + agentCacheEntries: 20_000, + agentCount: 5, + channelIngressEvents: 10_000, + cronJobs: 1_000, + cronRunLogs: 50_000, + deliveryQueueEntries: 50_000, + pluginStateEntries: 20_000, + queryRuns: 30, + }, + large: { + agentCacheEntries: 50_000, + agentCount: 10, + channelIngressEvents: 100_000, + cronJobs: 5_000, + cronRunLogs: 250_000, + deliveryQueueEntries: 200_000, + pluginStateEntries: 100_000, + queryRuns: 40, + }, +}; + +type CliOptions = { + output: string | null; + profile: ProfileId; + stateDir: string | null; +}; + +function parseFlagValue(flag: string): string | undefined { + const index = process.argv.indexOf(flag); + if (index === -1) { + return undefined; + } + const value = process.argv[index + 1]; + if (!value || value.startsWith("--")) { + throw new Error(`${flag} requires a value`); + } + return value; +} + +function hasFlag(flag: string): boolean { + return process.argv.includes(flag); +} + +function parseProfile(raw: string | undefined): ProfileId { + if (!raw) { + return "default"; + } + if (raw === "smoke" || raw === "default" || raw === "large") { + return raw; + } + throw new Error(`--profile must be one of smoke, default, large; got ${JSON.stringify(raw)}`); +} + +function parseOptions(): CliOptions { + return { + output: parseFlagValue("--output") ?? null, + profile: parseProfile(parseFlagValue("--profile")), + stateDir: parseFlagValue("--state-dir") ?? null, + }; +} + +function applyScale(config: ProfileConfig): ProfileConfig { + const scale = parseStrictIntegerOption({ + fallback: 1, + label: "SQLITE_PERF_SCALE", + min: 1, + raw: process.env["SQLITE_PERF_SCALE"], + }); + if (scale === 1) { + return config; + } + return { + agentCacheEntries: config.agentCacheEntries * scale, + agentCount: config.agentCount, + channelIngressEvents: config.channelIngressEvents * scale, + cronJobs: config.cronJobs * scale, + cronRunLogs: config.cronRunLogs * scale, + deliveryQueueEntries: config.deliveryQueueEntries * scale, + pluginStateEntries: config.pluginStateEntries * scale, + queryRuns: config.queryRuns, + }; +} + +function printUsage(): void { + console.log(`OpenClaw SQLite state benchmark + +Usage: + node --import tsx scripts/bench-sqlite-state.ts [options] + +Options: + --profile Data volume profile (default: default) + --state-dir Reuse a state directory instead of a temp dir + --output Write machine-readable JSON report + --help Show this text + +Environment: + SQLITE_PERF_SCALE= Multiplies row counts for the selected profile +`); +} + +function nowMs(): number { + return Number(process.hrtime.bigint()) / 1e6; +} + +function fileSize(pathname: string): number { + try { + return fs.statSync(pathname).size; + } catch { + return 0; + } +} + +function walSize(pathname: string): number { + return fileSize(`${pathname}-wal`); +} + +function stateRowCount(config: ProfileConfig): number { + return ( + config.channelIngressEvents + + config.cronJobs + + config.cronRunLogs + + config.deliveryQueueEntries + + config.pluginStateEntries + ); +} + +function seedStateDatabase(db: DatabaseSync, config: ProfileConfig): void { + db.exec("BEGIN IMMEDIATE;"); + try { + seedCronJobs(db, config.cronJobs); + seedCronRunLogs(db, config.cronRunLogs); + seedDeliveryQueue(db, config.deliveryQueueEntries); + seedPluginState(db, config.pluginStateEntries); + seedChannelIngress(db, config.channelIngressEvents); + db.exec("COMMIT;"); + } catch (err) { + db.exec("ROLLBACK;"); + throw err; + } +} + +function seedCronJobs(db: DatabaseSync, count: number): void { + const insert = db.prepare(` + INSERT INTO cron_jobs ( + store_key, job_id, name, description, enabled, delete_after_run, created_at_ms, + agent_id, session_key, schedule_kind, schedule_expr, schedule_tz, every_ms, + anchor_ms, at, stagger_ms, session_target, wake_mode, payload_kind, + payload_message, payload_model, payload_fallbacks_json, payload_thinking, + payload_timeout_seconds, payload_allow_unsafe_external_content, + payload_external_content_source_json, payload_light_context, payload_tools_allow_json, + delivery_mode, delivery_channel, delivery_to, delivery_thread_id, delivery_account_id, + delivery_best_effort, delivery_completion_mode, delivery_completion_to, + failure_delivery_mode, failure_delivery_channel, failure_delivery_to, + failure_delivery_account_id, failure_alert_disabled, failure_alert_after, + failure_alert_channel, failure_alert_to, failure_alert_cooldown_ms, + failure_alert_include_skipped, failure_alert_mode, failure_alert_account_id, + next_run_at_ms, running_at_ms, last_run_at_ms, last_run_status, last_error, + last_duration_ms, consecutive_errors, consecutive_skipped, schedule_error_count, + last_delivery_status, last_delivery_error, last_delivered, last_failure_alert_at_ms, + job_json, state_json, runtime_updated_at_ms, schedule_identity, sort_order, updated_at + ) VALUES ( + ?, ?, ?, NULL, ?, NULL, ?, ?, ?, 'every', NULL, NULL, ?, ?, NULL, NULL, + 'isolated', 'now', 'agentTurn', ?, 'openai/gpt-5.5', NULL, NULL, 60, + 0, NULL, 1, NULL, 'announce', 'telegram', ?, NULL, 'bench-account', + 1, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, + NULL, NULL, NULL, ?, NULL, ?, 'completed', NULL, ?, 0, 0, 0, 'sent', + NULL, 1, NULL, ?, '{}', ?, ?, ?, ? + ) + `); + for (let i = 0; i < count; i += 1) { + const jobId = `job-${String(i).padStart(8, "0")}`; + const storeKey = `/state/cron/jobs-${i % 8}.json`; + const updatedAt = 1_700_000_000_000 + i; + insert.run( + storeKey, + jobId, + `Benchmark job ${i}`, + i % 5 === 0 ? 0 : 1, + updatedAt - 100_000, + `agent-${i % 16}`, + `agent:agent-${i % 16}:main`, + 60_000 + (i % 120) * 1_000, + updatedAt - 60_000, + `Benchmark payload ${i}`, + `chat-${i % 32}`, + updatedAt + (i % 2_000) * 1_000, + updatedAt - 1_000, + 50 + (i % 500), + JSON.stringify({ id: jobId, seed: i }), + updatedAt, + `schedule-${i % 512}`, + i, + updatedAt, + ); + } +} + +function seedCronRunLogs(db: DatabaseSync, count: number): void { + const insert = db.prepare(` + INSERT INTO cron_run_logs ( + store_key, job_id, seq, ts, status, error, summary, diagnostics_summary, + delivery_status, delivery_error, delivered, session_id, session_key, run_id, + run_at_ms, duration_ms, next_run_at_ms, model, provider, total_tokens, + entry_json, created_at + ) VALUES (?, ?, ?, ?, ?, NULL, ?, NULL, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `); + for (let i = 0; i < count; i += 1) { + const jobId = `job-${String(i % Math.max(1, Math.floor(count / 20))).padStart(8, "0")}`; + const ts = 1_700_000_000_000 + i; + insert.run( + `/state/cron/jobs-${i % 8}.json`, + jobId, + Math.floor(i / 20), + ts, + i % 17 === 0 ? "failed" : "completed", + `run ${i}`, + i % 17 === 0 ? "failed" : "sent", + i % 17 === 0 ? 0 : 1, + `session-${i}`, + `agent:agent-${i % 16}:main`, + `run-${i}`, + ts, + 20 + (i % 1_000), + ts + 60_000, + "openai/gpt-5.5", + "openai", + 100 + (i % 2_000), + JSON.stringify({ ts, jobId, action: "finished" }), + ts, + ); + } +} + +function seedDeliveryQueue(db: DatabaseSync, count: number): void { + const insert = db.prepare(` + INSERT INTO delivery_queue_entries ( + queue_name, id, status, entry_kind, session_key, channel, target, account_id, + retry_count, last_attempt_at, last_error, recovery_state, platform_send_started_at, + entry_json, enqueued_at, updated_at, failed_at + ) VALUES (?, ?, ?, 'message', ?, ?, ?, ?, ?, ?, NULL, NULL, NULL, ?, ?, ?, ?) + `); + for (let i = 0; i < count; i += 1) { + const status = i % 13 === 0 ? "failed" : i % 3 === 0 ? "sending" : "pending"; + const enqueuedAt = 1_700_000_000_000 + i; + insert.run( + "outbound", + `delivery-${String(i).padStart(8, "0")}`, + status, + `agent:agent-${i % 16}:main`, + i % 2 === 0 ? "telegram" : "discord", + `target-${i % 256}`, + `account-${i % 8}`, + i % 5, + status === "failed" ? enqueuedAt + 500 : null, + JSON.stringify({ id: i, route: { channel: "telegram", to: `target-${i % 256}` } }), + enqueuedAt, + enqueuedAt + 100, + status === "failed" ? enqueuedAt + 1_000 : null, + ); + } +} + +function seedPluginState(db: DatabaseSync, count: number): void { + const insert = db.prepare(` + INSERT INTO plugin_state_entries ( + plugin_id, namespace, entry_key, value_json, created_at, expires_at + ) VALUES (?, ?, ?, ?, ?, ?) + `); + for (let i = 0; i < count; i += 1) { + insert.run( + `plugin-${i % 12}`, + `namespace-${i % 16}`, + `entry-${String(i).padStart(8, "0")}`, + JSON.stringify({ value: i, text: `payload ${i}` }), + 1_700_000_000_000 + i, + i % 10 === 0 ? 1_800_000_000_000 + i : null, + ); + } +} + +function seedChannelIngress(db: DatabaseSync, count: number): void { + const insert = db.prepare(` + INSERT INTO channel_ingress_events ( + queue_name, event_id, channel_id, account_id, status, lane_key, payload_json, + metadata_json, received_at, updated_at, claim_token, claim_owner, claimed_at, + attempts, last_attempt_at, last_error, failed_reason, failed_at, completed_at, + completed_metadata_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, NULL, NULL, NULL, ?, NULL, NULL, NULL, NULL, NULL, NULL) + `); + for (let i = 0; i < count; i += 1) { + insert.run( + "ingress", + `event-${String(i).padStart(8, "0")}`, + i % 2 === 0 ? "telegram" : "discord", + `account-${i % 8}`, + i % 11 === 0 ? "claimed" : "pending", + `lane-${i % 128}`, + JSON.stringify({ text: `message ${i}` }), + 1_700_000_000_000 + i, + 1_700_000_000_000 + i, + i % 3, + ); + } +} + +function seedAgentDatabase(db: DatabaseSync, count: number, agentIndex: number): void { + db.exec("BEGIN IMMEDIATE;"); + try { + const insert = db.prepare(` + INSERT INTO cache_entries (scope, key, value_json, blob, expires_at, updated_at) + VALUES (?, ?, ?, NULL, ?, ?) + `); + for (let i = 0; i < count; i += 1) { + insert.run( + i % 4 === 0 ? "session_entries" : `scope-${i % 16}`, + `agent-${agentIndex}-entry-${String(i).padStart(8, "0")}`, + JSON.stringify({ agentIndex, i, value: `cache ${i}` }), + i % 7 === 0 ? 1_800_000_000_000 + i : null, + 1_700_000_000_000 + i, + ); + } + db.exec("COMMIT;"); + } catch (err) { + db.exec("ROLLBACK;"); + throw err; + } +} + +function readIntegrity(db: DatabaseSync): string { + const row = db.prepare("PRAGMA integrity_check").get() as { integrity_check?: unknown }; + return typeof row.integrity_check === "string" ? row.integrity_check : "missing"; +} + +function checkpoint(db: DatabaseSync): void { + db.prepare("PRAGMA wal_checkpoint(TRUNCATE)").all(); +} + +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].toFixed(3)); +} + +function runTimedQuery( + db: DatabaseSync, + query: string, + params: unknown[], + runs: number, +): TimedQuery { + const statement = db.prepare(query); + const samples: number[] = []; + let rows = 0; + for (let i = 0; i < runs; i += 1) { + const started = nowMs(); + rows = statement.all(...params).length; + samples.push(nowMs() - started); + } + return { + p50Ms: percentile(samples, 50), + p95Ms: percentile(samples, 95), + query, + rows, + }; +} + +function runHotQueries(params: { + agentDb: DatabaseSync; + config: ProfileConfig; + stateDb: DatabaseSync; +}): TimedQuery[] { + return [ + runTimedQuery( + params.stateDb, + `SELECT job_id, name, updated_at + FROM cron_jobs + WHERE store_key = ? + ORDER BY sort_order ASC, updated_at ASC, job_id + LIMIT 50`, + ["/state/cron/jobs-0.json"], + params.config.queryRuns, + ), + runTimedQuery( + params.stateDb, + `SELECT job_id, next_run_at_ms + FROM cron_jobs + WHERE store_key = ? AND enabled = 1 AND next_run_at_ms IS NOT NULL + ORDER BY next_run_at_ms ASC, job_id + LIMIT 50`, + ["/state/cron/jobs-0.json"], + params.config.queryRuns, + ), + runTimedQuery( + params.stateDb, + `SELECT id, entry_json + FROM delivery_queue_entries + WHERE queue_name = ? AND status = ? + ORDER BY enqueued_at ASC, id + LIMIT 100`, + ["outbound", "pending"], + params.config.queryRuns, + ), + runTimedQuery( + params.stateDb, + `SELECT entry_key, value_json + FROM plugin_state_entries + WHERE plugin_id = ? AND namespace = ? + ORDER BY created_at ASC, entry_key + LIMIT 100`, + ["plugin-0", "namespace-0"], + params.config.queryRuns, + ), + runTimedQuery( + params.agentDb, + `SELECT key, value_json + FROM cache_entries + WHERE scope = ? + ORDER BY key ASC + LIMIT 100`, + ["session_entries"], + params.config.queryRuns, + ), + runTimedQuery( + params.agentDb, + `SELECT key, expires_at + FROM cache_entries + WHERE scope = ? AND expires_at IS NOT NULL + ORDER BY expires_at ASC, key + LIMIT 100`, + ["session_entries"], + params.config.queryRuns, + ), + ]; +} + +function printProofLines(report: BenchmarkReport): void { + const p95 = Math.max(...report.queries.map((query) => query.p95Ms)); + console.log(`SQLITE_PERF_PROFILE=${report.profile}`); + console.log(`SQLITE_PERF_STATE_ROWS=${report.rows.stateRows}`); + console.log(`SQLITE_PERF_AGENT_ROWS=${report.rows.agentCacheEntries}`); + console.log(`SQLITE_PERF_INTEGRITY=${report.integrity.state}`); + console.log(`SQLITE_PERF_WAL_BYTES_BEFORE=${report.walBytes.stateBefore}`); + console.log(`SQLITE_PERF_WAL_BYTES_AFTER=${report.walBytes.stateAfter}`); + console.log(`SQLITE_PERF_QUERY_P95_MS=${p95.toFixed(3)}`); + if (report.paths.artifact) { + console.log(`SQLITE_PERF_ARTIFACT=${report.paths.artifact}`); + } +} + +function main(): void { + if (hasFlag("--help")) { + printUsage(); + return; + } + const options = parseOptions(); + const config = applyScale(PROFILES[options.profile]); + const stateDir = + options.stateDir ?? fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-sqlite-perf-")); + const env = { OPENCLAW_STATE_DIR: stateDir }; + const started = nowMs(); + try { + const stateDatabase = openOpenClawStateDatabase({ env }); + const agentDatabases = Array.from({ length: config.agentCount }, (_, index) => + openOpenClawAgentDatabase({ agentId: `perf-agent-${index}`, env }), + ); + + const seedStarted = nowMs(); + seedStateDatabase(stateDatabase.db, config); + const perAgentEntries = Math.ceil(config.agentCacheEntries / config.agentCount); + agentDatabases.forEach((database, index) => + seedAgentDatabase(database.db, perAgentEntries, index), + ); + const seedMs = nowMs() - seedStarted; + + const stateWalBefore = walSize(stateDatabase.path); + const agentWalBefore = agentDatabases.map((database) => walSize(database.path)); + const stateIntegrity = readIntegrity(stateDatabase.db); + const agentIntegrity = agentDatabases.map((database) => readIntegrity(database.db)); + const queries = runHotQueries({ + agentDb: agentDatabases[0]?.db ?? stateDatabase.db, + config, + stateDb: stateDatabase.db, + }); + + const checkpointStarted = nowMs(); + checkpoint(stateDatabase.db); + agentDatabases.forEach((database) => checkpoint(database.db)); + const checkpointMs = nowMs() - checkpointStarted; + + const report: BenchmarkReport = { + integrity: { + agent: agentIntegrity, + state: stateIntegrity, + }, + node: process.version, + paths: { + agentDatabases: agentDatabases.map((database) => database.path), + artifact: options.output, + stateDatabase: stateDatabase.path, + stateDir, + }, + profile: options.profile, + queries, + rows: { + agentCacheEntries: perAgentEntries * config.agentCount, + agentDatabases: config.agentCount, + channelIngressEvents: config.channelIngressEvents, + cronJobs: config.cronJobs, + cronRunLogs: config.cronRunLogs, + deliveryQueueEntries: config.deliveryQueueEntries, + pluginStateEntries: config.pluginStateEntries, + stateRows: stateRowCount(config), + }, + timingsMs: { + checkpoint: Number(checkpointMs.toFixed(3)), + seed: Number(seedMs.toFixed(3)), + total: Number((nowMs() - started).toFixed(3)), + }, + walBytes: { + agentAfter: agentDatabases.map((database) => walSize(database.path)), + agentBefore: agentWalBefore, + stateAfter: walSize(stateDatabase.path), + stateBefore: stateWalBefore, + }, + }; + + if (options.output) { + fs.mkdirSync(path.dirname(options.output), { recursive: true }); + fs.writeFileSync(options.output, `${JSON.stringify(report, null, 2)}\n`, "utf8"); + } + printProofLines(report); + } finally { + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + if (!options.stateDir) { + fs.rmSync(stateDir, { recursive: true, force: true }); + } + } +} + +if (import.meta.url === pathToFileURL(process.argv[1] ?? "").href) { + main(); +} diff --git a/scripts/openclaw-performance-source-summary.mjs b/scripts/openclaw-performance-source-summary.mjs index d026114299d1..5b038331ee61 100644 --- a/scripts/openclaw-performance-source-summary.mjs +++ b/scripts/openclaw-performance-source-summary.mjs @@ -264,6 +264,40 @@ function validateExtensionMemoryArtifact(extensionMemory, filePath) { } } +function validateSqlitePerfArtifact(sqlitePerf, filePath) { + if (sqlitePerf?.profile !== "smoke") { + throw new Error(`[source-performance] invalid SQLite perf profile: ${filePath}`); + } + if (sqlitePerf?.integrity?.state !== "ok") { + throw new Error(`[source-performance] SQLite integrity check did not pass: ${filePath}`); + } + if ( + !Array.isArray(sqlitePerf?.integrity?.agent) || + sqlitePerf.integrity.agent.length === 0 || + sqlitePerf.integrity.agent.some((entry) => entry !== "ok") + ) { + throw new Error(`[source-performance] SQLite agent integrity check did not pass: ${filePath}`); + } + if ( + !isNonNegativeInteger(sqlitePerf?.rows?.stateRows) || + sqlitePerf.rows.stateRows <= 0 || + !isNonNegativeInteger(sqlitePerf?.rows?.agentCacheEntries) || + sqlitePerf.rows.agentCacheEntries <= 0 || + !finiteNumber(sqlitePerf?.timingsMs?.total) || + !finiteNumber(sqlitePerf?.walBytes?.stateBefore) || + sqlitePerf?.walBytes?.stateAfter !== 0 || + !Array.isArray(sqlitePerf?.queries) || + sqlitePerf.queries.length === 0 + ) { + throw new Error(`[source-performance] incomplete SQLite perf metrics: ${filePath}`); + } + for (const entry of sqlitePerf.queries) { + if (!finiteNumber(entry?.p50Ms) || !finiteNumber(entry?.p95Ms) || !finiteNumber(entry?.rows)) { + throw new Error(`[source-performance] incomplete SQLite query metrics: ${filePath}`); + } + } +} + function validateGatewaySummaryArtifact(gatewaySummary, filePath) { if (!Array.isArray(gatewaySummary?.observations)) { throw new Error(`[source-performance] missing gateway observation summary: ${filePath}`); @@ -284,6 +318,7 @@ function loadSourceArtifacts(sourceDir, { required = false } = {}) { const startupPath = path.join(sourceDir, "gateway-cpu", "gateway-startup-bench.json"); const cliPath = path.join(sourceDir, "cli-startup.json"); const extensionMemoryPath = path.join(sourceDir, "extension-memory.json"); + const sqlitePerfPath = path.join(sourceDir, "sqlite-perf-smoke.json"); const artifacts = { startup: required ? readRequiredJson(startupPath, "gateway startup artifact") @@ -292,12 +327,16 @@ function loadSourceArtifacts(sourceDir, { required = false } = {}) { extensionMemory: required ? readRequiredJson(extensionMemoryPath, "extension memory artifact") : readJsonIfExists(extensionMemoryPath), + sqlitePerf: readJsonIfExists(sqlitePerfPath), mockHelloSummaries: loadMockHelloSummaries(sourceDir, { required }), }; if (required) { validateStartupArtifact(artifacts.startup, startupPath); validateCliArtifact(artifacts.cli, cliPath); validateExtensionMemoryArtifact(artifacts.extensionMemory, extensionMemoryPath); + if (artifacts.sqlitePerf) { + validateSqlitePerfArtifact(artifacts.sqlitePerf, sqlitePerfPath); + } } return artifacts; } @@ -463,6 +502,25 @@ function buildExtensionMemoryRows(extensionMemory) { ]); } +function buildSqlitePerfRows(sqlitePerf) { + if (!sqlitePerf) { + return []; + } + const maxQueryP95 = Math.max(...sqlitePerf.queries.map((entry) => entry.p95Ms)); + return [ + [ + sqlitePerf.profile ?? "unknown", + String(sqlitePerf.rows?.stateRows ?? "n/a"), + String(sqlitePerf.rows?.agentCacheEntries ?? "n/a"), + sqlitePerf.integrity?.state ?? "n/a", + formatBytesAsMb(sqlitePerf.walBytes?.stateBefore), + formatBytesAsMb(sqlitePerf.walBytes?.stateAfter), + formatMs(maxQueryP95), + formatMs(sqlitePerf.timingsMs?.total), + ], + ]; +} + function buildMemoryDeltaRows(current, baseline) { if (!baseline) { return []; @@ -569,6 +627,21 @@ export function buildMarkdown(sourceDir, baselineSourceDir) { ["case", "command", "duration p50", "duration p95", "RSS p95", "exits"], buildCliRows(current.cli), ), + "## SQLite State Smoke", + "", + ...table( + [ + "profile", + "state rows", + "agent rows", + "integrity", + "WAL before", + "WAL after", + "query p95 max", + "total", + ], + buildSqlitePerfRows(current.sqlitePerf), + ), "## Observations", "", ...table(["kind", "id", "CPU core", "wall"], buildObservationRows(gatewaySummary)), diff --git a/src/state/openclaw-state-schema.generated.ts b/src/state/openclaw-state-schema.generated.ts index a70e130fd111..bb4a67fe492c 100644 --- a/src/state/openclaw-state-schema.generated.ts +++ b/src/state/openclaw-state-schema.generated.ts @@ -929,6 +929,9 @@ CREATE TABLE IF NOT EXISTS cron_jobs ( CREATE INDEX IF NOT EXISTS idx_cron_jobs_store_updated ON cron_jobs(store_key, sort_order ASC, updated_at DESC, job_id); +CREATE INDEX IF NOT EXISTS idx_cron_jobs_store_order + ON cron_jobs(store_key, sort_order ASC, updated_at ASC, job_id); + CREATE INDEX IF NOT EXISTS idx_cron_jobs_enabled_next_run ON cron_jobs(store_key, enabled, next_run_at_ms, job_id) WHERE next_run_at_ms IS NOT NULL; diff --git a/src/state/openclaw-state-schema.sql b/src/state/openclaw-state-schema.sql index adf75eb9e834..2dd9d03690a2 100644 --- a/src/state/openclaw-state-schema.sql +++ b/src/state/openclaw-state-schema.sql @@ -924,6 +924,9 @@ CREATE TABLE IF NOT EXISTS cron_jobs ( CREATE INDEX IF NOT EXISTS idx_cron_jobs_store_updated ON cron_jobs(store_key, sort_order ASC, updated_at DESC, job_id); +CREATE INDEX IF NOT EXISTS idx_cron_jobs_store_order + ON cron_jobs(store_key, sort_order ASC, updated_at ASC, job_id); + CREATE INDEX IF NOT EXISTS idx_cron_jobs_enabled_next_run ON cron_jobs(store_key, enabled, next_run_at_ms, job_id) WHERE next_run_at_ms IS NOT NULL; diff --git a/src/state/sqlite-query-plan.test.ts b/src/state/sqlite-query-plan.test.ts new file mode 100644 index 000000000000..bcbc3b9a3f77 --- /dev/null +++ b/src/state/sqlite-query-plan.test.ts @@ -0,0 +1,193 @@ +// SQLite query-plan tests pin hot OpenClaw state indexes used by perf proof. +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import type { DatabaseSync } from "node:sqlite"; +import { afterEach, describe, expect, it } from "vitest"; +import { + closeOpenClawAgentDatabasesForTest, + openOpenClawAgentDatabase, +} from "./openclaw-agent-db.js"; +import { + closeOpenClawStateDatabaseForTest, + openOpenClawStateDatabase, +} from "./openclaw-state-db.js"; + +function createTempStateDir(): string { + return fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-sqlite-plan-")); +} + +function explainQueryPlan( + db: DatabaseSync, + sql: string, + params: readonly (number | string | null)[] = [], +): string { + const rows = db.prepare(`EXPLAIN QUERY PLAN ${sql}`).all(...params) as Array<{ + detail?: unknown; + }>; + return rows + .map((row) => (typeof row.detail === "string" ? row.detail : JSON.stringify(row.detail ?? ""))) + .join("\n"); +} + +function expectPlanUsesIndex(params: { + db: DatabaseSync; + indexName: string; + params?: readonly (number | string | null)[]; + sql: string; +}): void { + expect(explainQueryPlan(params.db, params.sql, params.params)).toContain(params.indexName); +} + +function expectPlanIncludes(params: { + db: DatabaseSync; + expected: string; + params?: readonly (number | string | null)[]; + sql: string; +}): void { + expect(explainQueryPlan(params.db, params.sql, params.params)).toContain(params.expected); +} + +afterEach(() => { + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); +}); + +describe("sqlite hot query plans", () => { + it("uses shared state indexes for list and queue queries", () => { + const stateDir = createTempStateDir(); + const database = openOpenClawStateDatabase({ + env: { OPENCLAW_STATE_DIR: stateDir }, + }); + + expectPlanUsesIndex({ + db: database.db, + indexName: "idx_cron_jobs_store_order", + params: ["/state/cron/jobs.json"], + sql: ` + SELECT job_id, name, updated_at + FROM cron_jobs + WHERE store_key = ? + ORDER BY sort_order ASC, updated_at ASC, job_id + LIMIT 25 + `, + }); + expectPlanUsesIndex({ + db: database.db, + indexName: "idx_cron_jobs_enabled_next_run", + params: ["/state/cron/jobs.json"], + sql: ` + SELECT job_id, next_run_at_ms + FROM cron_jobs + WHERE store_key = ? AND enabled = 1 AND next_run_at_ms IS NOT NULL + ORDER BY next_run_at_ms ASC, job_id + LIMIT 25 + `, + }); + expectPlanUsesIndex({ + db: database.db, + indexName: "idx_cron_run_logs_store_ts", + params: ["/state/cron/jobs.json"], + sql: ` + SELECT job_id, seq, ts + FROM cron_run_logs + WHERE store_key = ? + ORDER BY ts DESC, seq DESC + LIMIT 50 + `, + }); + expectPlanUsesIndex({ + db: database.db, + indexName: "idx_cron_run_logs_job_status", + params: ["/state/cron/jobs.json", "job-1", "completed"], + sql: ` + SELECT seq, ts, status + FROM cron_run_logs + WHERE store_key = ? AND job_id = ? AND status = ? + ORDER BY ts DESC, seq DESC + LIMIT 50 + `, + }); + expectPlanUsesIndex({ + db: database.db, + indexName: "idx_delivery_queue_pending", + params: ["outbound", "pending"], + sql: ` + SELECT id, entry_json + FROM delivery_queue_entries + WHERE queue_name = ? AND status = ? + ORDER BY enqueued_at ASC, id + LIMIT 50 + `, + }); + expectPlanUsesIndex({ + db: database.db, + indexName: "idx_delivery_queue_session", + params: ["outbound", "pending", "agent:main:main"], + sql: ` + SELECT id, entry_json + FROM delivery_queue_entries + WHERE queue_name = ? AND status = ? AND session_key = ? + ORDER BY enqueued_at ASC, id + LIMIT 50 + `, + }); + expectPlanUsesIndex({ + db: database.db, + indexName: "idx_plugin_state_listing", + params: ["telegram", "kv"], + sql: ` + SELECT entry_key, value_json + FROM plugin_state_entries + WHERE plugin_id = ? AND namespace = ? + ORDER BY created_at ASC, entry_key + LIMIT 50 + `, + }); + expectPlanUsesIndex({ + db: database.db, + indexName: "idx_channel_ingress_pending", + params: ["ingress", "pending"], + sql: ` + SELECT event_id, payload_json + FROM channel_ingress_events + WHERE queue_name = ? AND status = ? + ORDER BY received_at ASC, event_id + LIMIT 50 + `, + }); + }); + + it("uses per-agent cache indexes for session metadata and expiry scans", () => { + const stateDir = createTempStateDir(); + const database = openOpenClawAgentDatabase({ + agentId: "worker-1", + env: { OPENCLAW_STATE_DIR: stateDir }, + }); + + expectPlanIncludes({ + db: database.db, + expected: "sqlite_autoindex_cache_entries_1", + params: ["session_entries"], + sql: ` + SELECT key, value_json + FROM cache_entries + WHERE scope = ? + ORDER BY key ASC + LIMIT 50 + `, + }); + expectPlanUsesIndex({ + db: database.db, + indexName: "idx_agent_cache_expiry", + params: ["session_entries"], + sql: ` + SELECT key, expires_at + FROM cache_entries + WHERE scope = ? AND expires_at IS NOT NULL + ORDER BY expires_at ASC, key + LIMIT 50 + `, + }); + }); +}); diff --git a/test/scripts/openclaw-performance-source-summary.test.ts b/test/scripts/openclaw-performance-source-summary.test.ts index 0e4ea33379df..fe649b2c3bf0 100644 --- a/test/scripts/openclaw-performance-source-summary.test.ts +++ b/test/scripts/openclaw-performance-source-summary.test.ts @@ -62,6 +62,23 @@ function writeSourceFixture(sourceDir: string) { { dir: "extensions/browser", maxRssMb: 80, deltaFromBaselineMb: 12, status: "ok" }, ], }); + writeJson(path.join(sourceDir, "sqlite-perf-smoke.json"), { + integrity: { agent: ["ok"], state: "ok" }, + profile: "smoke", + queries: [{ p50Ms: 0.1, p95Ms: 0.2, query: "SELECT 1", rows: 1 }], + rows: { + agentCacheEntries: 1000, + agentDatabases: 2, + channelIngressEvents: 1000, + cronJobs: 100, + cronRunLogs: 1000, + deliveryQueueEntries: 1000, + pluginStateEntries: 1000, + stateRows: 4100, + }, + timingsMs: { checkpoint: 1, seed: 100, total: 150 }, + walBytes: { agentAfter: [0], agentBefore: [1024], stateAfter: 0, stateBefore: 4096 }, + }); writeJson(path.join(sourceDir, "mock-hello", "run-001", "qa-suite-summary.json"), { counts: { failed: 0, passed: 1, total: 1 }, metrics: { @@ -118,6 +135,8 @@ describe("buildMarkdown", () => { expect(buildMarkdown(sourceDir, null)).toContain("run-001"); expect(buildMarkdown(sourceDir, null)).toContain("gateway health json"); + expect(buildMarkdown(sourceDir, null)).toContain("## SQLite State Smoke"); + expect(buildMarkdown(sourceDir, null)).toContain("4100"); }); it("rejects a missing source directory", () => { @@ -176,4 +195,45 @@ describe("buildMarkdown", () => { "[source-performance] incomplete gateway startup metrics for default:", ); }); + + it("allows source performance fixtures without older-ref SQLite smoke artifacts", () => { + const sourceDir = mkTmpRoot(); + writeSourceFixture(sourceDir); + fs.rmSync(path.join(sourceDir, "sqlite-perf-smoke.json")); + + expect(buildMarkdown(sourceDir, null)).toContain("## SQLite State Smoke"); + expect(buildMarkdown(sourceDir, null)).toContain("No data."); + }); + + it("rejects malformed SQLite perf smoke artifacts", () => { + const sourceDir = mkTmpRoot(); + writeSourceFixture(sourceDir); + writeJson(path.join(sourceDir, "sqlite-perf-smoke.json"), { + integrity: { agent: ["ok"], state: "ok" }, + profile: "smoke", + rows: { stateRows: 4100 }, + walBytes: { stateAfter: 1 }, + }); + + expect(() => buildMarkdown(sourceDir, null)).toThrow( + "[source-performance] incomplete SQLite perf metrics:", + ); + }); + + it("rejects SQLite perf smoke artifacts with failing agent integrity", () => { + const sourceDir = mkTmpRoot(); + writeSourceFixture(sourceDir); + writeJson(path.join(sourceDir, "sqlite-perf-smoke.json"), { + integrity: { agent: ["ok", "database disk image is malformed"], state: "ok" }, + profile: "smoke", + queries: [{ p50Ms: 0.1, p95Ms: 0.2, query: "SELECT 1", rows: 1 }], + rows: { agentCacheEntries: 1000, stateRows: 4100 }, + timingsMs: { total: 150 }, + walBytes: { stateAfter: 0, stateBefore: 4096 }, + }); + + expect(() => buildMarkdown(sourceDir, null)).toThrow( + "[source-performance] SQLite agent integrity check did not pass:", + ); + }); });