feat(FN-4042): add SQLite lock contention recovery in core database layer

Adds SQLite lock contention recovery for WAL-mode databases, covering both the core database layer (db.ts, central-db.ts, store.ts) and the run-audit subsystem, with comprehensive unit and integration test coverage and updated storage documentation. Fusion-Task-Id: FN-4042
2026-05-12 00:40:40 -07:00
parent 8b0fc50a84
commit 43c8aa58f9
8 changed files with 540 additions and 36 deletions
--- a/packages/core/src/tests/db.test.ts
+++ b/packages/core/src/tests/db.test.ts
@@ -16,6 +16,8 @@ import { join, dirname } from "node:path";
 import { tmpdir } from "node:os";
 import { fileURLToPath } from "node:url";
 import { rm } from "node:fs/promises";
+import { once } from "node:events";
+import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
 import { ensureRoadmapSchema } from "../../../../plugins/fusion-plugin-roadmap/src/roadmap-schema.js";

 const createdTmpDirs = new Set<string>();
@@ -53,6 +55,77 @@ afterAll(() => {
  cleanupTmpDirsSync();
 });

+async function holdWriteLock(
+  dbPath: string,
+  options?: { holdMs?: number; releaseMode?: "manual" | "timer" },
+): Promise<{
+  child: ChildProcessWithoutNullStreams;
+  release: () => Promise<void>;
+}> {
+  const releaseMode = options?.releaseMode ?? "manual";
+  const holdMs = options?.holdMs ?? 0;
+  const script = `
+    const { DatabaseSync } = require("node:sqlite");
+    const db = new DatabaseSync(${JSON.stringify(dbPath)});
+    db.exec("PRAGMA journal_mode = WAL");
+    db.exec("PRAGMA busy_timeout = 0");
+    db.exec("BEGIN IMMEDIATE");
+    process.stdout.write("LOCKED\\n");
+    const release = () => {
+      try { db.exec("COMMIT"); } catch {}
+      try { db.close(); } catch {}
+      process.exit(0);
+    };
+    if (${JSON.stringify(releaseMode)} === "timer") {
+      setTimeout(release, ${holdMs});
+    } else {
+      process.stdin.setEncoding("utf8");
+      process.stdin.on("data", (chunk) => {
+        if (chunk.includes("RELEASE")) release();
+      });
+    }
+  `;
+
+  const child = spawn(process.execPath, ["-e", script], {
+    stdio: ["pipe", "pipe", "pipe"],
+  });
+
+  const ready = new Promise<void>((resolve, reject) => {
+    let stderr = "";
+    child.stderr.on("data", (chunk) => {
+      stderr += chunk.toString();
+    });
+    child.stdout.on("data", (chunk) => {
+      if (chunk.toString().includes("LOCKED")) {
+        resolve();
+      }
+    });
+    child.once("exit", (code) => {
+      if (code !== 0) {
+        reject(new Error(`Lock helper exited early (${code}): ${stderr || "no stderr"}`));
+      }
+    });
+    child.once("error", reject);
+  });
+
+  await ready;
+
+  return {
+    child,
+    release: async () => {
+      if (child.exitCode !== null || child.killed) {
+        return;
+      }
+      if (releaseMode === "timer") {
+        await once(child, "exit");
+        return;
+      }
+      child.stdin.write("RELEASE\n");
+      await once(child, "exit");
+    },
+  };
+}
+
 describe("Database", () => {
  let tmpDir: string;
  let fusionDir: string;
@@ -662,6 +735,96 @@ describe("Database", () => {
      expect(rowA).toBeUndefined();
      expect(rowB).toBeUndefined();
    });
+
+    it("recovers outermost disk-backed transactions after a transient writer lock", async () => {
+      const dbPath = db.getPath();
+      db.exec("PRAGMA busy_timeout = 0");
+      const lock = await holdWriteLock(dbPath, { releaseMode: "timer", holdMs: 150 });
+      let callbackCalls = 0;
+
+      try {
+        db.transaction(() => {
+          callbackCalls += 1;
+          db.prepare(
+            "INSERT INTO tasks (id, description, \"column\", createdAt, updatedAt) VALUES (?, ?, ?, ?, ?)"
+          ).run("FN-LOCK-RECOVER", "Recovered after lock", "todo", "2025-01-01", "2025-01-01");
+        });
+      } finally {
+        await lock.release();
+      }
+
+      const row = db.prepare("SELECT id, description FROM tasks WHERE id = ?").get("FN-LOCK-RECOVER") as
+        | { id: string; description: string }
+        | undefined;
+      expect(callbackCalls).toBe(1);
+      expect(row).toEqual({ id: "FN-LOCK-RECOVER", description: "Recovered after lock" });
+    });
+
+    it("preserves nested savepoint rollback semantics after recovering the outer writer lock", async () => {
+      const dbPath = db.getPath();
+      db.exec("PRAGMA busy_timeout = 0");
+      const lock = await holdWriteLock(dbPath, { releaseMode: "timer", holdMs: 150 });
+      let callbackCalls = 0;
+
+      try {
+        db.transaction(() => {
+          callbackCalls += 1;
+          db.prepare(
+            "INSERT INTO tasks (id, description, \"column\", createdAt, updatedAt) VALUES (?, ?, ?, ?, ?)"
+          ).run("FN-LOCK-OUTER", "Outer task", "todo", "2025-01-01", "2025-01-01");
+
+          try {
+            db.transaction(() => {
+              db.prepare(
+                "INSERT INTO tasks (id, description, \"column\", createdAt, updatedAt) VALUES (?, ?, ?, ?, ?)"
+              ).run("FN-LOCK-INNER", "Inner task", "todo", "2025-01-01", "2025-01-01");
+              throw new Error("inner rollback");
+            });
+          } catch (error) {
+            expect((error as Error).message).toBe("inner rollback");
+          }
+
+          db.prepare(
+            "INSERT INTO tasks (id, description, \"column\", createdAt, updatedAt) VALUES (?, ?, ?, ?, ?)"
+          ).run("FN-LOCK-POST", "After inner rollback", "todo", "2025-01-01", "2025-01-01");
+        });
+      } finally {
+        await lock.release();
+      }
+
+      expect(callbackCalls).toBe(1);
+      expect(db.prepare("SELECT id FROM tasks WHERE id = ?").get("FN-LOCK-OUTER")).toBeDefined();
+      expect(db.prepare("SELECT id FROM tasks WHERE id = ?").get("FN-LOCK-INNER")).toBeUndefined();
+      expect(db.prepare("SELECT id FROM tasks WHERE id = ?").get("FN-LOCK-POST")).toBeDefined();
+    });
+
+    it("fails without invoking the callback when the lock outlives the recovery window", async () => {
+      const retryDb = new Database(fusionDir, {
+        busyTimeoutMs: 0,
+        lockRecoveryWindowMs: 100,
+        lockRecoveryDelayMs: 25,
+      });
+      retryDb.init();
+      const lock = await holdWriteLock(retryDb.getPath(), { releaseMode: "manual" });
+      let callbackCalls = 0;
+
+      try {
+        expect(() => {
+          retryDb.transaction(() => {
+            callbackCalls += 1;
+            retryDb.prepare(
+              "INSERT INTO tasks (id, description, \"column\", createdAt, updatedAt) VALUES (?, ?, ?, ?, ?)"
+            ).run("FN-LOCK-TIMEOUT", "Should not write", "todo", "2025-01-01", "2025-01-01");
+          });
+        }).toThrow(/BEGIN IMMEDIATE failed/);
+      } finally {
+        await lock.release();
+        retryDb.close();
+      }
+
+      expect(callbackCalls).toBe(0);
+      expect(db.prepare("SELECT id FROM tasks WHERE id = ?").get("FN-LOCK-TIMEOUT")).toBeUndefined();
+    });
  });

  describe("runPluginSchemaInits", () => {
--- a/packages/core/src/tests/run-audit.integration.test.ts
+++ b/packages/core/src/tests/run-audit.integration.test.ts
@@ -16,6 +16,8 @@ import { mkdtempSync } from "node:fs";
 import { rm } from "node:fs/promises";
 import { join } from "node:path";
 import { tmpdir } from "node:os";
+import { once } from "node:events";
+import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
 import { Database } from "../db.js";
 import { TaskStore } from "../store.js";
 import type { RunAuditEventInput, RunAuditEvent } from "../types.js";
@@ -24,6 +26,77 @@ function makeTmpDir(): string {
  return mkdtempSync(join(tmpdir(), "fn-run-audit-integration-test-"));
 }

+async function holdWriteLock(
+  dbPath: string,
+  options?: { holdMs?: number; releaseMode?: "manual" | "timer" },
+): Promise<{
+  child: ChildProcessWithoutNullStreams;
+  release: () => Promise<void>;
+}> {
+  const releaseMode = options?.releaseMode ?? "manual";
+  const holdMs = options?.holdMs ?? 0;
+  const script = `
+    const { DatabaseSync } = require("node:sqlite");
+    const db = new DatabaseSync(${JSON.stringify(dbPath)});
+    db.exec("PRAGMA journal_mode = WAL");
+    db.exec("PRAGMA busy_timeout = 0");
+    db.exec("BEGIN IMMEDIATE");
+    process.stdout.write("LOCKED\\n");
+    const release = () => {
+      try { db.exec("COMMIT"); } catch {}
+      try { db.close(); } catch {}
+      process.exit(0);
+    };
+    if (${JSON.stringify(releaseMode)} === "timer") {
+      setTimeout(release, ${holdMs});
+    } else {
+      process.stdin.setEncoding("utf8");
+      process.stdin.on("data", (chunk) => {
+        if (chunk.includes("RELEASE")) release();
+      });
+    }
+  `;
+
+  const child = spawn(process.execPath, ["-e", script], {
+    stdio: ["pipe", "pipe", "pipe"],
+  });
+
+  const ready = new Promise<void>((resolve, reject) => {
+    let stderr = "";
+    child.stderr.on("data", (chunk) => {
+      stderr += chunk.toString();
+    });
+    child.stdout.on("data", (chunk) => {
+      if (chunk.toString().includes("LOCKED")) {
+        resolve();
+      }
+    });
+    child.once("exit", (code) => {
+      if (code !== 0) {
+        reject(new Error(`Lock helper exited early (${code}): ${stderr || "no stderr"}`));
+      }
+    });
+    child.once("error", reject);
+  });
+
+  await ready;
+
+  return {
+    child,
+    release: async () => {
+      if (child.exitCode !== null || child.killed) {
+        return;
+      }
+      if (releaseMode === "timer") {
+        await once(child, "exit");
+        return;
+      }
+      child.stdin.write("RELEASE\n");
+      await once(child, "exit");
+    },
+  };
+}
+
 describe("Run Audit Integration", () => {
  let rootDir: string;
  let fusionDir: string;
@@ -253,6 +326,29 @@ describe("Run Audit Integration", () => {
    });
  });

+  describe("disk-backed lock recovery integration", () => {
+    it("keeps task and audit writes atomic under transient multi-connection writer contention", async () => {
+      const task = await store.createTask({ description: "Integration lock recovery task" });
+      const storeDb = (store as any).db as Database;
+      storeDb.exec("PRAGMA busy_timeout = 0");
+      const lock = await holdWriteLock(storeDb.getPath(), { releaseMode: "timer", holdMs: 150 });
+      const runContext = { runId: "run-integration-lock", agentId: "agent-integration-lock" };
+
+      try {
+        await store.updateTask(task.id, { title: "Recovered title" }, runContext);
+      } finally {
+        await lock.release();
+      }
+
+      const events = store.getRunAuditEvents({ runId: "run-integration-lock" });
+      expect(events).toHaveLength(1);
+      expect(events[0].mutationType).toBe("task:update");
+
+      const updatedTask = await store.getTask(task.id);
+      expect(updatedTask.title).toBe("Recovered title");
+    });
+  });
+
  describe("absent run context regression", () => {
    it("recordRunAuditEvent works with minimal required fields", () => {
      // Even without explicit timestamp or full context, should not crash
--- a/packages/core/src/tests/run-audit.test.ts
+++ b/packages/core/src/tests/run-audit.test.ts
@@ -3,6 +3,8 @@ import { mkdtempSync } from "node:fs";
 import { rm } from "node:fs/promises";
 import { join } from "node:path";
 import { tmpdir } from "node:os";
+import { once } from "node:events";
+import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
 import { Database } from "../db.js";
 import { TaskStore } from "../store.js";
 import type { RunAuditEventInput, RunAuditEventFilter } from "../types.js";
@@ -11,6 +13,77 @@ function makeTmpDir(): string {
  return mkdtempSync(join(tmpdir(), "fn-run-audit-test-"));
 }

+async function holdWriteLock(
+  dbPath: string,
+  options?: { holdMs?: number; releaseMode?: "manual" | "timer" },
+): Promise<{
+  child: ChildProcessWithoutNullStreams;
+  release: () => Promise<void>;
+}> {
+  const releaseMode = options?.releaseMode ?? "manual";
+  const holdMs = options?.holdMs ?? 0;
+  const script = `
+    const { DatabaseSync } = require("node:sqlite");
+    const db = new DatabaseSync(${JSON.stringify(dbPath)});
+    db.exec("PRAGMA journal_mode = WAL");
+    db.exec("PRAGMA busy_timeout = 0");
+    db.exec("BEGIN IMMEDIATE");
+    process.stdout.write("LOCKED\\n");
+    const release = () => {
+      try { db.exec("COMMIT"); } catch {}
+      try { db.close(); } catch {}
+      process.exit(0);
+    };
+    if (${JSON.stringify(releaseMode)} === "timer") {
+      setTimeout(release, ${holdMs});
+    } else {
+      process.stdin.setEncoding("utf8");
+      process.stdin.on("data", (chunk) => {
+        if (chunk.includes("RELEASE")) release();
+      });
+    }
+  `;
+
+  const child = spawn(process.execPath, ["-e", script], {
+    stdio: ["pipe", "pipe", "pipe"],
+  });
+
+  const ready = new Promise<void>((resolve, reject) => {
+    let stderr = "";
+    child.stderr.on("data", (chunk) => {
+      stderr += chunk.toString();
+    });
+    child.stdout.on("data", (chunk) => {
+      if (chunk.toString().includes("LOCKED")) {
+        resolve();
+      }
+    });
+    child.once("exit", (code) => {
+      if (code !== 0) {
+        reject(new Error(`Lock helper exited early (${code}): ${stderr || "no stderr"}`));
+      }
+    });
+    child.once("error", reject);
+  });
+
+  await ready;
+
+  return {
+    child,
+    release: async () => {
+      if (child.exitCode !== null || child.killed) {
+        return;
+      }
+      if (releaseMode === "timer") {
+        await once(child, "exit");
+        return;
+      }
+      child.stdin.write("RELEASE\n");
+      await once(child, "exit");
+    },
+  };
+}
+
 describe("Run Audit", () => {
  let rootDir: string;
  let fusionDir: string;
@@ -125,6 +198,30 @@ describe("Run Audit", () => {
      expect(events[0].id).toBe(event.id);
      expect(events[0].runId).toBe("run-002");
    });
+
+    it("retries a direct audit insert after transient disk-backed writer contention without duplicating events", async () => {
+      const storeDb = (store as any).db as Database;
+      storeDb.exec("PRAGMA busy_timeout = 0");
+      const lock = await holdWriteLock(storeDb.getPath(), { releaseMode: "timer", holdMs: 150 });
+
+      try {
+        const event = store.recordRunAuditEvent({
+          taskId: "FN-LOCK-AUDIT",
+          agentId: "agent-lock-audit",
+          runId: "run-lock-audit",
+          domain: "database",
+          mutationType: "task:update",
+          target: "FN-LOCK-AUDIT",
+          metadata: { source: "lock-test" },
+        });
+
+        const events = store.getRunAuditEvents({ runId: "run-lock-audit" });
+        expect(events).toHaveLength(1);
+        expect(events[0]).toMatchObject(event);
+      } finally {
+        await lock.release();
+      }
+    });
  });

  describe("getRunAuditEvents", () => {
@@ -315,6 +412,28 @@ describe("Run Audit", () => {
        expect(updatedTask.title).toBe("Updated title");
      });

+      it("logEntry() with runContext commits exactly one task mutation and one audit row after transient writer contention", async () => {
+        const task = await store.createTask({ description: "Test task for lock recovery" });
+        const storeDb = (store as any).db as Database;
+        storeDb.exec("PRAGMA busy_timeout = 0");
+        const lock = await holdWriteLock(storeDb.getPath(), { releaseMode: "timer", holdMs: 150 });
+        const runContext = { runId: "run-atomic-lock", agentId: "agent-atomic" };
+
+        try {
+          await store.logEntry(task.id, "Recovered under contention", undefined, runContext);
+        } finally {
+          await lock.release();
+        }
+
+        const events = store.getRunAuditEvents({ runId: "run-atomic-lock" });
+        expect(events).toHaveLength(1);
+        expect(events[0].mutationType).toBe("task:log");
+
+        const updatedTask = await store.getTask(task.id);
+        const matchingEntries = updatedTask.log.filter((entry) => entry.action === "Recovered under contention");
+        expect(matchingEntries).toHaveLength(1);
+      });
+
      it("methods without runContext do not record audit events (backward compat)", async () => {
        // Use a unique description to identify our task's audit events
        const uniqueDesc = "Test task backward compat unique " + Date.now();