From 93dc5a727376357306be132168b75329e5b4d614 Mon Sep 17 00:00:00 2001 From: Fusion Date: Tue, 12 May 2026 03:23:06 -0700 Subject: [PATCH] feat(FN-4083): add WAL enforcement, immediate write transactions, and concu Hardened SQLite concurrent-write safety in `@fusion/core` by auditing WAL enforcement and adding immediate write transactions to prevent stale reads under load, with executor semaphore limit tests in `@fusion/engine` validating the concurrency bounds; added a new `store-concurrent-writes.test.ts` in Fusion-Task-Id: FN-4083 Fusion-Task-Lineage: adabe51a-a9c4-481a-850d-985c373df51a --- .changeset/fn-4083-sqlite-concurrency.md | 5 + docs/storage.md | 19 +- .../core/src/__tests__/central-db.test.ts | 10 + packages/core/src/__tests__/db.test.ts | 32 ++- .../core/src/__tests__/fts5-guard.test.ts | 25 ++ .../__tests__/store-concurrent-writes.test.ts | 240 ++++++++++++++++++ .../core/src/__tests__/store-watcher.test.ts | 16 +- packages/core/src/archive-db.ts | 2 +- packages/core/src/central-db.ts | 5 +- packages/core/src/db.ts | 26 +- packages/core/src/store.ts | 4 +- .../engine/src/__tests__/concurrency.test.ts | 26 ++ .../src/__tests__/worktree-db-hydrate.test.ts | 2 + packages/engine/src/worktree-db-hydrate.ts | 2 + 14 files changed, 384 insertions(+), 30 deletions(-) create mode 100644 .changeset/fn-4083-sqlite-concurrency.md create mode 100644 packages/core/src/__tests__/store-concurrent-writes.test.ts diff --git a/.changeset/fn-4083-sqlite-concurrency.md b/.changeset/fn-4083-sqlite-concurrency.md new file mode 100644 index 000000000..7a39d9dd6 --- /dev/null +++ b/.changeset/fn-4083-sqlite-concurrency.md @@ -0,0 +1,5 @@ +--- +"@runfusion/fusion": patch +--- + +Improve SQLite write reliability under concurrent executor activity by enforcing WAL/busy-timeout setup on every disk-backed connection, using explicit immediate transactions for task+audit writes, and adding disk-backed concurrent-write regression coverage. diff --git a/docs/storage.md b/docs/storage.md index d1c9325d2..b4e3e425c 100644 --- a/docs/storage.md +++ b/docs/storage.md @@ -7,14 +7,21 @@ - `config.nextId` is retained only as a legacy compatibility field and optional seed source; runtime task creation no longer mutates it as allocator truth. - Startup allocator reconciliation bumps each active prefix sequence to `max(current nextSequence, max(existing task suffix)+1)` across live + archived tasks to self-heal stale allocator drift. -## SQLite write-path lock recovery (FN-4042) +## SQLite write-path lock recovery (FN-4042 / FN-4083) -- Project and central SQLite connections run in WAL mode with a configured `busy_timeout`, but executor-style writes now add a second bounded recovery layer around outermost write transactions. -- `Database.transaction()` and `CentralDatabase.transaction()` acquire outermost transactions with `BEGIN IMMEDIATE`, so writer-lock contention is detected before the callback executes. This allows retrying lock acquisition without re-running user code. +- Every disk-backed SQLite connection that Fusion opens for project storage (`fusion.db`), the central registry (`fusion-central.db`), archives (`archive.db`), and worktree hydration explicitly sets `PRAGMA busy_timeout = 5000` and `PRAGMA journal_mode = WAL` at connection open time before write work begins. +- Project database transactions now distinguish read and write intent: + - `Database.transaction()` uses `BEGIN` (DEFERRED) for outermost transactions so read-only callers do not reserve the writer lock up front. + - `Database.transactionImmediate()` uses `BEGIN IMMEDIATE` for write-heavy paths that must detect writer contention before user code runs. +- The shared task mutation path `atomicWriteTaskJsonWithAudit()` uses `transactionImmediate()`, so the task-row upsert and matching `runAuditEvents` insert still commit or roll back together, while lock contention is detected before the callback mutates in-memory state. +- `CentralDatabase.transaction()` remains `BEGIN IMMEDIATE`-based because its current callers are write-oriented coordination updates; nested transactions still use SQLite `SAVEPOINT` / `ROLLBACK TO` / `RELEASE` semantics in both databases. - Recovery is intentionally bounded: transient `SQLITE_BUSY` / `SQLITE_LOCKED` failures on outermost `BEGIN IMMEDIATE` and `COMMIT` are retried for a short additional window with small synchronous backoff sleeps. If the lock does not clear, the original write still fails loudly. -- Nested transactions still use SQLite `SAVEPOINT` / `ROLLBACK TO` / `RELEASE`, so inner rollback semantics are unchanged: an inner failure can roll back independently and the outer transaction can continue. -- Task writes that use `atomicWriteTaskJsonWithAudit()` still keep the task-row upsert and matching `runAuditEvents` insert in one SQLite transaction. They either commit together or roll back together; the compatibility `task.json` write still happens only after the SQLite transaction succeeds. -- Direct `recordRunAuditEvent()` writes now also execute inside the shared transaction helper so they benefit from the same lock recovery and do not duplicate rows during transient contention. +- Concurrent-write guarantees are layered: + - per-task mutations inside one engine process are serialized by `TaskStore.withTaskLock()` + - cross-task writes rely on WAL mode plus `busy_timeout` + - write-heavy transactional hot paths acquire `BEGIN IMMEDIATE` before mutating state + - compatibility `task.json` writes still happen only after the SQLite transaction succeeds +- Direct `recordRunAuditEvent()` writes continue to execute inside the shared transaction helper so they benefit from the same lock recovery and do not duplicate rows during transient contention. ## 1) Summary diff --git a/packages/core/src/__tests__/central-db.test.ts b/packages/core/src/__tests__/central-db.test.ts index 0809d84b8..26c00c4e9 100644 --- a/packages/core/src/__tests__/central-db.test.ts +++ b/packages/core/src/__tests__/central-db.test.ts @@ -42,6 +42,16 @@ describe("CentralDatabase", () => { expect(db.getSchemaVersion()).toBe(10); }); + it("should enable WAL mode and busy_timeout", () => { + db.init(); + + const journalMode = db.prepare("PRAGMA journal_mode").get() as { journal_mode: string }; + const busyTimeout = db.prepare("PRAGMA busy_timeout").get() as Record; + + expect(journalMode.journal_mode).toBe("wal"); + expect(Object.values(busyTimeout)[0]).toBe(5000); + }); + it("should seed lastModified on init", () => { db.init(); const lastModified = db.getLastModified(); diff --git a/packages/core/src/__tests__/db.test.ts b/packages/core/src/__tests__/db.test.ts index b34c45a3c..c6998dfba 100644 --- a/packages/core/src/__tests__/db.test.ts +++ b/packages/core/src/__tests__/db.test.ts @@ -736,14 +736,34 @@ describe("Database", () => { expect(rowB).toBeUndefined(); }); - it("recovers outermost disk-backed transactions after a transient writer lock", async () => { + it("allows deferred read-only transactions to start while another connection holds the writer lock", async () => { + const dbPath = db.getPath(); + db.exec("PRAGMA busy_timeout = 0"); + const lock = await holdWriteLock(dbPath, { releaseMode: "manual" }); + let callbackCalls = 0; + + try { + const rowCount = db.transaction(() => { + callbackCalls += 1; + return (db.prepare("SELECT COUNT(*) AS count FROM tasks").get() as { count: number }).count; + }); + + expect(rowCount).toBe(0); + } finally { + await lock.release(); + } + + expect(callbackCalls).toBe(1); + }); + + it("recovers outermost immediate 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(() => { + db.transactionImmediate(() => { callbackCalls += 1; db.prepare( "INSERT INTO tasks (id, description, \"column\", createdAt, updatedAt) VALUES (?, ?, ?, ?, ?)" @@ -760,14 +780,14 @@ describe("Database", () => { expect(row).toEqual({ id: "FN-LOCK-RECOVER", description: "Recovered after lock" }); }); - it("preserves nested savepoint rollback semantics after recovering the outer writer lock", async () => { + it("preserves nested savepoint rollback semantics after recovering the outer immediate 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(() => { + db.transactionImmediate(() => { callbackCalls += 1; db.prepare( "INSERT INTO tasks (id, description, \"column\", createdAt, updatedAt) VALUES (?, ?, ?, ?, ?)" @@ -798,7 +818,7 @@ describe("Database", () => { 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 () => { + it("fails without invoking the callback when an immediate lock outlives the recovery window", async () => { const retryDb = new Database(fusionDir, { busyTimeoutMs: 0, lockRecoveryWindowMs: 100, @@ -810,7 +830,7 @@ describe("Database", () => { try { expect(() => { - retryDb.transaction(() => { + retryDb.transactionImmediate(() => { callbackCalls += 1; retryDb.prepare( "INSERT INTO tasks (id, description, \"column\", createdAt, updatedAt) VALUES (?, ?, ?, ?, ?)" diff --git a/packages/core/src/__tests__/fts5-guard.test.ts b/packages/core/src/__tests__/fts5-guard.test.ts index 9d21deafe..b350e9a31 100644 --- a/packages/core/src/__tests__/fts5-guard.test.ts +++ b/packages/core/src/__tests__/fts5-guard.test.ts @@ -93,6 +93,31 @@ describe("FTS5 runtime guard", () => { }); }); + describe("ArchiveDatabase", () => { + let tmpDir: string; + let fusionDir: string; + let archive: ArchiveDatabase; + + beforeEach(() => { + tmpDir = makeTmpDir(); + fusionDir = join(tmpDir, ".fusion"); + archive = new ArchiveDatabase(fusionDir); + }); + + afterEach(async () => { + try { archive.close(); } catch { /* already closed */ } + await rm(tmpDir, { recursive: true, force: true }); + }); + + it("enables WAL mode and busy_timeout for disk-backed archives", () => { + archive.init(); + const journalMode = (archive as any).db.prepare("PRAGMA journal_mode").get() as { journal_mode: string }; + const busyTimeout = (archive as any).db.prepare("PRAGMA busy_timeout").get() as Record; + expect(journalMode.journal_mode).toBe("wal"); + expect(Object.values(busyTimeout)[0]).toBe(5000); + }); + }); + describe("TaskStore.searchTasks LIKE fallback", () => { let rootDir: string; let globalDir: string; diff --git a/packages/core/src/__tests__/store-concurrent-writes.test.ts b/packages/core/src/__tests__/store-concurrent-writes.test.ts new file mode 100644 index 000000000..d2593cec1 --- /dev/null +++ b/packages/core/src/__tests__/store-concurrent-writes.test.ts @@ -0,0 +1,240 @@ +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { mkdtempSync } from "node:fs"; +import { rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +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 { RunMutationContext, Task } from "../types.js"; + +function makeTmpDir(): string { + return mkdtempSync(join(tmpdir(), "fn-store-concurrent-test-")); +} + +async function holdWriteLock( + dbPath: string, + options?: { holdMs?: number; releaseMode?: "manual" | "timer" }, +): Promise<{ + child: ChildProcessWithoutNullStreams; + release: () => Promise; +}> { + 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 busy_timeout = 0"); + db.exec("PRAGMA journal_mode = WAL"); + db.exec("BEGIN IMMEDIATE"); + db.exec(\"INSERT INTO tasks (id, description, \\\"column\\\", createdAt, updatedAt) VALUES ('FN-LOCK-HELPER', 'lock helper', 'todo', '2025-01-01', '2025-01-01') ON CONFLICT(id) DO NOTHING\"); + 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((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"); + }, + }; +} + +async function createStores(rootDir: string, globalDir: string, count: number): Promise { + const stores = Array.from({ length: count }, () => new TaskStore(rootDir, globalDir)); + for (const store of stores) { + await store.init(); + } + return stores; +} + +describe("TaskStore concurrent writes", () => { + let rootDir: string; + let globalDir: string; + let fusionDir: string; + let stores: TaskStore[]; + let primary: TaskStore; + + beforeEach(async () => { + rootDir = makeTmpDir(); + globalDir = makeTmpDir(); + fusionDir = join(rootDir, ".fusion"); + stores = await createStores(rootDir, globalDir, 4); + primary = stores[0]; + }); + + afterEach(async () => { + for (const store of stores) { + try { + store.stopWatching(); + store.close(); + } catch { + // ignore + } + } + await rm(rootDir, { recursive: true, force: true }); + await rm(globalDir, { recursive: true, force: true }); + }); + + it("uses WAL on each disk-backed connection and recovers an immediate write after a transient lock", async () => { + const dbA = new Database(fusionDir, { busyTimeoutMs: 0 }); + const dbB = new Database(fusionDir, { busyTimeoutMs: 0 }); + dbA.init(); + dbB.init(); + + const journalA = dbA.prepare("PRAGMA journal_mode").get() as { journal_mode: string }; + const journalB = dbB.prepare("PRAGMA journal_mode").get() as { journal_mode: string }; + expect(journalA.journal_mode).toBe("wal"); + expect(journalB.journal_mode).toBe("wal"); + + const lock = await holdWriteLock(dbA.getPath(), { releaseMode: "timer", holdMs: 150 }); + let callbackCalls = 0; + + try { + dbB.transactionImmediate(() => { + callbackCalls += 1; + dbB.prepare( + 'INSERT INTO tasks (id, description, "column", createdAt, updatedAt) VALUES (?, ?, ?, ?, ?)', + ).run("FN-WAL-RECOVER", "Recovered write", "todo", "2025-01-01", "2025-01-01"); + }); + } finally { + await lock.release(); + dbA.close(); + dbB.close(); + } + + const verifyDb = new Database(fusionDir); + verifyDb.init(); + const row = verifyDb.prepare("SELECT id FROM tasks WHERE id = ?").get("FN-WAL-RECOVER"); + verifyDb.close(); + + expect(callbackCalls).toBe(1); + expect(row).toBeDefined(); + }); + + it("serializes same-task disk-backed writes through withTaskLock", async () => { + const task = await primary.createTask({ description: "Same-task serialization" }); + + await Promise.all( + Array.from({ length: 20 }, (_, index) => { + if (index % 2 === 0) { + return primary.logEntry(task.id, `same-task-log-${index}`); + } + return primary.updateTask(task.id, { title: `Title ${index}` }); + }), + ); + + const updated = await primary.getTask(task.id); + const customLogs = updated.log.filter((entry) => entry.action.startsWith("same-task-log-")); + + expect(customLogs).toHaveLength(10); + expect(updated.title).toBe("Title 19"); + }); + + it("updates different tasks concurrently across store connections without data loss", async () => { + const tasks = await Promise.all( + Array.from({ length: 16 }, (_, index) => primary.createTask({ description: `Concurrent task ${index}` })), + ); + + await Promise.all( + tasks.map((task, index) => + stores[index % stores.length].updateTask(task.id, { + title: `Updated title ${index}`, + description: `Updated description ${index}`, + }), + ), + ); + + const reloaded = await Promise.all(tasks.map((task) => primary.getTask(task.id))); + reloaded.forEach((task, index) => { + expect(task.title).toBe(`Updated title ${index}`); + expect(task.description).toBe(`Updated description ${index}`); + }); + }); + + it("records audit events atomically for concurrent logEntry writes with runContext", async () => { + const runContextBase: Omit = { + runId: "run-concurrent-log-entry", + }; + const tasks = await Promise.all( + Array.from({ length: 12 }, (_, index) => primary.createTask({ description: `Audit task ${index}` })), + ); + + await Promise.all( + tasks.map((task, index) => + stores[index % stores.length].logEntry( + task.id, + `audit-log-${index}`, + undefined, + { ...runContextBase, agentId: `agent-${index % 3}` }, + ), + ), + ); + + const events = primary.getRunAuditEvents({ runId: runContextBase.runId }); + expect(events).toHaveLength(tasks.length); + expect(events.every((event) => event.mutationType === "task:log")).toBe(true); + + const updatedTasks = await Promise.all(tasks.map((task) => primary.getTask(task.id))); + updatedTasks.forEach((task, index) => { + expect(task.log.some((entry) => entry.action === `audit-log-${index}`)).toBe(true); + }); + }); + + it("moves different tasks concurrently without SQLITE_BUSY failures", async () => { + const tasks: Task[] = await Promise.all( + Array.from({ length: 10 }, (_, index) => primary.createTask({ description: `Move task ${index}` })), + ); + + await Promise.all( + tasks.map((task, index) => stores[index % stores.length].moveTask(task.id, "todo")), + ); + + const moved = await Promise.all(tasks.map((task) => primary.getTask(task.id))); + moved.forEach((task) => { + expect(task.column).toBe("todo"); + expect(task.status).toBeUndefined(); + }); + }); +}); diff --git a/packages/core/src/__tests__/store-watcher.test.ts b/packages/core/src/__tests__/store-watcher.test.ts index f5638ad4a..48ed67028 100644 --- a/packages/core/src/__tests__/store-watcher.test.ts +++ b/packages/core/src/__tests__/store-watcher.test.ts @@ -77,14 +77,18 @@ describe("TaskStore", () => { await harness.store().watch(); const storeAny = harness.store() as any; - const firstCall = storeAny.checkForChanges(); - const secondCall = storeAny.checkForChanges(); + try { + const firstCall = storeAny.checkForChanges(); + const secondCall = storeAny.checkForChanges(); - expect(firstCall).toBeInstanceOf(Promise); - expect(secondCall).toBeInstanceOf(Promise); + expect(firstCall).toBeInstanceOf(Promise); + expect(secondCall).toBeInstanceOf(Promise); - await Promise.all([firstCall, secondCall]); - expect(storeAny.pollingInProgress).toBe(false); + await Promise.all([firstCall, secondCall]); + expect(storeAny.pollingInProgress).toBe(false); + } finally { + harness.store().stopWatching(); + } }); it("logs poll failures with context and keeps checkForChanges non-fatal", async () => { diff --git a/packages/core/src/archive-db.ts b/packages/core/src/archive-db.ts index 0e1eae3cb..3457e6600 100644 --- a/packages/core/src/archive-db.ts +++ b/packages/core/src/archive-db.ts @@ -63,10 +63,10 @@ export class ArchiveDatabase { mkdirSync(fusionDir, { recursive: true }); } this.db = new DatabaseSync(inMemory ? ":memory:" : join(fusionDir, "archive.db")); + this.db.exec("PRAGMA busy_timeout = 5000"); if (!inMemory) { this.db.exec("PRAGMA journal_mode = WAL"); } - this.db.exec("PRAGMA busy_timeout = 5000"); this._fts5Available = probeFts5(this.db); } diff --git a/packages/core/src/central-db.ts b/packages/core/src/central-db.ts index ae0c27bb1..85ff66c32 100644 --- a/packages/core/src/central-db.ts +++ b/packages/core/src/central-db.ts @@ -467,10 +467,11 @@ export class CentralDatabase { throw new Error(`Failed to open Fusion central database at ${this.dbPath}: ${message}`); } + // Wait up to the configured timeout for locks to clear before returning SQLITE_BUSY. + // Set this before other PRAGMAs so they also benefit. + this.db.exec(`PRAGMA busy_timeout = ${this.busyTimeoutMs}`); // Enable WAL mode for concurrent reader/writer access this.db.exec("PRAGMA journal_mode = WAL"); - // Wait up to the configured timeout for locks to clear before returning SQLITE_BUSY. - this.db.exec(`PRAGMA busy_timeout = ${this.busyTimeoutMs}`); // Enable foreign key enforcement this.db.exec("PRAGMA foreign_keys = ON"); } diff --git a/packages/core/src/db.ts b/packages/core/src/db.ts index df00df267..cad783ddd 100644 --- a/packages/core/src/db.ts +++ b/packages/core/src/db.ts @@ -33,6 +33,8 @@ const DEFAULT_SQLITE_BUSY_TIMEOUT_MS = 5_000; const DEFAULT_SQLITE_LOCK_RECOVERY_WINDOW_MS = 1_000; const DEFAULT_SQLITE_LOCK_RECOVERY_DELAY_MS = 50; +type TransactionMode = "deferred" | "immediate"; + // ── JSON Helpers ───────────────────────────────────────────────────── /** @@ -3279,20 +3281,26 @@ export class Database { * If the function throws, the transaction/savepoint is rolled back. * If the function returns normally, the transaction/savepoint is committed. * - * Outermost transactions acquire `BEGIN IMMEDIATE` so transient writer-lock - * contention is detected before user code runs, allowing bounded retry - * without re-executing the callback. Nested transactions remain savepoint-based. + * Outermost transactions default to `BEGIN` (DEFERRED) so read-only callers + * avoid taking a writer lock until they actually mutate state. + * Use `transactionImmediate()` for write-heavy paths that should acquire the + * RESERVED lock before user code runs and fail/retry before the callback executes. */ - transaction(fn: () => T): T { + transaction(fn: () => T, options?: { mode?: TransactionMode }): T { const depth = this.transactionDepth++; const isOutermost = depth === 0; const savepointName = `sp_${depth}`; + const mode: TransactionMode = options?.mode ?? "deferred"; try { if (isOutermost) { - this.runWithLockRecovery("BEGIN IMMEDIATE", () => { - this.db.exec("BEGIN IMMEDIATE"); - }); + if (mode === "immediate") { + this.runWithLockRecovery("BEGIN IMMEDIATE", () => { + this.db.exec("BEGIN IMMEDIATE"); + }); + } else { + this.db.exec("BEGIN"); + } } else { this.db.exec(`SAVEPOINT ${savepointName}`); } @@ -3324,6 +3332,10 @@ export class Database { } } + transactionImmediate(fn: () => T): T { + return this.transaction(fn, { mode: "immediate" }); + } + /** * Execute plugin-provided schema initialization hooks. * diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index b19d6906b..1c4d248b3 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -1740,7 +1740,7 @@ export class TaskStore extends EventEmitter { task: Task, auditInput?: RunAuditEventInput, ): Promise { - this.db.transaction(() => { + this.db.transactionImmediate(() => { // Upsert the task this.upsertTaskWithFtsRecovery(task); @@ -4137,7 +4137,7 @@ export class TaskStore extends EventEmitter { metadata: input.metadata, }; - this.db.transaction(() => { + this.db.transactionImmediate(() => { this.db.prepare(` INSERT INTO runAuditEvents ( id, timestamp, taskId, agentId, runId, domain, mutationType, target, metadata diff --git a/packages/engine/src/__tests__/concurrency.test.ts b/packages/engine/src/__tests__/concurrency.test.ts index 8415081b5..258145ce9 100644 --- a/packages/engine/src/__tests__/concurrency.test.ts +++ b/packages/engine/src/__tests__/concurrency.test.ts @@ -199,6 +199,32 @@ describe("AgentSemaphore", () => { expect(sem.activeCount).toBe(0); }); + it("keeps executor-style write sections within the configured execute concurrency", async () => { + const sem = new AgentSemaphore(2); + let concurrentWrites = 0; + let maxConcurrentWrites = 0; + + const performWrite = (taskId: string) => + sem.run(async () => { + concurrentWrites += 1; + maxConcurrentWrites = Math.max(maxConcurrentWrites, concurrentWrites); + await new Promise((resolve) => setTimeout(resolve, 10)); + concurrentWrites -= 1; + return taskId; + }, PRIORITY_EXECUTE); + + const completed = await Promise.all([ + performWrite("FN-1"), + performWrite("FN-2"), + performWrite("FN-3"), + performWrite("FN-4"), + ]); + + expect(completed).toEqual(["FN-1", "FN-2", "FN-3", "FN-4"]); + expect(maxConcurrentWrites).toBe(2); + expect(sem.activeCount).toBe(0); + }); + it("integration: shared semaphore limits triage + execution + merge together", async () => { const sem = new AgentSemaphore(2); let concurrent = 0; diff --git a/packages/engine/src/__tests__/worktree-db-hydrate.test.ts b/packages/engine/src/__tests__/worktree-db-hydrate.test.ts index bdc7e353a..efcb5470a 100644 --- a/packages/engine/src/__tests__/worktree-db-hydrate.test.ts +++ b/packages/engine/src/__tests__/worktree-db-hydrate.test.ts @@ -65,9 +65,11 @@ describe("hydrateWorktreeDb", () => { const db = new DatabaseSync(join(worktree, ".fusion", "fusion.db")); const tasks = (db.prepare("SELECT COUNT(*) as c FROM tasks WHERE id IN ('FN-A','FN-B','FN-C')").get() as any).c; const docs = (db.prepare("SELECT COUNT(*) as c FROM task_documents WHERE taskId='FN-B'").get() as any).c; + const journalMode = db.prepare("PRAGMA journal_mode").get() as { journal_mode: string }; db.close(); expect(tasks).toBe(3); expect(docs).toBe(1); + expect(journalMode.journal_mode).toBe("wal"); }); it("no-op when rootDir === worktreePath", async () => { diff --git a/packages/engine/src/worktree-db-hydrate.ts b/packages/engine/src/worktree-db-hydrate.ts index 29889588a..e0d139183 100644 --- a/packages/engine/src/worktree-db-hydrate.ts +++ b/packages/engine/src/worktree-db-hydrate.ts @@ -113,8 +113,10 @@ export async function hydrateWorktreeDb({ } srcDb = new DatabaseSync(srcDbPath); + srcDb.exec("PRAGMA busy_timeout = 5000"); dstDb = openWorktreeDbWithRecovery(dstDbPath, worktreePath); + dstDb.exec("PRAGMA busy_timeout = 5000"); dstDb.exec("PRAGMA journal_mode = WAL"); const srcTaskCols = getColumns(srcDb, "tasks");