From c63fdc631f03ee7efd49f2e1773b76ea69b9daf8 Mon Sep 17 00:00:00 2001 From: WhatsSkiLL Date: Thu, 30 Jul 2026 11:06:44 +0200 Subject: [PATCH] fix(sessions): release closed SQLite entry caches (#115794) * fix(sessions): release closed sqlite entry caches * chore(ci): register sqlite cache lifetime proof --------- Co-authored-by: IWhatsskill <284122573+IWhatsskill@users.noreply.github.com> --- config/knip.config.ts | 1 + ...lite-session-entry-cache-lifetime-proof.ts | 174 ++++++++++++++++++ ...ssion-accessor.sqlite-data-version.test.ts | 43 +++++ .../session-accessor.sqlite-entry-cache.ts | 40 ++-- 4 files changed, 233 insertions(+), 25 deletions(-) create mode 100644 scripts/sqlite-session-entry-cache-lifetime-proof.ts diff --git a/config/knip.config.ts b/config/knip.config.ts index e87f55d13462..fea9325546da 100644 --- a/config/knip.config.ts +++ b/config/knip.config.ts @@ -81,6 +81,7 @@ const repositoryScriptEntries = [ "scripts/qa-parity-report.ts!", "scripts/resolve-frozen-codex-live-suite.mjs!", "scripts/secrets/openclaw-bws-resolver.mjs!", + "scripts/sqlite-session-entry-cache-lifetime-proof.ts!", "scripts/sync-labels.ts!", "scripts/test-built-bundled-channel-entry-smoke.mjs!", "scripts/update-clawtributors.ts!", diff --git a/scripts/sqlite-session-entry-cache-lifetime-proof.ts b/scripts/sqlite-session-entry-cache-lifetime-proof.ts new file mode 100644 index 000000000000..8f3df2c20b54 --- /dev/null +++ b/scripts/sqlite-session-entry-cache-lifetime-proof.ts @@ -0,0 +1,174 @@ +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { pathToFileURL } from "node:url"; + +type SessionAccessorModule = typeof import("../src/config/sessions/session-accessor.js"); +type EntryCacheModule = + typeof import("../src/config/sessions/session-accessor.sqlite-entry-cache.js"); +type AgentDatabaseModule = typeof import("../src/state/openclaw-agent-db.js"); +type ReadOnlyDatabaseModule = typeof import("../src/state/openclaw-agent-db-readonly.js"); +type StateDatabaseModule = typeof import("../src/state/openclaw-state-db.js"); + +const repoRoot = process.env.PROOF_REPO_ROOT ?? process.cwd(); +const expectation = process.env.PROOF_EXPECTATION ?? "released"; +if (expectation !== "retained" && expectation !== "released") { + throw new Error("PROOF_EXPECTATION must be retained or released"); +} +const forceGc = globalThis.gc; +if (typeof forceGc !== "function") { + throw new Error("run with node --expose-gc"); +} + +const dataRoot = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-sqlite-entry-cache-proof-")); +const importSource = async (relativePath: string) => + import(pathToFileURL(path.join(repoRoot, relativePath)).href); +const { upsertSessionEntry } = (await importSource( + "src/config/sessions/session-accessor.js", +)) as SessionAccessorModule; +const { readSqliteSessionEntryCache } = (await importSource( + "src/config/sessions/session-accessor.sqlite-entry-cache.js", +)) as EntryCacheModule; +const { + OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP, + closeOpenClawAgentDatabases, + isOpenClawAgentDatabaseOpen, + openOpenClawAgentDatabase, +} = (await importSource("src/state/openclaw-agent-db.js")) as AgentDatabaseModule; +const { withOpenClawAgentDatabaseReadOnly } = (await importSource( + "src/state/openclaw-agent-db-readonly.js", +)) as ReadOnlyDatabaseModule; +const { closeOpenClawStateDatabase } = (await importSource( + "src/state/openclaw-state-db.js", +)) as StateDatabaseModule; + +const readOnlyEntryCount = 20; +const readOnlyPayloadBytes = 1_000_000; +const lruHandleCount = OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP + 2; +const lruPayloadBytes = 100_000; + +const createScope = (group: string, index: number) => ({ + agentId: "main", + env: { + ...process.env, + OPENCLAW_STATE_DIR: path.join(dataRoot, group, `agent-${index}`), + }, + sessionKey: `agent:main:${group}-${index}`, +}); + +const settleGc = async () => { + for (let attempt = 0; attempt < 12; attempt += 1) { + forceGc(); + await new Promise((resolve) => { + setImmediate(resolve); + }); + } +}; + +try { + const readOnlyScopes = Array.from({ length: readOnlyEntryCount }, (_, index) => + createScope("read-only", index), + ); + for (const [index, scope] of readOnlyScopes.entries()) { + fs.mkdirSync(scope.env.OPENCLAW_STATE_DIR, { recursive: true }); + await upsertSessionEntry(scope, { + label: `${index}:${"x".repeat(readOnlyPayloadBytes)}`, + sessionId: `read-only-${index}`, + updatedAt: index + 1, + }); + } + closeOpenClawAgentDatabases(); + closeOpenClawStateDatabase(); + + await settleGc(); + const readOnlyBaselineHeap = process.memoryUsage().heapUsed; + const readOnlyConnectionRefs: Array> = []; + + for (const scope of readOnlyScopes) { + const result = withOpenClawAgentDatabaseReadOnly((database) => { + const snapshot = readSqliteSessionEntryCache(database, { cache: true }); + assert.deepEqual(snapshot.keys, [scope.sessionKey]); + readOnlyConnectionRefs.push(new WeakRef(database.db)); + return snapshot.keys.length; + }, scope); + assert.equal(result.found, true); + } + + await settleGc(); + const readOnlyHeapDeltaBytes = process.memoryUsage().heapUsed - readOnlyBaselineHeap; + const readOnlyRetainedConnections = readOnlyConnectionRefs.filter((reference) => + reference.deref(), + ).length; + + const lruConnectionRefs: Array> = []; + const lruPaths: string[] = []; + for (let index = 0; index < lruHandleCount; index += 1) { + const scope = createScope("lru", index); + fs.mkdirSync(scope.env.OPENCLAW_STATE_DIR, { recursive: true }); + await upsertSessionEntry(scope, { + label: `${index}:${"y".repeat(lruPayloadBytes)}`, + sessionId: `lru-${index}`, + updatedAt: index + 1, + }); + const database = openOpenClawAgentDatabase(scope); + const snapshot = readSqliteSessionEntryCache(database, { cache: true }); + assert.deepEqual(snapshot.keys, [scope.sessionKey]); + lruConnectionRefs.push(new WeakRef(database.db)); + lruPaths.push(database.path); + } + + const lruEvictedCount = lruHandleCount - OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP; + assert.equal(isOpenClawAgentDatabaseOpen(lruPaths[0]!), false); + assert.equal(isOpenClawAgentDatabaseOpen(lruPaths.at(-1)!), true); + await settleGc(); + const lruEvictedRetainedConnections = lruConnectionRefs + .slice(0, lruEvictedCount) + .filter((reference) => reference.deref()).length; + const lruLiveRetainedConnections = lruConnectionRefs + .slice(lruEvictedCount) + .filter((reference) => reference.deref()).length; + + const retained = + readOnlyRetainedConnections === readOnlyEntryCount && + readOnlyHeapDeltaBytes > 6_000_000 && + lruEvictedRetainedConnections === lruEvictedCount && + lruLiveRetainedConnections === OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP; + const released = + readOnlyRetainedConnections <= 1 && + readOnlyHeapDeltaBytes < 6_000_000 && + lruEvictedRetainedConnections === 0 && + lruLiveRetainedConnections === OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP; + const verdict = expectation === "released" ? released : retained; + + process.stdout.write( + `${JSON.stringify({ + schema: "sqlite-session-entry-cache-lifetime-proof-v2", + test_runner: "none", + affected_owners: { + database: "real-node-sqlite", + read_only_lifecycle: "withOpenClawAgentDatabaseReadOnly", + writable_lifecycle: "openOpenClawAgentDatabase LRU", + cache: "readSqliteSessionEntryCache", + }, + read_only_entry_count: readOnlyEntryCount, + read_only_payload_bytes_per_entry: readOnlyPayloadBytes, + read_only_retained_connections: readOnlyRetainedConnections, + read_only_heap_delta_bytes: readOnlyHeapDeltaBytes, + lru_handle_cap: OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP, + lru_opened_handles: lruHandleCount, + lru_evicted_connections: lruEvictedCount, + lru_evicted_retained_connections: lruEvictedRetainedConnections, + lru_live_retained_connections: lruLiveRetainedConnections, + expectation, + verdict: verdict ? "PASS" : "FAIL", + })}\n`, + ); + if (!verdict) { + process.exitCode = 1; + } +} finally { + closeOpenClawAgentDatabases(); + closeOpenClawStateDatabase(); + fs.rmSync(dataRoot, { force: true, recursive: true }); +} diff --git a/src/config/sessions/session-accessor.sqlite-data-version.test.ts b/src/config/sessions/session-accessor.sqlite-data-version.test.ts index 8b9b9a756d42..870bcc15cfc5 100644 --- a/src/config/sessions/session-accessor.sqlite-data-version.test.ts +++ b/src/config/sessions/session-accessor.sqlite-data-version.test.ts @@ -2,6 +2,7 @@ import path from "node:path"; import { DatabaseSync } from "node:sqlite"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"; +import { clearNodeSqliteKyselyCacheForDatabase } from "../../infra/kysely-sync.js"; import { configureSqliteConnectionPragmas } from "../../infra/sqlite-wal.js"; import { closeOpenClawAgentDatabasesForTest, @@ -16,6 +17,7 @@ import { openSessionEntryReadView, upsertSessionEntry, } from "./session-accessor.js"; +import { readSqliteSessionEntryCache } from "./session-accessor.sqlite-entry-cache.js"; import { ensureTranscriptSessionRoot } from "./session-accessor.sqlite-transcript-state.js"; const parseSessionEntryCalls = vi.hoisted(() => vi.fn()); @@ -189,6 +191,47 @@ describe("SQLite session entry cache", () => { expect(second).toEqual(first); }); + it("keeps same-path caches isolated by live connection", async () => { + const scope = createSessionScope("connection-identity"); + await upsertSessionEntry(scope, { + label: "connection-identity", + sessionId: "connection-identity", + updatedAt: 1, + }); + const primary = openOpenClawAgentDatabase(scope); + const first = listSessionEntries({ ...scope, clone: false }); + const alternate = new DatabaseSync(primary.path, { readOnly: true }); + + try { + parseSessionEntryCalls.mockClear(); + const alternateSnapshot = readSqliteSessionEntryCache( + { agentId: primary.agentId, db: alternate }, + { cache: true }, + ); + expect(alternateSnapshot.entries.get(scope.sessionKey)?.label).toBe("connection-identity"); + const alternateEntry = alternateSnapshot.entries.get(scope.sessionKey); + expect(parseSessionEntryCalls).toHaveBeenCalledOnce(); + + parseSessionEntryCalls.mockClear(); + const second = listSessionEntries({ ...scope, clone: false }); + + expect(second[0]?.entry).toBe(first[0]?.entry); + expect(parseSessionEntryCalls).not.toHaveBeenCalled(); + + parseSessionEntryCalls.mockClear(); + const alternateAgain = readSqliteSessionEntryCache( + { agentId: primary.agentId, db: alternate }, + { cache: true }, + ); + + expect(alternateAgain.entries.get(scope.sessionKey)).toBe(alternateEntry); + expect(parseSessionEntryCalls).not.toHaveBeenCalled(); + } finally { + clearNodeSqliteKyselyCacheForDatabase(alternate); + alternate.close(); + } + }); + it("reuses parsed entries and list projections after a same-connection non-entry write", async () => { const scope = createSessionScope("same-connection-non-entry"); const siblingScope = { ...scope, sessionKey: "agent:main:same-connection-non-entry-2" }; diff --git a/src/config/sessions/session-accessor.sqlite-entry-cache.ts b/src/config/sessions/session-accessor.sqlite-entry-cache.ts index b813792d2d1f..13ee489c6398 100644 --- a/src/config/sessions/session-accessor.sqlite-entry-cache.ts +++ b/src/config/sessions/session-accessor.sqlite-entry-cache.ts @@ -8,7 +8,7 @@ import { getSessionKysely } from "./session-accessor.sqlite-scope.js"; import { parseSqliteSessionEntryJson } from "./session-accessor.sqlite-status.js"; import type { SessionEntry } from "./types.js"; -type SessionEntryCacheDatabase = Pick; +type SessionEntryCacheDatabase = Pick; export type SqliteSessionEntryCacheSnapshot = { entries: Map; @@ -17,7 +17,6 @@ export type SqliteSessionEntryCacheSnapshot = { }; type SqliteSessionEntryCache = SqliteSessionEntryCacheSnapshot & { - connection: DatabaseSync; listProjections: Map; updatedAtByKey: Map; validityToken: SqliteSessionEntryCacheValidityToken; @@ -36,9 +35,10 @@ type SqliteSessionEntryCacheValidityToken = { const MAX_INCREMENTAL_ENTRY_READ_KEYS = 500; // One parsed snapshot per opened agent database bounds memory to the process's database set. -// The connection-local validity token plus tracked-write invalidation keeps it current; -// without both, every read would re-query and re-parse every entry_json document. -const sessionEntryCaches = new Map(); +// Weak connection ownership lets closed read-only and evicted database handles release their +// snapshots. The connection-local validity token plus tracked-write invalidation keeps live +// snapshots current; without both, every read would re-query and re-parse every entry_json document. +const sessionEntryCaches = new WeakMap(); function readDataVersion(database: DatabaseSync): number { const row = database.prepare("PRAGMA data_version").get() as { data_version?: unknown }; @@ -154,7 +154,7 @@ function incrementallyRevalidateSessionEntrySnapshot( // already cheaper to reload than to preserve individual parsed identities. if (changedKeys.length > MAX_INCREMENTAL_ENTRY_READ_KEYS) { const loaded = loadSessionEntrySnapshot(database); - return { ...loaded, connection: database.db, validityToken }; + return { ...loaded, validityToken }; } const entries = new Map(cached.entries); @@ -184,7 +184,6 @@ function incrementallyRevalidateSessionEntrySnapshot( listEntries: createLazyListProjections(entries, listProjections), listProjections, updatedAtByKey, - connection: database.db, validityToken, }; } @@ -197,17 +196,11 @@ export function readSqliteSessionEntryCache( return loadSessionEntrySnapshot(database); } const validityToken = readCacheValidityToken(database.db); - const cached = sessionEntryCaches.get(database.path); - if ( - cached?.connection === database.db && - cacheValidityTokensEqual(cached.validityToken, validityToken) - ) { + const cached = sessionEntryCaches.get(database.db); + if (cached && cacheValidityTokensEqual(cached.validityToken, validityToken)) { return cached; } - if ( - cached?.connection === database.db && - cached.validityToken.dataVersion === validityToken.dataVersion - ) { + if (cached && cached.validityToken.dataVersion === validityToken.dataVersion) { // updated_at is entry-controlled, not a rowversion. Other connections can rewrite entry_json // without advancing it, so data_version changes must fully reload or same-ms rewrites go stale. // Accessor-owned writes on this connection hard-invalidate; unrelated local writes can safely diff. @@ -221,25 +214,22 @@ export function readSqliteSessionEntryCache( // publishing their mixed result could temporarily omit or retain the wrong keys. const reloadToken = readCacheValidityToken(database.db); const loaded = loadSessionEntrySnapshot(database); - const next = { ...loaded, connection: database.db, validityToken: reloadToken }; - sessionEntryCaches.set(database.path, next); + const next = { ...loaded, validityToken: reloadToken }; + sessionEntryCaches.set(database.db, next); return next; } - sessionEntryCaches.set(database.path, revalidated); + sessionEntryCaches.set(database.db, revalidated); return revalidated; } const loaded = loadSessionEntrySnapshot(database); - const next = { ...loaded, connection: database.db, validityToken }; - sessionEntryCaches.set(database.path, next); + const next = { ...loaded, validityToken }; + sessionEntryCaches.set(database.db, next); return next; } function invalidateTrackedCache(database: OpenClawAgentDatabase): void { const invalidate = () => { - const cached = sessionEntryCaches.get(database.path); - if (cached?.connection === database.db) { - sessionEntryCaches.delete(database.path); - } + sessionEntryCaches.delete(database.db); }; if (deferOpenClawAgentPostCommitPublication(database, invalidate)) { return;