mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-21 02:06:43 +00:00
perf(state): cap the per-agent SQLite handle cache with LRU eviction (#111411)
* perf(state): cap the per-agent SQLite handle cache with LRU eviction Multi-tenant hosts open one WAL database per agent (~3 file descriptors each); the process-local handle cache was unbounded, so large fleets exhausted descriptors. Cache hits now refresh LRU recency, and cache-miss opens evict the oldest non-transactional handle before constructing the new one, capped at 64 open handles. Eviction closes the process-local handle only; registry rows and durable data are untouched and evicted databases reopen transparently on next access. * fix(state): fleet-wide introspection reads stop opening writable agent databases Live testing an 81-agent gateway surfaced three full-fleet sweeps that opened every agent's SQLite database writable (schema ensure + registry write transaction per open): the 60s health snapshot's per-agent session summaries, usage-cost cache reads behind the gateway usage endpoints, and the zalouser doctor detector that ran from every CLI startup. Each sweep churned the new bounded handle cache and saturated the event loop. Session listing gains a readonly non-registering variant (SDK exposes it as an additive readOnly flag on listSessionEntries); health and usage reads use it, with transient-lock tolerance owned by the health caller. The zalouser detector gates on the channel/credentials actually existing, collects once per pass, and detection reads go readonly while migration apply stays writable. The usage refresh queue's flat 50ms busy retry becomes exponential backoff capped at 5s, and refreshing summaries retain their cache timestamp so the 30s TTL gates fleet rescans. * perf(state): make per-agent handle eviction and reopen cheap Eviction only works if reopen is cheap. Reopens after eviction previously repeated full first-open work; ensureOpenClawAgentSchema takes BEGIN IMMEDIATE and the registry upsert writes the shared state DB, so reopens blocked in synchronous busy waits (observed 14-22s event-loop stalls with reconcile workers holding write transactions). Owner/schema validation and registration now run once per path per process with invalidation on quarantine and disposal; the read-only integrity guard still runs on every physical open. Eviction closes with a PASSIVE WAL checkpoint instead of TRUNCATE, which waits on readers; orderly dispose keeps TRUNCATE so sidecars are flushed before unlink. * ci: regenerate plugin-SDK API baseline, drop prod test-seam exports, mock readonly accessor The additive readOnly flag on the SDK listSessionEntries changed the declaration surface, so the generated API baseline is regenerated via the sanctioned script. Knip forbids production exports consumed only by tests: the usage-cache testApi export is removed outright and the refresh-queue seam moves behind an env-gated global exposed by a test-support module. health.plugins.test's full session-accessor mock gains the newly imported readonly listing.
This commit is contained in:
Peter Steinberger
@@ -53,7 +53,7 @@ fa2df02bede6ed8843e5c2bd605c6ea5cd20313c305593fbd371dba6f1b931c3 module/command
|
||||
eb4c757fe0086c1dbfa4c3f3caf3dcff0d3cab3924c608237f08f740a6ee5f59 module/command-status
|
||||
235a9e4d983042c3db156efcab0e333984cbbf13ebdfcc44a3a5ab40cf3edb4f module/config-contracts
|
||||
20f3f8042de53e4eee61b64de9102c8c202b9299e6a29235647a4729f70145f2 module/config-mutation
|
||||
0c0bb80f81d0c769f343e3be95f01b3d734077c12e831367440952052aafbeb7 module/config-runtime
|
||||
6f23fd2d777f7189c38b34a51c6899abdad4ee4e6d74458eb3183c8ce03b8677 module/config-runtime
|
||||
c1ea9510dfda047609a99d5d2cd1f1560f5d469a36e6b695766213d695c25b0f module/conversation-runtime
|
||||
d9267aacc65aeebf0046d4eef691e5a510764063e6c7942b1d6bc7a282ca67a1 module/core
|
||||
4af19d59c2f18674e7d7f7dc1b358b644dc707e6bd601dc47168bd9e4a669940 module/dedupe-runtime
|
||||
@@ -116,7 +116,7 @@ b6b8edc50ecab8386c9acd8f374a207212b5a99c8f518538bbcf0c458dda3881 module/runtime
|
||||
17a6a199714ba8308e62928c0491bcf9fd214c923c3879aa99a047a30138253d module/secret-ref-runtime
|
||||
596a315d426121c9620b314e3a9a7f523840b46e007d94d0d5e83cdedf789d15 module/security-runtime
|
||||
31b785e74f1f8f56241b7756ef6a5d86199c5ce177cbb1c234a261866972f270 module/session-discussion
|
||||
f59099aa2d536246d4b1297f01bcea66796351a9259debdd45909afeb7a42d86 module/session-store-runtime
|
||||
9d7d884330397701c7de9f5b6800b970b2b349027491704184cb1bd6cda1fd00 module/session-store-runtime
|
||||
695971d31b3e16f0bf9b643acc30df312fe94b3c0dfb27a2e6156c8a00b5e261 module/setup
|
||||
9129c17df1523903f34beffd348ce177fcaa2983e155a8c854a50cf1bd7e99c4 module/setup-runtime
|
||||
cd431f6ba8327b81438b7a63b1963120f200f5abd145fb6aa7c5c561339cb0b1 module/setup-tools
|
||||
|
||||
@@ -149,10 +149,28 @@ describe("zalouser doctor state migration", () => {
|
||||
await expect(fs.access(`${filePath}.migrated`)).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
it("does not inspect agent session stores when zalouser has never been configured", async () => {
|
||||
const migration = findMigration("zalouser-direct-session-keys");
|
||||
const context = createDoctorContext(env);
|
||||
const config = { agents: { list: [{ id: "worker-1" }] } };
|
||||
|
||||
await expect(
|
||||
migration.detectLegacyState({ config, env, stateDir, oauthDir: stateDir, context }),
|
||||
).resolves.toBeNull();
|
||||
for (const agentId of ["main", "worker-1"]) {
|
||||
await expect(
|
||||
fs.access(path.join(stateDir, "agents", agentId, "agent", "openclaw-agent.sqlite")),
|
||||
).rejects.toThrow();
|
||||
}
|
||||
});
|
||||
|
||||
it("moves legacy group-shaped DM sessions to canonical direct keys", async () => {
|
||||
const legacyKey = "agent:main:zalouser:group:user-1";
|
||||
const canonicalKey = "agent:main:zalouser:direct:user-1";
|
||||
const config = { session: { store: storePath, dmScope: "per-channel-peer" as const } };
|
||||
const config = {
|
||||
channels: { zalouser: {} },
|
||||
session: { store: storePath, dmScope: "per-channel-peer" as const },
|
||||
};
|
||||
await upsertSessionEntry({
|
||||
agentId: "main",
|
||||
env,
|
||||
@@ -198,6 +216,7 @@ describe("zalouser doctor state migration", () => {
|
||||
const secondLegacyKey = "agent:main:zalouser:group:user-2";
|
||||
const canonicalKey = "agent:main:zalouser:direct:alice";
|
||||
const config = {
|
||||
channels: { zalouser: {} },
|
||||
session: {
|
||||
store: storePath,
|
||||
dmScope: "per-channel-peer" as const,
|
||||
|
||||
@@ -84,6 +84,7 @@ async function collectLegacyZalouserCredentialSources(
|
||||
function collectLegacyZalouserDmEntries(
|
||||
config: OpenClawConfig,
|
||||
env: NodeJS.ProcessEnv,
|
||||
options: { readOnly?: boolean } = {},
|
||||
): LegacyZalouserDmEntry[] {
|
||||
const entries = new Map<string, LegacyZalouserDmEntry>();
|
||||
const fallbackAccountId = config.channels?.zalouser?.defaultAccount?.trim() || "default";
|
||||
@@ -93,7 +94,11 @@ function collectLegacyZalouserDmEntries(
|
||||
]);
|
||||
for (const agentId of agentIds) {
|
||||
const storePath = resolveStorePath(config.session?.store, { agentId, env });
|
||||
const storedEntries = listSessionEntries({ agentId, storePath });
|
||||
const storedEntries = listSessionEntries({
|
||||
agentId,
|
||||
storePath,
|
||||
...(options.readOnly ? { readOnly: true } : {}),
|
||||
});
|
||||
const entryByKey = new Map(storedEntries.map(({ sessionKey, entry }) => [sessionKey, entry]));
|
||||
for (const { sessionKey, entry } of storedEntries) {
|
||||
const parsed = parseAgentSessionKey(sessionKey);
|
||||
@@ -228,10 +233,17 @@ export const stateMigrations: PluginDoctorStateMigration[] = [
|
||||
{
|
||||
id: "zalouser-direct-session-keys",
|
||||
label: "Zalo Personal direct-message sessions",
|
||||
detectLegacyState({ config, env }) {
|
||||
const count = collectLegacyZalouserDmEntries(config, env).flatMap(
|
||||
({ legacyKeys }) => legacyKeys,
|
||||
).length;
|
||||
async detectLegacyState({ config, env }) {
|
||||
// A never-configured channel cannot own legacy DMs, so do not scan every agent DB at startup.
|
||||
// Removed config defers leftover-row detection until zalouser is configured again.
|
||||
if (
|
||||
config.channels?.zalouser === undefined &&
|
||||
(await collectLegacyZalouserCredentialSources(env)).length === 0
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
const pending = collectLegacyZalouserDmEntries(config, env, { readOnly: true });
|
||||
const count = pending.flatMap(({ legacyKeys }) => legacyKeys).length;
|
||||
return count > 0
|
||||
? { preview: [`- Zalo Personal direct-message session keys: ${count} legacy row(s)`] }
|
||||
: null;
|
||||
|
||||
@@ -25,6 +25,7 @@ describe("getHealthSnapshot plugin state", () => {
|
||||
}));
|
||||
vi.doMock("../config/sessions/session-accessor.js", () => ({
|
||||
listSessionEntries: () => [],
|
||||
listSessionEntriesReadOnly: () => [],
|
||||
}));
|
||||
vi.doMock("../channels/plugins/read-only.js", () => ({
|
||||
listReadOnlyChannelPluginsForConfig: () => [],
|
||||
|
||||
@@ -73,7 +73,7 @@ async function loadFreshHealthModulesForTest() {
|
||||
loadSessionStore: () => testStore,
|
||||
}));
|
||||
vi.doMock("../config/sessions/session-accessor.js", () => ({
|
||||
listSessionEntries: (scope?: { agentId?: string; storePath?: string }) => {
|
||||
listSessionEntriesReadOnly: (scope?: { agentId?: string; storePath?: string }) => {
|
||||
listHealthSessionEntriesCalls.push(scope ?? {});
|
||||
return Object.entries(testStore).map(([sessionKey, entry]) => ({ sessionKey, entry }));
|
||||
},
|
||||
|
||||
+16
-5
@@ -366,11 +366,22 @@ const resolveAgentOrder = (cfg: OpenClawConfig) => {
|
||||
};
|
||||
|
||||
const buildSessionSummary = async (storePath: string, agentId?: string) => {
|
||||
const { listSessionEntries } = await import("../config/sessions/session-accessor.js");
|
||||
const sessions = listSessionEntries({
|
||||
...(agentId ? { agentId } : {}),
|
||||
storePath,
|
||||
})
|
||||
const { listSessionEntriesReadOnly } = await import("../config/sessions/session-accessor.js");
|
||||
const { isTransientSqliteError } = await import("../infra/unhandled-rejections.js");
|
||||
let listed: ReturnType<typeof listSessionEntriesReadOnly>;
|
||||
try {
|
||||
listed = listSessionEntriesReadOnly({
|
||||
...(agentId ? { agentId } : {}),
|
||||
storePath,
|
||||
});
|
||||
} catch (error) {
|
||||
if (!isTransientSqliteError(error)) {
|
||||
throw error;
|
||||
}
|
||||
// Health is best-effort: an empty snapshot beats failing on a transient lock.
|
||||
listed = [];
|
||||
}
|
||||
const sessions = listed
|
||||
.filter(({ sessionKey }) => sessionKey !== "global" && sessionKey !== "unknown")
|
||||
.map(({ sessionKey, entry }) => ({ key: sessionKey, updatedAt: entry?.updatedAt ?? 0 }))
|
||||
.toSorted((a, b) => b.updatedAt - a.updatedAt);
|
||||
|
||||
@@ -11,6 +11,7 @@ import { resolveStorePath } from "./paths.js";
|
||||
import { clearPluginOwnedSessionState } from "./plugin-host-cleanup.js";
|
||||
import {
|
||||
listSqliteSessionEntries,
|
||||
listSqliteSessionEntriesReadOnly,
|
||||
loadExactSqliteSessionEntry,
|
||||
loadSqliteSessionEntry,
|
||||
loadSqliteSessionEntryReadOnly,
|
||||
@@ -319,6 +320,16 @@ export function listSessionEntries(scope: SessionEntryListScope = {}): SessionEn
|
||||
return listSqliteSessionEntries(scope);
|
||||
}
|
||||
|
||||
/**
|
||||
* Health/status introspection must not join the writable lifecycle or register databases;
|
||||
* doing so churns fleet-wide agent handles on every health tick.
|
||||
*/
|
||||
export function listSessionEntriesReadOnly(
|
||||
scope: SessionEntryListScope = {},
|
||||
): SessionEntrySummary[] {
|
||||
return listSqliteSessionEntriesReadOnly(scope);
|
||||
}
|
||||
|
||||
/**
|
||||
* Borrowed keyed view over one resolved store for synchronous read-only hot paths.
|
||||
* Unlike loadSessionEntry, `get` is a raw exact persisted-key probe with no alias
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
import fs from "node:fs";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { cleanupTempDirs, makeTempDir } from "../../../test/helpers/temp-dir.js";
|
||||
import {
|
||||
closeOpenClawAgentDatabasesForTest,
|
||||
isOpenClawAgentDatabaseOpen,
|
||||
resolveOpenClawAgentSqlitePath,
|
||||
} from "../../state/openclaw-agent-db.js";
|
||||
import {
|
||||
closeOpenClawStateDatabaseForTest,
|
||||
openOpenClawStateDatabase,
|
||||
} from "../../state/openclaw-state-db.js";
|
||||
import {
|
||||
listSessionEntries,
|
||||
listSessionEntriesReadOnly,
|
||||
upsertSessionEntry,
|
||||
} from "./session-accessor.js";
|
||||
|
||||
const tempDirs: string[] = [];
|
||||
|
||||
function countRegisteredAgentDatabases(env: NodeJS.ProcessEnv): number {
|
||||
const row = openOpenClawStateDatabase({ env })
|
||||
.db.prepare("SELECT count(*) AS count FROM agent_databases")
|
||||
.get() as { count: number };
|
||||
return row.count;
|
||||
}
|
||||
|
||||
function clearRegisteredAgentDatabases(env: NodeJS.ProcessEnv): void {
|
||||
openOpenClawStateDatabase({ env }).db.prepare("DELETE FROM agent_databases").run();
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
closeOpenClawAgentDatabasesForTest();
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
cleanupTempDirs(tempDirs);
|
||||
});
|
||||
|
||||
describe("session accessor readonly listing", () => {
|
||||
it("returns the same entries as the writable listing for a populated agent database", async () => {
|
||||
const stateDir = makeTempDir(tempDirs, "openclaw-session-readonly-populated-");
|
||||
const env = { OPENCLAW_STATE_DIR: stateDir };
|
||||
const listScope = { agentId: "worker-1", env };
|
||||
|
||||
await upsertSessionEntry(
|
||||
{ ...listScope, sessionKey: "agent:worker-1:main" },
|
||||
{ sessionId: "session-1", updatedAt: 10 },
|
||||
);
|
||||
await upsertSessionEntry(
|
||||
{ ...listScope, sessionKey: "agent:worker-1:telegram:dm:42" },
|
||||
{ sessionId: "session-2", updatedAt: 20 },
|
||||
);
|
||||
const writableEntries = listSessionEntries(listScope);
|
||||
closeOpenClawAgentDatabasesForTest();
|
||||
|
||||
expect(listSessionEntriesReadOnly(listScope)).toEqual(writableEntries);
|
||||
});
|
||||
|
||||
it("returns an empty list without creating or registering a missing agent database", () => {
|
||||
const stateDir = makeTempDir(tempDirs, "openclaw-session-readonly-missing-");
|
||||
const env = { OPENCLAW_STATE_DIR: stateDir };
|
||||
const agentId = "worker-1";
|
||||
const databasePath = resolveOpenClawAgentSqlitePath({ agentId, env });
|
||||
clearRegisteredAgentDatabases(env);
|
||||
|
||||
expect(listSessionEntriesReadOnly({ agentId, env })).toEqual([]);
|
||||
expect(fs.existsSync(databasePath)).toBe(false);
|
||||
expect(countRegisteredAgentDatabases(env)).toBe(0);
|
||||
});
|
||||
|
||||
it("does not register a populated database during readonly health-style listing", async () => {
|
||||
const stateDir = makeTempDir(tempDirs, "openclaw-session-readonly-registry-");
|
||||
const env = { OPENCLAW_STATE_DIR: stateDir };
|
||||
const agentId = "worker-1";
|
||||
const scope = { agentId, env };
|
||||
|
||||
await upsertSessionEntry(
|
||||
{ ...scope, sessionKey: "agent:worker-1:main" },
|
||||
{ sessionId: "session-1", updatedAt: 10 },
|
||||
);
|
||||
const databasePath = resolveOpenClawAgentSqlitePath({ agentId, env });
|
||||
closeOpenClawAgentDatabasesForTest();
|
||||
clearRegisteredAgentDatabases(env);
|
||||
|
||||
expect(listSessionEntriesReadOnly(scope)).toHaveLength(1);
|
||||
expect(countRegisteredAgentDatabases(env)).toBe(0);
|
||||
expect(isOpenClawAgentDatabaseOpen(databasePath)).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -1,3 +1,4 @@
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
import type { MsgContext } from "../../auto-reply/templating.js";
|
||||
import {
|
||||
executeSqliteQuerySync,
|
||||
@@ -131,6 +132,26 @@ export function listSqliteSessionEntries(
|
||||
): SessionEntrySummary[] {
|
||||
const resolved = resolveSqliteScope({ ...scope, sessionKey: "" });
|
||||
const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved));
|
||||
return listSqliteSessionEntriesFromDatabase(database);
|
||||
}
|
||||
|
||||
/**
|
||||
* Lists session entries without opening the agent database writable.
|
||||
* Transient lock errors propagate: only the caller knows whether "empty" is an
|
||||
* acceptable degradation (health snapshots) or hides real state (migration detection).
|
||||
*/
|
||||
export function listSqliteSessionEntriesReadOnly(
|
||||
scope: Partial<Omit<SessionAccessScope, "sessionKey">> = {},
|
||||
): SessionEntrySummary[] {
|
||||
const resolved = resolveSqliteScope({ ...scope, sessionKey: "" });
|
||||
const result = withOpenClawAgentDatabaseReadOnly(
|
||||
(database) => listSqliteSessionEntriesFromDatabase(database),
|
||||
toDatabaseOptions(resolved),
|
||||
);
|
||||
return result.found ? result.value : [];
|
||||
}
|
||||
|
||||
function listSqliteSessionEntriesFromDatabase(database: { db: DatabaseSync }) {
|
||||
const db = getSessionKysely(database.db);
|
||||
const rows = executeSqliteQuerySync(
|
||||
database.db,
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
// Stable SQLite accessor surface. Domain owners live in the focused modules below.
|
||||
export {
|
||||
listSqliteSessionEntries,
|
||||
listSqliteSessionEntriesReadOnly,
|
||||
listSqliteSessionEntriesByStatus,
|
||||
listSqliteSessionTranscriptInstances,
|
||||
loadExactSqliteSessionEntry,
|
||||
|
||||
@@ -122,6 +122,7 @@ export type {
|
||||
export {
|
||||
clearPluginOwnedSessionState,
|
||||
listSessionEntries,
|
||||
listSessionEntriesReadOnly,
|
||||
loadExactSessionEntry,
|
||||
loadSessionEntry,
|
||||
loadSessionEntryReadOnly,
|
||||
|
||||
@@ -460,6 +460,34 @@ describe("gateway usage helpers", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("keeps refreshing cost summaries fresh for the TTL window", async () => {
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(new Date("2026-02-05T00:00:00.000Z"));
|
||||
vi.mocked(loadCostUsageSummaryFromCache).mockResolvedValueOnce({
|
||||
...costSummary({ totalTokens: 1, totalCost: 0 }),
|
||||
cacheStatus: {
|
||||
status: "refreshing",
|
||||
cachedFiles: 1,
|
||||
pendingFiles: 1,
|
||||
staleFiles: 1,
|
||||
},
|
||||
});
|
||||
|
||||
const config = {} as OpenClawConfig;
|
||||
await testApi.loadCostUsageSummaryCached({ startMs: 1, endMs: 2, config });
|
||||
|
||||
const entry = testApi.costUsageCache.get("agent:__default__:1-2:gateway");
|
||||
expect(entry?.updatedAt).toBe(Date.now());
|
||||
|
||||
await vi.advanceTimersByTimeAsync(29_999);
|
||||
await testApi.loadCostUsageSummaryCached({ startMs: 1, endMs: 2, config });
|
||||
expect(vi.mocked(loadCostUsageSummaryFromCache)).toHaveBeenCalledTimes(1);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await testApi.loadCostUsageSummaryCached({ startMs: 1, endMs: 2, config });
|
||||
expect(vi.mocked(loadCostUsageSummaryFromCache)).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("keeps cost usage cache entries scoped by agentId", async () => {
|
||||
const config = {} as OpenClawConfig;
|
||||
|
||||
|
||||
@@ -905,12 +905,7 @@ async function loadCostUsageSummaryCached(params: {
|
||||
const cacheKey = `${params.agentScope === "all" ? "all" : `agent:${params.agentId ?? "__default__"}`}:${params.startMs}-${params.endMs}:${dayBucketKey}`;
|
||||
const now = Date.now();
|
||||
const cached = costUsageCache.get(cacheKey);
|
||||
if (
|
||||
cached?.summary &&
|
||||
cached.updatedAt &&
|
||||
now - cached.updatedAt < COST_USAGE_CACHE_TTL_MS &&
|
||||
cached.summary.cacheStatus?.status !== "refreshing"
|
||||
) {
|
||||
if (cached?.summary && cached.updatedAt && now - cached.updatedAt < COST_USAGE_CACHE_TTL_MS) {
|
||||
return cached.summary;
|
||||
}
|
||||
|
||||
@@ -941,9 +936,11 @@ async function loadCostUsageSummaryCached(params: {
|
||||
})
|
||||
)
|
||||
.then((summary) => {
|
||||
// Refresh work is independent; retaining freshness prevents fleet rescans while it runs.
|
||||
// The short TTL still picks up a completed refresh promptly.
|
||||
setCostUsageCache(cacheKey, {
|
||||
summary,
|
||||
updatedAt: summary.cacheStatus?.status === "refreshing" ? undefined : Date.now(),
|
||||
updatedAt: Date.now(),
|
||||
});
|
||||
return summary;
|
||||
})
|
||||
|
||||
@@ -0,0 +1,82 @@
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { cleanupTempDirs, makeTempDir } from "../../test/helpers/temp-dir.js";
|
||||
import {
|
||||
closeOpenClawAgentDatabasesForTest,
|
||||
openOpenClawAgentDatabase,
|
||||
resolveOpenClawAgentSqlitePath,
|
||||
} from "../state/openclaw-agent-db.js";
|
||||
import {
|
||||
closeOpenClawStateDatabaseForTest,
|
||||
openOpenClawStateDatabase,
|
||||
} from "../state/openclaw-state-db.js";
|
||||
import { withEnv } from "../test-utils/env.js";
|
||||
import {
|
||||
isSessionCostUsageRefreshRunning,
|
||||
readSessionCostUsageRollupRows,
|
||||
writeSessionCostUsageRollup,
|
||||
} from "./session-cost-usage-cache.sqlite.js";
|
||||
|
||||
const tempDirs: string[] = [];
|
||||
|
||||
function countRegisteredAgentDatabases(): number {
|
||||
const row = openOpenClawStateDatabase()
|
||||
.db.prepare("SELECT count(*) AS count FROM agent_databases")
|
||||
.get() as {
|
||||
count: number;
|
||||
};
|
||||
return row.count;
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
closeOpenClawAgentDatabasesForTest();
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
cleanupTempDirs(tempDirs);
|
||||
});
|
||||
|
||||
describe("session cost usage SQLite cache", () => {
|
||||
it("returns empty values without creating a missing agent database", () => {
|
||||
const stateDir = makeTempDir(tempDirs, "openclaw-usage-cache-missing-");
|
||||
|
||||
withEnv({ OPENCLAW_STATE_DIR: stateDir }, () => {
|
||||
const databasePath = resolveOpenClawAgentSqlitePath({ agentId: "worker-1" });
|
||||
|
||||
expect(readSessionCostUsageRollupRows("worker-1", databasePath)).toEqual([]);
|
||||
expect(isSessionCostUsageRefreshRunning("worker-1", databasePath)).toBe(false);
|
||||
expect(fs.existsSync(databasePath)).toBe(false);
|
||||
expect(fs.existsSync(path.join(stateDir, "state", "openclaw.sqlite"))).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
it("does not register readonly cache reads while writes still register", () => {
|
||||
const stateDir = makeTempDir(tempDirs, "openclaw-usage-cache-registry-");
|
||||
|
||||
withEnv({ OPENCLAW_STATE_DIR: stateDir }, () => {
|
||||
const agentId = "worker-1";
|
||||
const database = openOpenClawAgentDatabase({ agentId });
|
||||
const databasePath = database.path;
|
||||
closeOpenClawAgentDatabasesForTest();
|
||||
|
||||
const stateDatabase = openOpenClawStateDatabase();
|
||||
stateDatabase.db.prepare("DELETE FROM agent_databases").run();
|
||||
expect(countRegisteredAgentDatabases()).toBe(0);
|
||||
|
||||
expect(readSessionCostUsageRollupRows(agentId, databasePath)).toEqual([]);
|
||||
expect(isSessionCostUsageRefreshRunning(agentId, databasePath)).toBe(false);
|
||||
expect(countRegisteredAgentDatabases()).toBe(0);
|
||||
|
||||
expect(
|
||||
writeSessionCostUsageRollup({
|
||||
agentId,
|
||||
databasePath,
|
||||
rollupId: "session.jsonl",
|
||||
previousValueJson: null,
|
||||
valueJson: "{}",
|
||||
updatedAt: 1,
|
||||
}),
|
||||
).toBe(true);
|
||||
expect(countRegisteredAgentDatabases()).toBe(1);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -1,11 +1,11 @@
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
import { normalizeAgentId } from "../routing/session-key.js";
|
||||
import { withOpenClawAgentDatabaseReadOnly } from "../state/openclaw-agent-db-readonly.js";
|
||||
import type { DB as OpenClawAgentKyselyDatabase } from "../state/openclaw-agent-db.generated.js";
|
||||
import {
|
||||
openOpenClawAgentDatabase,
|
||||
runOpenClawAgentWriteTransaction,
|
||||
} from "../state/openclaw-agent-db.js";
|
||||
import { runOpenClawAgentWriteTransaction } from "../state/openclaw-agent-db.js";
|
||||
// Per-agent SQLite storage for rebuildable per-session usage rollups.
|
||||
import { executeSqliteQuerySync, getNodeSqliteKysely } from "./kysely-sync.js";
|
||||
import { isTransientSqliteError } from "./unhandled-rejections.js";
|
||||
|
||||
const LEGACY_CACHE_SCOPE = "session-cost-usage";
|
||||
const LEGACY_CACHE_KEY = "cache";
|
||||
@@ -26,11 +26,24 @@ type SessionCostUsageRollupRow = {
|
||||
valueJson: string;
|
||||
};
|
||||
|
||||
function openCacheDatabase(agentId: string | undefined, databasePath?: string) {
|
||||
return openOpenClawAgentDatabase({
|
||||
agentId: normalizeAgentId(agentId),
|
||||
...(databasePath ? { path: databasePath } : {}),
|
||||
});
|
||||
function readCacheDatabase<T>(
|
||||
agentId: string | undefined,
|
||||
databasePath: string | undefined,
|
||||
operation: (database: { db: DatabaseSync }) => T,
|
||||
): T | undefined {
|
||||
try {
|
||||
const result = withOpenClawAgentDatabaseReadOnly(operation, {
|
||||
agentId: normalizeAgentId(agentId),
|
||||
...(databasePath ? { path: databasePath } : {}),
|
||||
});
|
||||
return result.found ? result.value : undefined;
|
||||
} catch (error) {
|
||||
if (!isTransientSqliteError(error)) {
|
||||
throw error;
|
||||
}
|
||||
// Usage rollups are rebuildable cache; stale or empty data beats failing the dashboard.
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
function readCacheValue(
|
||||
@@ -39,18 +52,21 @@ function readCacheValue(
|
||||
key: string,
|
||||
databasePath?: string,
|
||||
): string | null {
|
||||
const database = openCacheDatabase(agentId, databasePath);
|
||||
const kysely = getNodeSqliteKysely<AgentCacheDatabase>(database.db);
|
||||
const row = executeSqliteQuerySync(
|
||||
database.db,
|
||||
kysely
|
||||
.selectFrom("cache_entries")
|
||||
.select("value_json")
|
||||
.where("scope", "=", scope)
|
||||
.where("key", "=", key)
|
||||
.limit(1),
|
||||
).rows[0];
|
||||
return row?.value_json ?? null;
|
||||
return (
|
||||
readCacheDatabase(agentId, databasePath, (database) => {
|
||||
const kysely = getNodeSqliteKysely<AgentCacheDatabase>(database.db);
|
||||
const row = executeSqliteQuerySync(
|
||||
database.db,
|
||||
kysely
|
||||
.selectFrom("cache_entries")
|
||||
.select("value_json")
|
||||
.where("scope", "=", scope)
|
||||
.where("key", "=", key)
|
||||
.limit(1),
|
||||
).rows[0];
|
||||
return row?.value_json ?? null;
|
||||
}) ?? null
|
||||
);
|
||||
}
|
||||
|
||||
function deleteCacheValueIfUnchanged(params: {
|
||||
@@ -84,18 +100,21 @@ export function readSessionCostUsageRollupRows(
|
||||
agentId?: string,
|
||||
databasePath?: string,
|
||||
): SessionCostUsageRollupRow[] {
|
||||
const database = openCacheDatabase(agentId, databasePath);
|
||||
const kysely = getNodeSqliteKysely<AgentCacheDatabase>(database.db);
|
||||
return executeSqliteQuerySync(
|
||||
database.db,
|
||||
kysely
|
||||
.selectFrom("cache_entries")
|
||||
.select(["key", "value_json", "updated_at"])
|
||||
.where("scope", "=", ROLLUP_SCOPE),
|
||||
).rows.flatMap((row) =>
|
||||
row.value_json === null
|
||||
? []
|
||||
: [{ key: row.key, valueJson: row.value_json, updatedAt: row.updated_at }],
|
||||
return (
|
||||
readCacheDatabase(agentId, databasePath, (database) => {
|
||||
const kysely = getNodeSqliteKysely<AgentCacheDatabase>(database.db);
|
||||
return executeSqliteQuerySync(
|
||||
database.db,
|
||||
kysely
|
||||
.selectFrom("cache_entries")
|
||||
.select(["key", "value_json", "updated_at"])
|
||||
.where("scope", "=", ROLLUP_SCOPE),
|
||||
).rows.flatMap((row) =>
|
||||
row.value_json === null
|
||||
? []
|
||||
: [{ key: row.key, valueJson: row.value_json, updatedAt: row.updated_at }],
|
||||
);
|
||||
}) ?? []
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { testing as testApi } from "./session-cost-usage.test-support.js";
|
||||
|
||||
describe("session cost usage refresh backoff", () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers();
|
||||
testApi.clearUsageCostRefreshesForTest();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
testApi.clearUsageCostRefreshesForTest();
|
||||
vi.restoreAllMocks();
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("doubles consecutive busy delays, caps them, and resets after success", async () => {
|
||||
let calls = 0;
|
||||
const refresh = vi
|
||||
.spyOn(testApi.usageCostRefreshRuntime, "refreshCostUsageCacheForAgent")
|
||||
.mockImplementation(async () => {
|
||||
calls += 1;
|
||||
if (calls <= 10) {
|
||||
return "busy";
|
||||
}
|
||||
if (calls === 11) {
|
||||
testApi.requestCostUsageCacheRefresh({
|
||||
agentId: "backoff-test",
|
||||
sessionFiles: ["next-session.jsonl"],
|
||||
});
|
||||
return "refreshed";
|
||||
}
|
||||
if (calls === 12) {
|
||||
return "busy";
|
||||
}
|
||||
return "refreshed";
|
||||
});
|
||||
|
||||
testApi.requestCostUsageCacheRefresh({ agentId: "backoff-test" });
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(refresh).toHaveBeenCalledTimes(1);
|
||||
|
||||
for (const [delayMs, expectedCalls] of [
|
||||
[50, 2],
|
||||
[100, 3],
|
||||
[200, 4],
|
||||
[400, 5],
|
||||
[800, 6],
|
||||
[1_600, 7],
|
||||
[3_200, 8],
|
||||
[5_000, 9],
|
||||
[5_000, 10],
|
||||
] as const) {
|
||||
await vi.advanceTimersByTimeAsync(delayMs - 1);
|
||||
expect(refresh).toHaveBeenCalledTimes(expectedCalls - 1);
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(refresh).toHaveBeenCalledTimes(expectedCalls);
|
||||
}
|
||||
|
||||
await vi.advanceTimersByTimeAsync(4_999);
|
||||
expect(refresh).toHaveBeenCalledTimes(10);
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(refresh).toHaveBeenCalledTimes(12);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(49);
|
||||
expect(refresh).toHaveBeenCalledTimes(12);
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(refresh).toHaveBeenCalledTimes(13);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,32 @@
|
||||
import "./session-cost-usage.js";
|
||||
|
||||
type UsageCostRefreshParams = {
|
||||
agentId?: string;
|
||||
sessionFiles?: string[];
|
||||
};
|
||||
|
||||
type SessionCostUsageTestApi = {
|
||||
clearUsageCostRefreshesForTest(): void;
|
||||
requestCostUsageCacheRefresh(params?: UsageCostRefreshParams): void;
|
||||
usageCostRefreshRuntime: {
|
||||
refreshCostUsageCacheForAgent(params?: UsageCostRefreshParams): Promise<"busy" | "refreshed">;
|
||||
};
|
||||
};
|
||||
|
||||
function getTestApi(): SessionCostUsageTestApi {
|
||||
return (globalThis as Record<PropertyKey, unknown>)[
|
||||
Symbol.for("openclaw.sessionCostUsageTestApi")
|
||||
] as SessionCostUsageTestApi;
|
||||
}
|
||||
|
||||
export const testing: SessionCostUsageTestApi = {
|
||||
clearUsageCostRefreshesForTest() {
|
||||
getTestApi().clearUsageCostRefreshesForTest();
|
||||
},
|
||||
requestCostUsageCacheRefresh(params) {
|
||||
getTestApi().requestCostUsageCacheRefresh(params);
|
||||
},
|
||||
get usageCostRefreshRuntime() {
|
||||
return getTestApi().usageCostRefreshRuntime;
|
||||
},
|
||||
};
|
||||
@@ -112,6 +112,8 @@ const USAGE_COST_ROLLUP_VERSION = 2;
|
||||
const USAGE_COST_TRANSCRIPT_STAT_CONCURRENCY = 32;
|
||||
const USAGE_COST_FILE_ANCHOR_BYTES = 4096;
|
||||
const USAGE_COST_DIRECT_REFRESH_RETRY_MS = 25;
|
||||
const USAGE_COST_REFRESH_RETRY_MIN_MS = 50;
|
||||
const USAGE_COST_REFRESH_RETRY_MAX_MS = 5_000;
|
||||
const logger = createSubsystemLogger("usage-cost-cache");
|
||||
|
||||
type UsageCostRefreshState = {
|
||||
@@ -122,6 +124,7 @@ type UsageCostRefreshState = {
|
||||
pendingSessionFiles: Set<string>;
|
||||
running: boolean;
|
||||
sessionsDir: string;
|
||||
busyRetryDelayMs: number;
|
||||
timer?: ReturnType<typeof setTimeout>;
|
||||
};
|
||||
|
||||
@@ -1618,6 +1621,8 @@ async function refreshCostUsageCacheForAgent(params?: {
|
||||
}
|
||||
}
|
||||
|
||||
const usageCostRefreshRuntime = { refreshCostUsageCacheForAgent };
|
||||
|
||||
async function refreshCostUsageCache(params?: {
|
||||
config?: OpenClawConfig;
|
||||
agentId?: string;
|
||||
@@ -1768,6 +1773,7 @@ function requestCostUsageCacheRefresh(params?: {
|
||||
pendingSessionFiles: new Set(),
|
||||
running: false,
|
||||
sessionsDir: resolveSessionTranscriptsDirForAgent(params?.agentId),
|
||||
busyRetryDelayMs: USAGE_COST_REFRESH_RETRY_MIN_MS,
|
||||
};
|
||||
mergeUsageCostRefreshRequest(state, params);
|
||||
usageCostRefreshes.set(refreshKey, state);
|
||||
@@ -1827,7 +1833,7 @@ async function runQueuedUsageCostRefresh(
|
||||
state.pendingSessionFiles.clear();
|
||||
}
|
||||
state.fullRefreshRequested = false;
|
||||
const result = await refreshCostUsageCacheForAgent({
|
||||
const result = await usageCostRefreshRuntime.refreshCostUsageCacheForAgent({
|
||||
config: state.config,
|
||||
agentId: state.agentId,
|
||||
databasePath: state.databasePath,
|
||||
@@ -1842,9 +1848,15 @@ async function runQueuedUsageCostRefresh(
|
||||
state.pendingSessionFiles.add(sessionFile);
|
||||
}
|
||||
}
|
||||
retryDelayMs = 50;
|
||||
retryDelayMs = state.busyRetryDelayMs;
|
||||
// Contention among many per-agent refreshes must degrade to polling, not a 20Hz spin.
|
||||
state.busyRetryDelayMs = Math.min(
|
||||
state.busyRetryDelayMs * 2,
|
||||
USAGE_COST_REFRESH_RETRY_MAX_MS,
|
||||
);
|
||||
break;
|
||||
}
|
||||
state.busyRetryDelayMs = USAGE_COST_REFRESH_RETRY_MIN_MS;
|
||||
}
|
||||
} catch (error) {
|
||||
logger.warn(`background refresh failed: ${formatErrorMessage(error)}`, { error });
|
||||
@@ -1858,6 +1870,23 @@ async function runQueuedUsageCostRefresh(
|
||||
}
|
||||
}
|
||||
|
||||
function clearUsageCostRefreshesForTest(): void {
|
||||
for (const state of usageCostRefreshes.values()) {
|
||||
if (state.timer) {
|
||||
clearTimeout(state.timer);
|
||||
}
|
||||
}
|
||||
usageCostRefreshes.clear();
|
||||
}
|
||||
|
||||
if (process.env.VITEST || process.env.NODE_ENV === "test") {
|
||||
(globalThis as Record<PropertyKey, unknown>)[Symbol.for("openclaw.sessionCostUsageTestApi")] = {
|
||||
requestCostUsageCacheRefresh,
|
||||
usageCostRefreshRuntime,
|
||||
clearUsageCostRefreshesForTest,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Scan all transcript files to discover sessions not in the session store.
|
||||
* Returns basic metadata for each discovered session.
|
||||
|
||||
@@ -34,7 +34,7 @@ type MountEntry = { mountPoint: string; fsType: string; source?: string };
|
||||
|
||||
export type SqliteWalMaintenance = {
|
||||
checkpoint: () => boolean;
|
||||
close: () => boolean;
|
||||
close: (options?: { checkpointMode?: SqliteWalCheckpointMode }) => boolean;
|
||||
};
|
||||
|
||||
/** Options controlling WAL autocheckpoint and periodic checkpoint behavior. */
|
||||
@@ -508,12 +508,15 @@ export function configureSqliteWalMaintenance(
|
||||
|
||||
return {
|
||||
checkpoint,
|
||||
close: () => {
|
||||
close: (closeOptions) => {
|
||||
if (timer) {
|
||||
clearInterval(timer);
|
||||
timer = null;
|
||||
}
|
||||
return checkpoint();
|
||||
// Cache eviction passes PASSIVE: a TRUNCATE close-checkpoint waits on
|
||||
// readers and has starved the event loop for seconds under fleet churn.
|
||||
// Orderly dispose/delete keeps TRUNCATE so sidecars are flushed for unlink.
|
||||
return runCheckpoint(closeOptions?.checkpointMode ?? checkpointMode);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -16,6 +16,7 @@ import {
|
||||
deleteSessionEntryLifecycle as deleteAccessorSessionEntryLifecycle,
|
||||
loadTranscriptEventsSync as loadAccessorTranscriptEventsSync,
|
||||
listSessionEntries as listAccessorSessionEntries,
|
||||
listSessionEntriesReadOnly as listAccessorSessionEntriesReadOnly,
|
||||
loadSessionEntry,
|
||||
patchSessionEntry as patchAccessorSessionEntry,
|
||||
readSessionUpdatedAt as readAccessorSessionUpdatedAt,
|
||||
@@ -360,11 +361,17 @@ export function getSessionEntry(params: SessionStoreReadParams): SessionEntry |
|
||||
return entry ? projectPluginSessionEntry(entry) : undefined;
|
||||
}
|
||||
|
||||
/** Lists session entries for one agent. */
|
||||
/**
|
||||
* Lists session entries for one agent. `readOnly` reads without joining the
|
||||
* agent database writable lifecycle (no create/register/migrate) — required
|
||||
* for detection/introspection paths that may run across the whole fleet.
|
||||
* One flagged entry instead of a second export keeps the SDK surface budget flat.
|
||||
*/
|
||||
export function listSessionEntries(
|
||||
params: SessionStoreListParams = {},
|
||||
params: SessionStoreListParams & { readOnly?: boolean } = {},
|
||||
): SessionStoreEntrySummary[] {
|
||||
return listAccessorSessionEntries({
|
||||
const list = params.readOnly ? listAccessorSessionEntriesReadOnly : listAccessorSessionEntries;
|
||||
return list({
|
||||
...(params.agentId !== undefined ? { agentId: params.agentId } : {}),
|
||||
...(params.env !== undefined ? { env: params.env } : {}),
|
||||
...(params.hydrateSkillPromptRefs !== undefined
|
||||
|
||||
@@ -0,0 +1,180 @@
|
||||
// Agent database cache tests cover bounded process-local SQLite handle ownership.
|
||||
import fs from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
closeOpenClawAgentDatabasesForTest,
|
||||
disposeOpenClawAgentDatabaseByPath,
|
||||
isOpenClawAgentDatabaseOpen,
|
||||
listOpenClawRegisteredAgentDatabases,
|
||||
OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP,
|
||||
openOpenClawAgentDatabase,
|
||||
} from "./openclaw-agent-db.js";
|
||||
import { closeOpenClawStateDatabaseForTest } from "./openclaw-state-db.js";
|
||||
|
||||
const tempStateDirs: string[] = [];
|
||||
|
||||
function createTempStateDir(): string {
|
||||
const stateDir = fs.realpathSync(
|
||||
fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-agent-db-cache-")),
|
||||
);
|
||||
tempStateDirs.push(stateDir);
|
||||
return stateDir;
|
||||
}
|
||||
|
||||
function fillAgentDatabaseCache(env: NodeJS.ProcessEnv, prefix: string): void {
|
||||
for (let index = 0; index < OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP; index += 1) {
|
||||
openOpenClawAgentDatabase({ agentId: `${prefix}-${index}`, env });
|
||||
}
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
closeOpenClawAgentDatabasesForTest();
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
for (const stateDir of tempStateDirs.splice(0)) {
|
||||
fs.rmSync(stateDir, { force: true, recursive: true });
|
||||
}
|
||||
});
|
||||
|
||||
describe("openclaw agent database handle cache", () => {
|
||||
it("keeps only the capped number of open handles", () => {
|
||||
const env = { OPENCLAW_STATE_DIR: createTempStateDir() };
|
||||
const databases = Array.from({ length: OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP + 1 }, (_, index) =>
|
||||
openOpenClawAgentDatabase({ agentId: `worker-${index}`, env }),
|
||||
);
|
||||
const leastRecentlyUsed = databases[0]!;
|
||||
|
||||
expect(databases.filter((database) => database.db.isOpen)).toHaveLength(
|
||||
OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP,
|
||||
);
|
||||
expect(isOpenClawAgentDatabaseOpen(leastRecentlyUsed.path)).toBe(false);
|
||||
expect(leastRecentlyUsed.db.isOpen).toBe(false);
|
||||
});
|
||||
|
||||
it("refreshes cache-hit recency before evicting the true LRU handle", () => {
|
||||
const env = { OPENCLAW_STATE_DIR: createTempStateDir() };
|
||||
const recentlyUsed = openOpenClawAgentDatabase({ agentId: "recently-used", env });
|
||||
const untouched = Array.from({ length: OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP - 1 }, (_, index) =>
|
||||
openOpenClawAgentDatabase({ agentId: `untouched-${index}`, env }),
|
||||
);
|
||||
const leastRecentlyUsed = untouched[0]!;
|
||||
|
||||
expect(openOpenClawAgentDatabase({ agentId: "recently-used", env })).toBe(recentlyUsed);
|
||||
openOpenClawAgentDatabase({ agentId: "newest", env });
|
||||
|
||||
expect(recentlyUsed.db.isOpen).toBe(true);
|
||||
expect(isOpenClawAgentDatabaseOpen(recentlyUsed.path)).toBe(true);
|
||||
expect(leastRecentlyUsed.db.isOpen).toBe(false);
|
||||
expect(isOpenClawAgentDatabaseOpen(leastRecentlyUsed.path)).toBe(false);
|
||||
});
|
||||
|
||||
it("never evicts an LRU handle with an open transaction", () => {
|
||||
const env = { OPENCLAW_STATE_DIR: createTempStateDir() };
|
||||
const transactionOwner = openOpenClawAgentDatabase({ agentId: "transaction-owner", env });
|
||||
transactionOwner.db.exec("BEGIN IMMEDIATE");
|
||||
try {
|
||||
const untouched = Array.from({ length: OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP - 1 }, (_, index) =>
|
||||
openOpenClawAgentDatabase({ agentId: `untouched-${index}`, env }),
|
||||
);
|
||||
const leastRecentlyUsed = untouched[0]!;
|
||||
openOpenClawAgentDatabase({ agentId: "newest", env });
|
||||
|
||||
expect(transactionOwner.db.isOpen).toBe(true);
|
||||
expect(transactionOwner.db.isTransaction).toBe(true);
|
||||
expect(isOpenClawAgentDatabaseOpen(transactionOwner.path)).toBe(true);
|
||||
expect(leastRecentlyUsed.db.isOpen).toBe(false);
|
||||
expect(isOpenClawAgentDatabaseOpen(leastRecentlyUsed.path)).toBe(false);
|
||||
} finally {
|
||||
transactionOwner.db.exec("ROLLBACK");
|
||||
}
|
||||
});
|
||||
|
||||
it("reopens an evicted database without losing durable rows", () => {
|
||||
const env = { OPENCLAW_STATE_DIR: createTempStateDir() };
|
||||
const evicted = openOpenClawAgentDatabase({ agentId: "evicted", env });
|
||||
evicted.db
|
||||
.prepare(
|
||||
"INSERT INTO auth_profile_state (state_key, state_json, updated_at) VALUES (?, ?, ?)",
|
||||
)
|
||||
.run("cache-eviction", JSON.stringify({ preserved: true }), 42);
|
||||
|
||||
fillAgentDatabaseCache(env, "filler");
|
||||
expect(evicted.db.isOpen).toBe(false);
|
||||
|
||||
const reopened = openOpenClawAgentDatabase({ agentId: "evicted", env });
|
||||
expect(reopened).not.toBe(evicted);
|
||||
expect(
|
||||
reopened.db
|
||||
.prepare("SELECT state_json, updated_at FROM auth_profile_state WHERE state_key = ?")
|
||||
.get("cache-eviction"),
|
||||
).toEqual({ state_json: JSON.stringify({ preserved: true }), updated_at: 42 });
|
||||
});
|
||||
|
||||
it("registers a first open without refreshing registry metadata after eviction", () => {
|
||||
const env = { OPENCLAW_STATE_DIR: createTempStateDir() };
|
||||
const nowSpy = vi.spyOn(Date, "now").mockReturnValue(1_000);
|
||||
try {
|
||||
const evicted = openOpenClawAgentDatabase({ agentId: "evicted", env });
|
||||
expect(
|
||||
listOpenClawRegisteredAgentDatabases({ env }).find(
|
||||
(entry) => entry.agentId === "evicted" && entry.path === evicted.path,
|
||||
),
|
||||
).toMatchObject({ lastSeenAt: 1_000 });
|
||||
|
||||
nowSpy.mockReturnValue(2_000);
|
||||
fillAgentDatabaseCache(env, "registry-filler");
|
||||
expect(evicted.db.isOpen).toBe(false);
|
||||
|
||||
openOpenClawAgentDatabase({ agentId: "evicted", env });
|
||||
expect(
|
||||
listOpenClawRegisteredAgentDatabases({ env }).find(
|
||||
(entry) => entry.agentId === "evicted" && entry.path === evicted.path,
|
||||
),
|
||||
).toMatchObject({ lastSeenAt: 1_000 });
|
||||
} finally {
|
||||
nowSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("validates ownership when an evicted path is requested for another agent", () => {
|
||||
const env = { OPENCLAW_STATE_DIR: createTempStateDir() };
|
||||
const evicted = openOpenClawAgentDatabase({ agentId: "worker-a", env });
|
||||
fillAgentDatabaseCache(env, "ownership-filler");
|
||||
expect(evicted.db.isOpen).toBe(false);
|
||||
|
||||
expect(() =>
|
||||
openOpenClawAgentDatabase({ agentId: "worker-b", env, path: evicted.path }),
|
||||
).toThrow(/belongs to agent worker-a/);
|
||||
});
|
||||
|
||||
it("revalidates and registers a database after explicit disposal", () => {
|
||||
const env = { OPENCLAW_STATE_DIR: createTempStateDir() };
|
||||
const nowSpy = vi.spyOn(Date, "now").mockReturnValue(1_000);
|
||||
try {
|
||||
const disposed = openOpenClawAgentDatabase({ agentId: "disposed", env });
|
||||
expect(
|
||||
listOpenClawRegisteredAgentDatabases({ env }).find(
|
||||
(entry) => entry.agentId === "disposed" && entry.path === disposed.path,
|
||||
),
|
||||
).toMatchObject({ lastSeenAt: 1_000 });
|
||||
|
||||
expect(disposeOpenClawAgentDatabaseByPath(disposed.path, { env })).toBe(true);
|
||||
expect(
|
||||
listOpenClawRegisteredAgentDatabases({ env }).some(
|
||||
(entry) => entry.agentId === "disposed" && entry.path === disposed.path,
|
||||
),
|
||||
).toBe(false);
|
||||
|
||||
nowSpy.mockReturnValue(2_000);
|
||||
openOpenClawAgentDatabase({ agentId: "disposed", env, path: disposed.path });
|
||||
expect(
|
||||
listOpenClawRegisteredAgentDatabases({ env }).find(
|
||||
(entry) => entry.agentId === "disposed" && entry.path === disposed.path,
|
||||
),
|
||||
).toMatchObject({ lastSeenAt: 2_000 });
|
||||
} finally {
|
||||
nowSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -70,14 +70,22 @@ export { resolveOpenClawAgentSqlitePath } from "./openclaw-agent-db.paths.js";
|
||||
* OpenClaw state database for discovery and maintenance.
|
||||
*/
|
||||
const OPENCLAW_AGENT_DB_SLOW_OPEN_MS = 1_000;
|
||||
// Each WAL database consumes roughly three file descriptors, so the fixed cap
|
||||
// satisfies the bounded-cache policy within a predictable FD budget, without config.
|
||||
export const OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP = 64;
|
||||
const agentDbLog = createSubsystemLogger("state/agent-db");
|
||||
const cachedDatabases = new Map<string, OpenClawAgentDatabase>();
|
||||
// External schema changes under a live process are unsupported: doctor migrations
|
||||
// require restart, so successful owner/schema validation is process-stable.
|
||||
const validatedAgentDatabasePaths = new Map<string, string>();
|
||||
const terminalOpenLatch = createSqliteTerminalOpenLatch({
|
||||
closeByPath: closeOpenClawAgentDatabaseByPath,
|
||||
});
|
||||
|
||||
/** Latch background verification damage so later opens fail without rescanning. */
|
||||
export function recordOpenClawAgentDatabaseOpenFailure(pathname: string, error: Error): void {
|
||||
// Quarantine revokes this process's trust because doctor may replace the file.
|
||||
validatedAgentDatabasePaths.delete(path.resolve(pathname));
|
||||
terminalOpenLatch.record(pathname, error);
|
||||
}
|
||||
|
||||
@@ -153,6 +161,8 @@ export function openOpenClawAgentDatabase(
|
||||
`OpenClaw agent database ${pathname} is already open for agent ${cached.agentId}; requested agent ${agentId}.`,
|
||||
);
|
||||
}
|
||||
cachedDatabases.delete(pathname);
|
||||
cachedDatabases.set(pathname, cached);
|
||||
return cached;
|
||||
}
|
||||
// Latched paths are quarantined; every fresh open fails fast here until
|
||||
@@ -187,14 +197,24 @@ export function openOpenClawAgentDatabase(
|
||||
}
|
||||
const openStartedAt = Date.now();
|
||||
ensureOpenClawAgentDatabasePermissions(pathname, databaseOptions);
|
||||
// Free a slot before constructing the new handle: under real descriptor
|
||||
// pressure the 65th open would otherwise fail before eviction could run.
|
||||
evictLruAgentDatabaseHandles();
|
||||
const sqlite = requireNodeSqlite();
|
||||
const db = new sqlite.DatabaseSync(pathname);
|
||||
// Eviction churn must avoid schema/registry busy waits on the event loop while
|
||||
// reconcile workers hold write transactions on these same agent databases.
|
||||
const isValidatedReopen = validatedAgentDatabasePaths.get(pathname) === agentId;
|
||||
const walMaintenance = (() => {
|
||||
let maintenance: OpenClawAgentDatabase["walMaintenance"] | undefined;
|
||||
try {
|
||||
db.exec(`PRAGMA busy_timeout = ${OPENCLAW_SQLITE_BUSY_TIMEOUT_MS};`);
|
||||
assertSupportedAgentSchemaVersion(db, pathname);
|
||||
assertExistingAgentSchemaOwner(readExistingAgentSchemaMeta(db), agentId, pathname);
|
||||
if (!isValidatedReopen) {
|
||||
assertSupportedAgentSchemaVersion(db, pathname);
|
||||
assertExistingAgentSchemaOwner(readExistingAgentSchemaMeta(db), agentId, pathname);
|
||||
}
|
||||
// Integrity is not process-stable: the file can be damaged while evicted.
|
||||
// This guard is read-only (no busy waits), so every physical open pays it.
|
||||
assertAgentDatabaseIntegrityBeforeMutation(db, pathname);
|
||||
configureSqlitePreSchemaPragmas(db, {
|
||||
busyTimeoutMs: OPENCLAW_SQLITE_BUSY_TIMEOUT_MS,
|
||||
@@ -206,7 +226,9 @@ export function openOpenClawAgentDatabase(
|
||||
foreignKeys: true,
|
||||
synchronous: "NORMAL",
|
||||
});
|
||||
ensureOpenClawAgentSchema(db, agentId, pathname);
|
||||
if (!isValidatedReopen) {
|
||||
ensureOpenClawAgentSchema(db, agentId, pathname);
|
||||
}
|
||||
return maintenance;
|
||||
} catch (err) {
|
||||
maintenance?.close();
|
||||
@@ -222,11 +244,14 @@ export function openOpenClawAgentDatabase(
|
||||
})();
|
||||
ensureOpenClawAgentDatabasePermissions(pathname, databaseOptions);
|
||||
const database = { agentId, db, path: pathname, walMaintenance };
|
||||
try {
|
||||
registerOpenClawAgentDatabase({ agentId, path: pathname, env: options.env });
|
||||
} catch (error) {
|
||||
closeCachedOpenClawAgentDatabase(database);
|
||||
throw error;
|
||||
if (!isValidatedReopen) {
|
||||
try {
|
||||
registerOpenClawAgentDatabase({ agentId, path: pathname, env: options.env });
|
||||
} catch (error) {
|
||||
closeCachedOpenClawAgentDatabase(database);
|
||||
throw error;
|
||||
}
|
||||
validatedAgentDatabasePaths.set(pathname, agentId);
|
||||
}
|
||||
cachedDatabases.set(pathname, database);
|
||||
terminalOpenLatch.clear(pathname);
|
||||
@@ -312,14 +337,59 @@ export function runOpenClawAgentWriteTransaction<T>(
|
||||
|
||||
let unregisterExitClose: (() => void) | null = null;
|
||||
|
||||
function closeCachedOpenClawAgentDatabase(database: OpenClawAgentDatabase): void {
|
||||
database.walMaintenance.close();
|
||||
function closeCachedOpenClawAgentDatabase(
|
||||
database: OpenClawAgentDatabase,
|
||||
options: { eviction?: boolean } = {},
|
||||
): void {
|
||||
// Eviction must stay cheap: PASSIVE skips waiting on concurrent readers,
|
||||
// whose drained TRUNCATE checkpoints blocked the event loop for seconds.
|
||||
database.walMaintenance.close(options.eviction ? { checkpointMode: "PASSIVE" } : undefined);
|
||||
clearNodeSqliteKyselyCacheForDatabase(database.db);
|
||||
if (database.db.isOpen) {
|
||||
database.db.close();
|
||||
}
|
||||
}
|
||||
|
||||
function evictLruAgentDatabaseHandles(): void {
|
||||
// Callers re-fetch handles from this cache at each operation entry and use
|
||||
// them within one synchronous section, so eviction can never close a handle
|
||||
// mid-use; a handle retained across an eviction-triggering open goes stale.
|
||||
while (cachedDatabases.size >= OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP) {
|
||||
let evicted = false;
|
||||
for (const [pathname, database] of cachedDatabases) {
|
||||
// A synchronous transaction owns its handle through COMMIT or ROLLBACK;
|
||||
// closing it here would violate the transaction commit-section contract.
|
||||
if (database.db.isTransaction) {
|
||||
continue;
|
||||
}
|
||||
// Registry rows are durable discovery metadata; only explicit disposal
|
||||
// unregisters them, while eviction closes this process-local handle.
|
||||
closeCachedOpenClawAgentDatabase(database, { eviction: true });
|
||||
cachedDatabases.delete(pathname);
|
||||
agentDbLog.debug("evicted OpenClaw agent database handle", {
|
||||
agentId: database.agentId,
|
||||
openHandles: cachedDatabases.size,
|
||||
path: pathname,
|
||||
});
|
||||
evicted = true;
|
||||
break;
|
||||
}
|
||||
if (!evicted) {
|
||||
// Every handle is mid-transaction: sync commit sections bound concurrent
|
||||
// transactions at call-nesting depth, so this stays a pathological safety
|
||||
// valve; exceeding the cap beats failing an unrelated agent's open.
|
||||
agentDbLog.warn(
|
||||
"agent database handle cap exceeded; all cached handles are in transactions",
|
||||
{
|
||||
cap: OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP,
|
||||
openHandles: cachedDatabases.size,
|
||||
},
|
||||
);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Return whether the exact cached agent database pathname is still open. */
|
||||
export function isOpenClawAgentDatabaseOpen(pathname: string): boolean {
|
||||
const database = cachedDatabases.get(path.resolve(pathname));
|
||||
@@ -352,6 +422,8 @@ export function disposeOpenClawAgentDatabaseByPath(
|
||||
// Require the cache's exact lexical owner. Following a symlink or accepting
|
||||
// an uncached path could unregister a database another process now owns.
|
||||
const resolvedPath = path.resolve(pathname);
|
||||
// Disposal can be followed by file deletion or recreation, so revalidate next open.
|
||||
validatedAgentDatabasePaths.delete(resolvedPath);
|
||||
const database = cachedDatabases.get(resolvedPath);
|
||||
if (!database || database.path !== resolvedPath) {
|
||||
return false;
|
||||
@@ -383,5 +455,6 @@ export function closeOpenClawAgentDatabases(): void {
|
||||
/** Close cached agent handles and clear terminal failure latches for test isolation. */
|
||||
export function closeOpenClawAgentDatabasesForTest(): void {
|
||||
closeOpenClawAgentDatabases();
|
||||
validatedAgentDatabasePaths.clear();
|
||||
terminalOpenLatch.clearAll();
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user