diff --git a/src/plugin-state/plugin-state-store.sqlite.ts b/src/plugin-state/plugin-state-store.sqlite.ts index b078b4a54b75..2c02452cd7c1 100644 --- a/src/plugin-state/plugin-state-store.sqlite.ts +++ b/src/plugin-state/plugin-state-store.sqlite.ts @@ -9,6 +9,7 @@ import { } from "../infra/kysely-sync.js"; import { requireNodeSqlite } from "../infra/node-sqlite.js"; import { normalizeSqliteNumber } from "../infra/sqlite-number.js"; +import { withExistingOpenClawStateDatabaseReadOnly } from "../state/openclaw-state-db-readonly.js"; import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js"; import { closeOpenClawStateDatabase, @@ -398,6 +399,33 @@ function openPluginStateDatabase( } } +/** Read plugin state without joining the shared writable database lifecycle. */ +function withPluginStateDatabaseReadOnly( + operationName: PluginStateStoreOperation, + operation: (store: PluginStateDatabase) => T, + options: OpenClawStateDatabaseOptions = {}, +): T | undefined { + const pathname = resolveOpenClawStateSqlitePath(options.env ?? process.env); + let operationStarted = false; + try { + return withExistingOpenClawStateDatabaseReadOnly(({ db, path }) => { + operationStarted = true; + return operation({ db, path }); + }, options); + } catch (error) { + if (!operationStarted) { + throw wrapPluginStateError( + error, + operationName, + "PLUGIN_STATE_OPEN_FAILED", + "Failed to open the plugin state database.", + pathname, + ); + } + throw error; + } +} + function countRow(row: CountRow | undefined): number { const raw = row?.count ?? 0; return typeof raw === "bigint" ? Number(raw) : raw; @@ -907,21 +935,28 @@ export function pluginStateLookup(params: { key: string; env?: NodeJS.ProcessEnv; }): unknown { + const pathname = resolveOpenClawStateSqlitePath(params.env ?? process.env); try { - const { db } = openPluginStateDatabase("lookup", envOptions(params.env)); - const row = selectPluginStateEntry(db, { - pluginId: params.pluginId, - namespace: params.namespace, - key: params.key, - now: Date.now(), - }); - return row ? parseStoredJson(row.value_json, "lookup") : undefined; + return withPluginStateDatabaseReadOnly( + "lookup", + ({ db }) => { + const row = selectPluginStateEntry(db, { + pluginId: params.pluginId, + namespace: params.namespace, + key: params.key, + now: Date.now(), + }); + return row ? parseStoredJson(row.value_json, "lookup") : undefined; + }, + envOptions(params.env), + ); } catch (error) { throw wrapPluginStateError( error, "lookup", "PLUGIN_STATE_READ_FAILED", "Failed to read plugin state entry.", + pathname, ); } } @@ -1023,20 +1058,29 @@ export function pluginStateEntries(params: { namespace: string; env?: NodeJS.ProcessEnv; }): PluginStateEntry[] { + const pathname = resolveOpenClawStateSqlitePath(params.env ?? process.env); try { - const { db } = openPluginStateDatabase("entries", envOptions(params.env)); - const rows = selectPluginStateEntries(db, { - pluginId: params.pluginId, - namespace: params.namespace, - now: Date.now(), - }); - return rows.map((row) => rowToEntry(row, "entries")); + return ( + withPluginStateDatabaseReadOnly( + "entries", + ({ db }) => { + const rows = selectPluginStateEntries(db, { + pluginId: params.pluginId, + namespace: params.namespace, + now: Date.now(), + }); + return rows.map((row) => rowToEntry(row, "entries")); + }, + envOptions(params.env), + ) ?? [] + ); } catch (error) { throw wrapPluginStateError( error, "entries", "PLUGIN_STATE_READ_FAILED", "Failed to list plugin state entries.", + pathname, ); } } @@ -1065,23 +1109,31 @@ export function pluginStateEntriesInKeyRange(params: { message: "Plugin state key range must have an increasing exclusive upper bound.", }); } + const pathname = resolveOpenClawStateSqlitePath(params.env ?? process.env); try { - const { db } = openPluginStateDatabase("entries", envOptions(params.env)); - return selectPluginStateEntriesInKeyRange(db, { - pluginId: params.pluginId, - namespace: params.namespace, - keyStartInclusive: params.keyStartInclusive, - keyEndExclusive: params.keyEndExclusive, - limit: params.limit, - order: params.order ?? "asc", - now: Date.now(), - }).map((row) => rowToEntry(row, "entries")); + return ( + withPluginStateDatabaseReadOnly( + "entries", + ({ db }) => + selectPluginStateEntriesInKeyRange(db, { + pluginId: params.pluginId, + namespace: params.namespace, + keyStartInclusive: params.keyStartInclusive, + keyEndExclusive: params.keyEndExclusive, + limit: params.limit, + order: params.order ?? "asc", + now: Date.now(), + }).map((row) => rowToEntry(row, "entries")), + envOptions(params.env), + ) ?? [] + ); } catch (error) { throw wrapPluginStateError( error, "entries", "PLUGIN_STATE_READ_FAILED", "Failed to list plugin state entries by key range.", + pathname, ); } } @@ -1147,15 +1199,22 @@ function setMaxPluginStateEntriesPerPluginForTests(value?: number): void { } export function countPluginStateLiveEntries(pluginId: string, env?: NodeJS.ProcessEnv): number { + const pathname = resolveOpenClawStateSqlitePath(env ?? process.env); try { - const { db } = openPluginStateDatabase("entries", envOptions(env)); - return countLivePluginStateEntries(db, { pluginId, now: Date.now() }); + return ( + withPluginStateDatabaseReadOnly( + "entries", + ({ db }) => countLivePluginStateEntries(db, { pluginId, now: Date.now() }), + envOptions(env), + ) ?? 0 + ); } catch (error) { throw wrapPluginStateError( error, "entries", "PLUGIN_STATE_READ_FAILED", "Failed to count plugin state entries.", + pathname, ); } } diff --git a/src/plugin-state/plugin-state-store.test.ts b/src/plugin-state/plugin-state-store.test.ts index 09ad1f6166ac..70e4a81d6339 100644 --- a/src/plugin-state/plugin-state-store.test.ts +++ b/src/plugin-state/plugin-state-store.test.ts @@ -1,9 +1,19 @@ // Plugin state store tests cover per-plugin persisted state reads and writes. -import { rmSync, statSync } from "node:fs"; +import { chmodSync, existsSync, rmSync, statSync } from "node:fs"; import path from "node:path"; +import { DatabaseSync } from "node:sqlite"; import { MAX_DATE_TIMESTAMP_MS } from "@openclaw/normalization-core/number-coercion"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { openOpenClawStateDatabase } from "../state/openclaw-state-db.js"; +import { + clearOpenClawDatabaseQuarantine, + recordOpenClawDatabaseQuarantine, +} from "../state/openclaw-quarantine-store.js"; +import { + clearOpenClawStateDatabaseOpenFailure, + OPENCLAW_STATE_SCHEMA_VERSION, + openOpenClawStateDatabase, + recordOpenClawStateDatabaseOpenFailure, +} from "../state/openclaw-state-db.js"; import { resolveOpenClawStateSqlitePath } from "../state/openclaw-state-db.paths.js"; import { createOpenClawTestState, @@ -12,9 +22,11 @@ import { } from "../test-utils/openclaw-test-state.js"; import { closePluginStateDatabase, + countPluginStateLiveEntries, createCorePluginStateSyncKeyedStore, createPluginStateKeyedStore, createPluginStateSyncKeyedStore, + isPluginStateDatabaseOpen, pluginStateEntriesInKeyRange, registerPluginStateSyncSequencedJournalEntry, resetPluginStateStoreForTests, @@ -849,6 +861,159 @@ describe("plugin state keyed store", () => { }); }); + it("keeps plugin-state reads outside the writable database lifecycle", async () => { + await withPluginStateTestState(async () => { + const store = createPluginStateKeyedStore("discord", { + namespace: "read-only", + maxEntries: 10, + }); + await store.register("k", { ok: true }); + resetPluginStateStoreForTests(); + + expect(isPluginStateDatabaseOpen()).toBe(false); + await expect(store.lookup("k")).resolves.toEqual({ ok: true }); + await expect(store.entries()).resolves.toMatchObject([{ key: "k", value: { ok: true } }]); + expect( + pluginStateEntriesInKeyRange({ + pluginId: "discord", + namespace: "read-only", + keyStartInclusive: "k", + keyEndExclusive: "l", + limit: 1, + }), + ).toMatchObject([{ key: "k", value: { ok: true } }]); + expect(countPluginStateLiveEntries("discord")).toBe(1); + expect(isPluginStateDatabaseOpen()).toBe(false); + }); + }); + + it("treats a missing plugin-state database as empty without creating it", async () => { + await withOpenClawTestState( + { label: "plugin-state-read-only-missing", applyEnv: false }, + async (state) => { + const store = createPluginStateKeyedStore("discord", { + namespace: "read-only-missing", + maxEntries: 10, + env: state.env, + }); + const databasePath = resolveOpenClawStateSqlitePath(state.env); + + expect(existsSync(databasePath)).toBe(false); + await expect(store.lookup("k")).resolves.toBeUndefined(); + await expect(store.entries()).resolves.toEqual([]); + expect(countPluginStateLiveEntries("discord", state.env)).toBe(0); + expect(existsSync(databasePath)).toBe(false); + }, + ); + }); + + it("fails closed for process-local and persisted database quarantine", async () => { + await withPluginStateTestState(async () => { + const store = createPluginStateKeyedStore("discord", { + namespace: "quarantine", + maxEntries: 10, + }); + await store.register("k", { ok: true }); + const databasePath = resolveOpenClawStateSqlitePath(testState?.env); + closePluginStateDatabase(); + + recordOpenClawStateDatabaseOpenFailure(databasePath, new Error("latched failure")); + await expect(store.lookup("k")).rejects.toMatchObject({ + code: "PLUGIN_STATE_OPEN_FAILED", + path: databasePath, + }); + clearOpenClawStateDatabaseOpenFailure(databasePath); + + expect( + recordOpenClawDatabaseQuarantine({ + env: testState?.env, + kind: "state", + path: databasePath, + reason: "persisted failure", + }), + ).toBe(true); + await expect(store.lookup("k")).rejects.toMatchObject({ + code: "PLUGIN_STATE_OPEN_FAILED", + path: databasePath, + }); + expect(clearOpenClawDatabaseQuarantine(databasePath, { env: testState?.env })).toBe(true); + }); + }); + + it("fails closed for a newer shared-state schema", async () => { + await withPluginStateTestState(async () => { + const store = createPluginStateKeyedStore("discord", { + namespace: "newer-schema", + maxEntries: 10, + }); + await store.register("k", { ok: true }); + const databasePath = resolveOpenClawStateSqlitePath(testState?.env); + openOpenClawStateDatabase().db.exec( + `PRAGMA user_version = ${OPENCLAW_STATE_SCHEMA_VERSION + 1};`, + ); + closePluginStateDatabase(); + + try { + await expect(store.lookup("k")).rejects.toMatchObject({ + code: "PLUGIN_STATE_OPEN_FAILED", + path: databasePath, + }); + } finally { + const database = new DatabaseSync(databasePath); + try { + database.exec(`PRAGMA user_version = ${OPENCLAW_STATE_SCHEMA_VERSION};`); + } finally { + database.close(); + } + } + }); + }); + + it.runIf(process.platform !== "win32")( + "reports inaccessible explicit state directories instead of treating them as empty", + async () => { + await withPluginStateTestState(async () => { + const store = createPluginStateKeyedStore("discord", { + namespace: "inaccessible", + maxEntries: 10, + }); + await store.register("k", { ok: true }); + const databasePath = resolveOpenClawStateSqlitePath(testState?.env); + closePluginStateDatabase(); + chmodSync(testState?.stateDir ?? "", 0o000); + try { + await expect(store.lookup("k")).rejects.toMatchObject({ + code: "PLUGIN_STATE_OPEN_FAILED", + path: databasePath, + }); + } finally { + chmodSync(testState?.stateDir ?? "", 0o700); + } + }); + }, + ); + + it.runIf(process.platform !== "win32")( + "reuses a process-held state database when its directory becomes inaccessible", + async () => { + await withPluginStateTestState(async () => { + const store = createPluginStateKeyedStore("discord", { + namespace: "inaccessible-open-handle", + maxEntries: 10, + }); + await store.register("k", { ok: true }); + const database = openOpenClawStateDatabase(); + chmodSync(testState?.stateDir ?? "", 0o000); + try { + await expect(store.lookup("k")).resolves.toEqual({ ok: true }); + expect(database.db.isOpen).toBe(true); + } finally { + chmodSync(testState?.stateDir ?? "", 0o700); + } + }); + }, + ); + it("does not close a shared state database opened before the plugin-state probe", async () => { await withPluginStateTestState(async () => { const database = openOpenClawStateDatabase(); diff --git a/src/state/openclaw-state-db-readonly.ts b/src/state/openclaw-state-db-readonly.ts index fd76fa731a95..12289d6acd76 100644 --- a/src/state/openclaw-state-db-readonly.ts +++ b/src/state/openclaw-state-db-readonly.ts @@ -1,3 +1,4 @@ +import { statSync } from "node:fs"; import path from "node:path"; import type { DatabaseSync } from "node:sqlite"; import { clearNodeSqliteKyselyCacheForDatabase } from "../infra/kysely-sync.js"; @@ -7,6 +8,7 @@ import { readSqliteUserVersion, } from "../infra/sqlite-user-version.js"; import { + assertOpenClawStateDatabaseFreshOpenAllowed, evictOpenClawStateDatabaseAfterCorruption, getOpenClawStateDatabaseIfOpen, OPENCLAW_SQLITE_BUSY_TIMEOUT_MS, @@ -20,6 +22,24 @@ type OpenClawStateReadOnlyDatabase = { path: string; }; +type ReusedOpenClawStateReadOnlyDatabase = { reused: false } | { reused: true; value: T }; + +function resolveReadOnlyPath(options: OpenClawStateDatabaseOptions): string { + return path.resolve(options.path ?? resolveOpenClawStateSqlitePath(options.env ?? process.env)); +} + +function existingPathOrUndefined(pathname: string): string | undefined { + try { + statSync(pathname); + return pathname; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + return undefined; + } + throw error; + } +} + function assertSupportedSchemaVersion(db: DatabaseSync, pathname: string): void { const userVersion = readSqliteUserVersion(db); if (userVersion > OPENCLAW_STATE_SCHEMA_VERSION) { @@ -32,6 +52,45 @@ function assertSupportedSchemaVersion(db: DatabaseSync, pathname: string): void } } +function withOpenClawStateDatabaseReadOnlyIfOpen( + operation: (database: OpenClawStateReadOnlyDatabase) => T, + options: OpenClawStateDatabaseOptions, + pathname: string, +): ReusedOpenClawStateReadOnlyDatabase { + const opened = getOpenClawStateDatabaseIfOpen(options); + if (!opened || opened.db.isTransaction) { + return { reused: false }; + } + try { + // Process-local terminal failures evict this handle. Persisted quarantine + // is checked on the next physical open so hot reads do not poll metadata. + // A newer build can migrate this file while the handle stays open, so the + // forward-compatibility gate still runs before any reused read. + assertSupportedSchemaVersion(opened.db, pathname); + return { reused: true, value: operation(opened) }; + } catch (error) { + evictOpenClawStateDatabaseAfterCorruption(opened, error); + throw error; + } +} + +function withFreshOpenClawStateDatabaseReadOnly( + operation: (database: OpenClawStateReadOnlyDatabase) => T, + options: OpenClawStateDatabaseOptions, + pathname: string, +): T { + assertOpenClawStateDatabaseFreshOpenAllowed(options); + const db = openNodeSqliteDatabase(pathname, { readOnly: true }); + try { + db.exec(`PRAGMA busy_timeout = ${OPENCLAW_SQLITE_BUSY_TIMEOUT_MS};`); + assertSupportedSchemaVersion(db, pathname); + return operation({ db, path: pathname }); + } finally { + clearNodeSqliteKyselyCacheForDatabase(db); + db.close(); + } +} + /** * Read shared state without joining the writable lifecycle. * @@ -42,32 +101,34 @@ export function withOpenClawStateDatabaseReadOnly( operation: (database: OpenClawStateReadOnlyDatabase) => T, options: OpenClawStateDatabaseOptions = {}, ): T { - const pathname = path.resolve( - options.path ?? resolveOpenClawStateSqlitePath(options.env ?? process.env), - ); + const pathname = resolveReadOnlyPath(options); // Reusing a handle this process already holds keeps row loops cheap: opening // and closing a connection per call made shared-state reads scale with row // count. An in-flight transaction is skipped so callers never observe // uncommitted rows a fresh read-only connection could not have seen. - const opened = getOpenClawStateDatabaseIfOpen(options); - if (opened && !opened.db.isTransaction) { - try { - // A newer build can migrate this file while the handle stays open, so the - // forward-compatibility gate still runs before any reused read. - assertSupportedSchemaVersion(opened.db, pathname); - return operation(opened); - } catch (error) { - evictOpenClawStateDatabaseAfterCorruption(opened, error); - throw error; - } - } - const db = openNodeSqliteDatabase(pathname, { readOnly: true }); - try { - db.exec(`PRAGMA busy_timeout = ${OPENCLAW_SQLITE_BUSY_TIMEOUT_MS};`); - assertSupportedSchemaVersion(db, pathname); - return operation({ db, path: pathname }); - } finally { - clearNodeSqliteKyselyCacheForDatabase(db); - db.close(); + const reused = withOpenClawStateDatabaseReadOnlyIfOpen(operation, options, pathname); + if (reused.reused) { + return reused.value; } + return withFreshOpenClawStateDatabaseReadOnly(operation, options, pathname); +} + +/** Read existing shared state while preserving non-missing filesystem failures. */ +export function withExistingOpenClawStateDatabaseReadOnly( + operation: (database: OpenClawStateReadOnlyDatabase) => T, + options: OpenClawStateDatabaseOptions = {}, +): T | undefined { + const pathname = resolveReadOnlyPath(options); + const reused = withOpenClawStateDatabaseReadOnlyIfOpen(operation, options, pathname); + if (reused.reused) { + return reused.value; + } + const existingPath = existingPathOrUndefined(pathname); + return existingPath === undefined + ? undefined + : withFreshOpenClawStateDatabaseReadOnly( + operation, + { ...options, path: existingPath }, + existingPath, + ); } diff --git a/src/state/openclaw-state-db.ts b/src/state/openclaw-state-db.ts index ed6e355e2a5a..4194810416eb 100644 --- a/src/state/openclaw-state-db.ts +++ b/src/state/openclaw-state-db.ts @@ -158,6 +158,41 @@ export function clearOpenClawStateDatabaseOpenFailure(pathname: string): void { terminalOpenLatch.clear(pathname); } +/** Reject shared-state access after a process-local terminal failure. */ +function assertOpenClawStateDatabaseOpenAllowed(options: OpenClawStateDatabaseOptions = {}): void { + const pathname = resolveDatabasePath(options); + const terminalFailure = terminalOpenLatch.get(pathname); + if (terminalFailure) { + throw terminalFailure; + } +} + +/** Reject a fresh shared-state open after known corruption until repair clears it. */ +export function assertOpenClawStateDatabaseFreshOpenAllowed( + options: OpenClawStateDatabaseOptions = {}, +): void { + assertOpenClawStateDatabaseOpenAllowed(options); + const env = options.env ?? process.env; + const pathname = resolveDatabasePath(options); + let quarantineFailure: Error | undefined; + try { + const quarantine = readOpenClawDatabaseQuarantine(pathname, { env }); + if (quarantine) { + quarantineFailure = createOpenClawDatabaseVerificationError( + "state", + pathname, + quarantine.reason, + ); + } + } catch { + // A broken quarantine store must not brick every state read. + // The process latch and daily verifier still cover known damage. + } + if (quarantineFailure) { + throw quarantineFailure; + } +} + type OpenClawStateMetadataDatabase = Pick; const stateDbLog = createSubsystemLogger("state/db"); @@ -425,26 +460,11 @@ function ensureSchema(db: DatabaseSync, pathname: string): void { export async function openExistingOpenClawStateDatabaseReadOnly( options: OpenClawStateDatabaseOptions = {}, ): Promise { - const env = options.env ?? process.env; const pathname = resolveDatabasePath(options); if (!existsSync(pathname)) { return undefined; } - const terminalFailure = terminalOpenLatch.get(pathname); - if (terminalFailure) { - throw terminalFailure; - } - try { - const quarantine = readOpenClawDatabaseQuarantine(pathname, { env }); - if (quarantine) { - throw createOpenClawDatabaseVerificationError("state", pathname, quarantine.reason); - } - } catch (error) { - if (error instanceof Error && error.name === "SqliteIntegrityError") { - throw error; - } - // A broken quarantine store must not brick read-only diagnostics. - } + assertOpenClawStateDatabaseFreshOpenAllowed(options); const prepared = await prepareSqliteReadOnlyLocation(pathname); let db: DatabaseSync; try { @@ -549,10 +569,7 @@ export function openOpenClawStateDatabase( const pathname = resolveDatabasePath(options); // Latched paths are quarantined: the recorder closed any live handle, and // every open fails fast here until doctor repairs the file and clears it. - const terminalFailure = terminalOpenLatch.get(pathname); - if (terminalFailure) { - throw terminalFailure; - } + assertOpenClawStateDatabaseOpenAllowed(options); const cached = cachedDatabases.get(pathname); if (cached?.db.isOpen) { return cached; @@ -563,23 +580,7 @@ export function openOpenClawStateDatabase( clearNodeSqliteKyselyCacheForDatabase(cached.db); cachedDatabases.delete(pathname); } - let quarantineFailure: Error | undefined; - try { - const quarantine = readOpenClawDatabaseQuarantine(pathname, { env }); - if (quarantine) { - quarantineFailure = createOpenClawDatabaseVerificationError( - "state", - pathname, - quarantine.reason, - ); - } - } catch { - // A broken quarantine store must not brick every state open. - // The process latch and daily verifier still cover known damage. - } - if (quarantineFailure) { - throw quarantineFailure; - } + assertOpenClawStateDatabaseFreshOpenAllowed(options); ensureOpenClawStatePermissions(pathname, env); const db = openNodeSqliteDatabase(pathname); enableNodeSqliteKyselyStatementCache(db);