diff --git a/dsh.ts b/dsh.ts index 3b0b8fa..31d5778 100644 --- a/dsh.ts +++ b/dsh.ts @@ -369,6 +369,13 @@ export function apply(ctx: DshContext, input: Config = {}): void { assembly.contexts.push({ name: "graph-memory:recall", text }); } } catch (error) { + // A failed recall must not stay pinned in recallCache: the next + // system-prompt/assemble would re-await the same rejected Promise and + // log "[graph-memory] DSH recall failed" on every round until the next + // agent/inbox/claimed clears it. Evict the entry so the next assembly + // performs a fresh recall attempt. + const current = recallCache.get(key); + if (current && current.query === query) recallCache.delete(key); ctx.logger.warn(`[graph-memory] DSH recall failed: ${String(error)}`); } return next(); diff --git a/src/store/db.ts b/src/store/db.ts index 9537039..ab281cb 100755 --- a/src/store/db.ts +++ b/src/store/db.ts @@ -46,6 +46,12 @@ export function openDb(dbPath: string): DatabaseSyncInstance { const db = new DatabaseSync(resolved); db.exec("PRAGMA journal_mode = WAL"); db.exec("PRAGMA foreign_keys = ON"); + // Multiple plugin instances (DSH fibers/profiles, tests) may open the same + // database file. Without a busy timeout, a write that collides with another + // connection's transaction fails immediately with SQLITE_BUSY ("database is + // locked") and the error surfaces as a spurious recall/extraction failure. + // Wait up to 5s for the lock instead so transient contention is retried. + db.exec("PRAGMA busy_timeout = 5000"); migrate(db); return db; } diff --git a/test/db.test.ts b/test/db.test.ts new file mode 100644 index 0000000..afd141a --- /dev/null +++ b/test/db.test.ts @@ -0,0 +1,88 @@ +/** + * graph-memory — SQLite 连接回归测试 + * + * By: adoresever + * Email: Wywelljob@gmail.com + */ + +import { describe, expect, it } from "vitest"; +import { spawn } from "node:child_process"; +import { mkdtempSync, writeFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { openDb } from "../src/store/db.ts"; + +describe("openDb", () => { + it("configures a busy timeout on every connection", () => { + const dir = mkdtempSync(join(tmpdir(), "gm-db-")); + const db = openDb(join(dir, "graph-memory.db")); + try { + const row = db.prepare("PRAGMA busy_timeout").get() as { timeout: number }; + expect(Number(row.timeout)).toBe(5000); + } finally { + db.close(); + } + }); + + it("waits for a write lock held by another process instead of throwing database is locked", async () => { + const dir = mkdtempSync(join(tmpdir(), "gm-db-")); + const dbPath = join(dir, "graph-memory.db"); + const db = openDb(dbPath); + + // Child process mirrors a second host connection (e.g. another DSH + // process sharing the same database file): it takes the write lock, holds + // it briefly, then commits. Its event loop is its own, so it can release + // the lock while this process' busy handler is waiting. The script lives + // in the project root so require("@photostructure/sqlite") resolves. + const childScript = ` + const { DatabaseSync } = require("@photostructure/sqlite"); + const file = process.argv[2]; + const db = new DatabaseSync(file); + db.exec("PRAGMA journal_mode = WAL"); + db.exec("PRAGMA busy_timeout = 5000"); + db.exec("BEGIN IMMEDIATE"); + console.log("LOCKED"); + setTimeout(() => { + try { db.exec("COMMIT"); } catch (e) { console.error("child commit failed:", String(e)); process.exit(1); } + db.close(); + }, 500); + `; + const scriptPath = join(process.cwd(), ".gm-lock-holder.cjs"); + writeFileSync(scriptPath, childScript); + + try { + const child = spawn(process.execPath, [scriptPath, dbPath], { stdio: ["ignore", "pipe", "pipe"] }); + const locked = new Promise((resolve, reject) => { + let settled = false; + child.stdout.on("data", (chunk: Buffer) => { + if (chunk.toString().includes("LOCKED") && !settled) { + settled = true; + resolve(); + } + }); + child.on("error", reject); + child.on("exit", (code) => { + if (!settled && code !== 0) reject(new Error(`child exited early with code ${code}`)); + }); + }); + + await locked; + + // This write collides with the child's open transaction. Without a busy + // timeout it fails instantly with SQLITE_BUSY; with one it blocks until + // the child commits (~500ms) and then succeeds. + const t0 = Date.now(); + db.prepare(` + INSERT INTO gm_nodes (id, type, name, description, content, status, validated_count, source_sessions, created_at, updated_at) + VALUES (?, 'SKILL', ?, ?, ?, 'active', 1, '[]', ?, ?) + `).run("n-busy-timeout", "busy-timeout-node", "desc", "content", Date.now(), Date.now()); + const elapsed = Date.now() - t0; + + expect(elapsed).toBeGreaterThanOrEqual(300); + } finally { + db.close(); + rmSync(scriptPath, { force: true }); + rmSync(dir, { recursive: true, force: true }); + } + }); +}); diff --git a/test/dsh-recall-cache.test.ts b/test/dsh-recall-cache.test.ts new file mode 100644 index 0000000..268d02c --- /dev/null +++ b/test/dsh-recall-cache.test.ts @@ -0,0 +1,126 @@ +/** + * graph-memory — DSH recallCache 回归测试 + * + * By: adoresever + * Email: Wywelljob@gmail.com + * + * 覆盖 dsh.ts 的 system-prompt/assemble 召回缓存:一次失败的召回必须被 + * 从 recallCache 驱逐,否则同一 rejected Promise 会被钉在缓存里,之后每轮 + * assemble 都 re-await 它,把单次瞬时错误变成每轮必现的重复报错。 + */ + +import { describe, expect, it, vi } from "vitest"; + +const { recallMock } = vi.hoisted(() => ({ recallMock: vi.fn() })); + +vi.mock("../src/recaller/recall.ts", () => { + class MockRecaller { + static instances: MockRecaller[] = []; + recall = recallMock; + setEmbedFn = vi.fn(); + syncEmbed = vi.fn(); + constructor(_db: unknown, _cfg: unknown) { + MockRecaller.instances.push(this); + } + } + return { Recaller: MockRecaller }; +}); + +import { apply } from "../dsh.ts"; + +interface Handler { + (...args: any[]): any; +} + +function mockCtx() { + const handlers = new Map(); + const disposers: Array<() => void | Promise> = []; + return { + handlers, + disposers, + ctx: { + logger: { info() {}, warn() {}, error() {} }, + llm: { + stream() { + throw new Error("llm.stream should not be used in this test"); + }, + }, + tools: { register() { return () => {}; } }, + credentials: { + async resolve() { return undefined; }, + }, + on(event: string, listener: Handler) { + handlers.set(event, listener); + return () => {}; + }, + effect(register: () => () => void | Promise) { + const disposer = register(); + disposers.push(disposer); + return () => {}; + }, + }, + }; +} + +async function dispose(harness: { disposers: Array<() => void | Promise> }): Promise { + for (const disposer of harness.disposers) await disposer(); +} + +describe("DSH recall cache", () => { + beforeEach(() => { + recallMock.mockReset(); + }); + + it("evicts a failed recall so the next assembly retries instead of re-awaiting the rejected promise", async () => { + recallMock + .mockRejectedValueOnce(new Error("transient: database is locked")) + .mockResolvedValue({ nodes: [], edges: [] }); + + const harness = mockCtx(); + const { ctx, handlers } = harness; + apply(ctx, { dbPath: ":memory:", extractionEnabled: false }); + try { + const claimed = handlers.get("agent/inbox/claimed")!; + const assemble = handlers.get("system-prompt/assemble")!; + + claimed({ agent: { id: "agent-1" }, message: { source: { kind: "user" }, content: "remember anything about sqlite?" } }); + + const assembly = { contexts: [] as any[] }; + await assemble(assembly, { agent: { id: "agent-1" } }, async () => {}); + // First attempt failed (recall rejected). + expect(recallMock).toHaveBeenCalledTimes(1); + + await assemble(assembly, { agent: { id: "agent-1" } }, async () => {}); + // Regression: without eviction the second assembly re-awaits the pinned + // rejected Promise (still 1 call) and fails again. With eviction it must + // run a fresh recall. + expect(recallMock).toHaveBeenCalledTimes(2); + } finally { + await dispose(harness); + } + }); + + it("keeps caching a successful recall for the same agent and query", async () => { + recallMock.mockResolvedValue({ nodes: [], edges: [] }); + + const harness = mockCtx(); + const { ctx, handlers } = harness; + apply(ctx, { dbPath: ":memory:", extractionEnabled: false }); + try { + const claimed = handlers.get("agent/inbox/claimed")!; + const assemble = handlers.get("system-prompt/assemble")!; + + claimed({ agent: { id: "agent-2" }, message: { source: { kind: "user" }, content: "what do I know about pagerank?" } }); + + const assembly = { contexts: [] as any[] }; + await assemble(assembly, { agent: { id: "agent-2" } }, async () => {}); + await assemble(assembly, { agent: { id: "agent-2" } }, async () => {}); + await assemble(assembly, { agent: { id: "agent-2" } }, async () => {}); + + // Success must stay cached: repeated assemblies reuse the same recall. + expect(recallMock).toHaveBeenCalledTimes(1); + } finally { + await dispose(harness); + } + }); +});