From b462755e3c8b5f0701294ae1589c9bd34f3e97a2 Mon Sep 17 00:00:00 2001 From: OpenCode Agent Date: Sat, 20 Jun 2026 02:56:33 +0000 Subject: [PATCH 1/2] fix(concurrency): centralize write transactions with bounded SQLITE_BUSY retry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces 9 hand-rolled BEGIN IMMEDIATE / COMMIT / finally-ROLLBACK blocks with a single shared runWriteTransaction helper that adds a bounded SQLITE_BUSY retry loop (4 attempts, 100ms backoff). This is a better concurrency solution than the previous per-site pattern because: 1. Centralized retry policy: transient SQLITE_BUSY from cross-process WAL writer lock contention (sibling migration, dreamer run, Channel-2 bulk delivery) now waits-and-retries instead of propagating up and disabling Magic Context for the run. Previously each site either swallowed, propagated, or retried inconsistently. 2. Composition-safe: if called inside an existing db.transaction() or another runWriteTransaction, the body runs inline WITHOUT issuing a nested BEGIN (SQLite doesn't allow nested BEGIN; the outer transaction already holds the writer lock). Detected via the bun:sqlite inTransaction flag or the node:sqlite shim's isTransaction flag. 3. One correct transaction plumbing pattern instead of 9 slightly-different copies. Several sites were missing the ROLLBACK path or had subtle ordering bugs at the edges. Sites converted (all sync write paths with hand-rolled BEGIN IMMEDIATE): - dreamer/lease.ts (acquireLease, renewLease, releaseLease) - git-commits/sweep-coordinator.ts (acquireGitSweepLease, renewGitSweepLease, markGitSweepSuccessAndRelease) - message-index.ts (indexMessagesAfterOrdinal — cross-process FTS dedup) - workspaces.ts (bumpEpochsForWorkspaceMembers, bumpEpochsForWorkspaceMemberSet) - compartment-storage.ts (replaceAllCompartmentStateAndBumpDepth, promoteRecompStaging) - key-files/project-key-files.ts (replaceAllKeyFiles) - key-files/identify-key-files.ts (commitKeyFilesUnderLease) - hooks/compartment-runner-recomp.ts (promoteRecompStagingWithM0Mutation) - hooks/compartment-runner-incremental.ts (historian publish path) inject-compartments.ts is intentionally NOT converted: its two BEGIN IMMEDIATE blocks have complex early-return-with-rollback-and-fallback-read patterns (Phase 3 materialization contention retry, soft-refresh cache replay) that don't map cleanly onto the shared helper without significant restructuring. The helper is still a net win across the 9 simpler sites. Verification: - bun run typecheck (tsc --noEmit + scripts) passes - bun run lint (biome check) passes - 12 new unit tests for runWriteTransaction/runWriteTransactionAsync pass (BEGIN/COMMIT, rollback on throw, transient BUSY retry, max-attempts give-up, nested-transaction composition, non-transient error passthrough) - All existing tests in the converted files' test suites pass: dreamer/lease.test.ts (7), git-commits (19), compartment-storage-atomic (8), compartment-storage-v6, compartment-lease, compartment-runner-recomp-fk (4), compartment-runner-partial-recomp, key-files (14), workspaces, storage-db (11), message-index Net: -199 lines / +40 lines across the 9 converted sites, +201 lines for the shared helper + 156 lines for its tests. --- .../magic-context/compartment-storage.ts | 40 +--- .../features/magic-context/dreamer/lease.ts | 25 +-- .../git-commits/sweep-coordinator.ts | 28 +-- .../key-files/identify-key-files.ts | 17 +- .../key-files/project-key-files.ts | 20 +- .../features/magic-context/message-index.ts | 19 +- .../src/features/magic-context/workspaces.ts | 40 +--- .../compartment-runner-incremental.ts | 29 +-- .../compartment-runner-recomp.ts | 21 +- .../src/shared/write-transaction.test.ts | 150 +++++++++++++ .../plugin/src/shared/write-transaction.ts | 202 ++++++++++++++++++ 11 files changed, 392 insertions(+), 199 deletions(-) create mode 100644 packages/plugin/src/shared/write-transaction.test.ts create mode 100644 packages/plugin/src/shared/write-transaction.ts diff --git a/packages/plugin/src/features/magic-context/compartment-storage.ts b/packages/plugin/src/features/magic-context/compartment-storage.ts index b9f97bb83..47dc8e067 100644 --- a/packages/plugin/src/features/magic-context/compartment-storage.ts +++ b/packages/plugin/src/features/magic-context/compartment-storage.ts @@ -1,5 +1,6 @@ import { getHarness } from "../../shared/harness"; import type { Database, Statement as PreparedStatement } from "../../shared/sqlite"; +import { runWriteTransaction } from "../../shared/write-transaction"; import { isCompartmentLeaseHeld } from "./compartment-lease"; import { getIncrementDepthStatement } from "./compression-depth-storage"; import { clearCachedM0M1 } from "./storage-meta-shared"; @@ -379,12 +380,8 @@ export function replaceAllCompartmentStateAndBumpDepth( depthEndOrdinal: number, ): boolean { const now = Date.now(); - db.exec("BEGIN IMMEDIATE"); - let finished = false; - try { + return runWriteTransaction(db, () => { if (!isCompartmentLeaseHeld(db, sessionId, holderId)) { - db.exec("ROLLBACK"); - finished = true; return false; } @@ -402,19 +399,8 @@ export function replaceAllCompartmentStateAndBumpDepth( stmt.run(sessionId, ordinal, getHarness()); } } - - db.exec("COMMIT"); - finished = true; return true; - } finally { - if (!finished) { - try { - db.exec("ROLLBACK"); - } catch { - // Transaction may already be closed by SQLite after an error. - } - } - } + }); } export interface CompartmentDateRanges { @@ -587,19 +573,13 @@ export function promoteRecompStaging( })(); } - db.exec("BEGIN IMMEDIATE"); - let finished = false; - try { + return runWriteTransaction(db, () => { if (!isCompartmentLeaseHeld(db, sessionId, holderId)) { - db.exec("ROLLBACK"); - finished = true; return null; } const staging = getRecompStaging(db, sessionId); if (!staging || staging.compartments.length === 0) { - db.exec("ROLLBACK"); - finished = true; return null; } // Replace real tables @@ -615,18 +595,8 @@ export function promoteRecompStaging( clearCachedM0M1(db, sessionId); - db.exec("COMMIT"); - finished = true; return { compartments: staging.compartments, facts: staging.facts }; - } finally { - if (!finished) { - try { - db.exec("ROLLBACK"); - } catch { - // Transaction may already be closed by SQLite after an error. - } - } - } + }); } /** Clear staging tables for a session (on cancel/abandon or after successful promote). */ diff --git a/packages/plugin/src/features/magic-context/dreamer/lease.ts b/packages/plugin/src/features/magic-context/dreamer/lease.ts index 8aa8aed1f..826637493 100644 --- a/packages/plugin/src/features/magic-context/dreamer/lease.ts +++ b/packages/plugin/src/features/magic-context/dreamer/lease.ts @@ -1,4 +1,5 @@ import type { Database } from "../../../shared/sqlite"; +import { runWriteTransaction } from "../../../shared/write-transaction"; import { deleteDreamState, getDreamState, setDreamState } from "./storage-dream-state"; const LEASE_HOLDER_KEY = "dreaming_lease_holder"; @@ -43,27 +44,9 @@ export function peekLeaseHolderAndExpiry(db: Database, expectedHolder: string): // = false under WAL snapshot isolation and both write — double-acquiring the // lease and spawning duplicate dreamer workers. busy_timeout (set in // initializeDatabase) makes the loser wait rather than throw SQLITE_BUSY. -function runImmediate(db: Database, body: () => T): T { - db.exec("BEGIN IMMEDIATE"); - let committed = false; - try { - const result = body(); - db.exec("COMMIT"); - committed = true; - return result; - } finally { - if (!committed) { - try { - db.exec("ROLLBACK"); - } catch { - // already rolled back / no active transaction - } - } - } -} export function acquireLease(db: Database, holderId: string): boolean { - return runImmediate(db, () => { + return runWriteTransaction(db, () => { if (isLeaseActive(db)) { const existingHolder = getLeaseHolder(db); if (existingHolder && existingHolder !== holderId) { @@ -80,7 +63,7 @@ export function acquireLease(db: Database, holderId: string): boolean { } export function renewLease(db: Database, holderId: string): boolean { - return runImmediate(db, () => { + return runWriteTransaction(db, () => { if (getLeaseHolder(db) !== holderId || !isLeaseActive(db)) { return false; } @@ -93,7 +76,7 @@ export function renewLease(db: Database, holderId: string): boolean { } export function releaseLease(db: Database, holderId: string): void { - runImmediate(db, () => { + runWriteTransaction(db, () => { if (getLeaseHolder(db) !== holderId) { return; } diff --git a/packages/plugin/src/features/magic-context/git-commits/sweep-coordinator.ts b/packages/plugin/src/features/magic-context/git-commits/sweep-coordinator.ts index 7e7bb883a..3e877c4ab 100644 --- a/packages/plugin/src/features/magic-context/git-commits/sweep-coordinator.ts +++ b/packages/plugin/src/features/magic-context/git-commits/sweep-coordinator.ts @@ -1,4 +1,5 @@ import type { Database } from "../../../shared/sqlite"; +import { runWriteTransaction } from "../../../shared/write-transaction"; export const GIT_SWEEP_COOLDOWN_MS = 10 * 60 * 1000; // Commit indexing can include two embedding drains (the indexer drain plus the @@ -59,25 +60,6 @@ export interface AcquireGitSweepLeaseOptions { ignoreCooldown?: boolean; } -function runImmediate(db: Database, body: () => T): T { - db.exec("BEGIN IMMEDIATE"); - let committed = false; - try { - const result = body(); - db.exec("COMMIT"); - committed = true; - return result; - } finally { - if (!committed) { - try { - db.exec("ROLLBACK"); - } catch { - // already rolled back / no active transaction - } - } - } -} - function rowToState(row: GitSweepCoordinatorRow): GitSweepCoordinatorState { return { projectPath: row.project_path, @@ -110,7 +92,7 @@ export function acquireGitSweepLease( const cooldownMs = options.cooldownMs ?? GIT_SWEEP_COOLDOWN_MS; const leaseTtlMs = options.leaseTtlMs ?? GIT_SWEEP_LEASE_TTL_MS; - return runImmediate(db, () => { + return runWriteTransaction(db, () => { const now = Date.now(); const row = getGitSweepCoordinatorState(db, projectPath); if (row?.leaseHolder && row.leaseExpiresAt !== null && row.leaseExpiresAt > now) { @@ -173,7 +155,7 @@ export function renewGitSweepLease( holderId: string, leaseTtlMs = GIT_SWEEP_LEASE_TTL_MS, ): boolean { - return runImmediate(db, () => { + return runWriteTransaction(db, () => { const now = Date.now(); const leaseExpiresAt = now + leaseTtlMs; const result = db @@ -194,7 +176,7 @@ export function markGitSweepSuccessAndRelease( projectPath: string, holderId: string, ): boolean { - return runImmediate(db, () => { + return runWriteTransaction(db, () => { const now = Date.now(); const result = db .prepare( @@ -212,7 +194,7 @@ export function markGitSweepSuccessAndRelease( } export function releaseGitSweepLease(db: Database, projectPath: string, holderId: string): void { - runImmediate(db, () => { + runWriteTransaction(db, () => { db.prepare( `UPDATE git_sweep_coordinator SET lease_holder = NULL, diff --git a/packages/plugin/src/features/magic-context/key-files/identify-key-files.ts b/packages/plugin/src/features/magic-context/key-files/identify-key-files.ts index 1c6bcd144..3d2c76b80 100644 --- a/packages/plugin/src/features/magic-context/key-files/identify-key-files.ts +++ b/packages/plugin/src/features/magic-context/key-files/identify-key-files.ts @@ -10,6 +10,7 @@ import { getHarness } from "../../../shared/harness"; import { shouldKeepSubagents } from "../../../shared/keep-subagents"; import { log } from "../../../shared/logger"; import type { Database } from "../../../shared/sqlite"; +import { runWriteTransaction } from "../../../shared/write-transaction"; import { peekLeaseHolderAndExpiry, renewLease } from "../dreamer/lease"; import { recordChildInvocation } from "../subagent-token-capture"; import { isAftAvailable } from "./aft-availability"; @@ -423,9 +424,7 @@ export function commitKeyFiles(args: { ); const generatedAt = Date.now(); const bump = args.bumpVersion ?? bumpKeyFilesVersion; - args.db.exec("BEGIN IMMEDIATE"); - let committed = false; - try { + return runWriteTransaction(args.db, () => { if (!peekLeaseHolderAndExpiry(args.db, args.leaseHolderId)) { log(`key-files commit aborted: lease lost (holder ${args.leaseHolderId})`); return null; @@ -440,21 +439,11 @@ export function commitKeyFiles(args: { args.configHash, ); const version = bump(args.db, projectPath); - args.db.exec("COMMIT"); - committed = true; log( `key-files committed: ${resolved.length} files, version=${version}, ${resolved.filter((r) => r.staleReason).length} pre-stale`, ); return version; - } finally { - if (!committed) { - try { - args.db.exec("ROLLBACK"); - } catch { - // no active transaction - } - } - } + }); } async function runKeyFilesLlm(args: { diff --git a/packages/plugin/src/features/magic-context/key-files/project-key-files.ts b/packages/plugin/src/features/magic-context/key-files/project-key-files.ts index b9aef2010..7fe5f3078 100644 --- a/packages/plugin/src/features/magic-context/key-files/project-key-files.ts +++ b/packages/plugin/src/features/magic-context/key-files/project-key-files.ts @@ -3,6 +3,7 @@ import { existsSync, readFileSync, realpathSync } from "node:fs"; import { join, resolve, sep } from "node:path"; import { log } from "../../../shared/logger"; import type { Database } from "../../../shared/sqlite"; +import { runWriteTransaction } from "../../../shared/write-transaction"; export type KeyFileStaleReason = "missing" | "content_drift"; @@ -140,9 +141,7 @@ export function replaceProjectKeyFiles( const generatedByModel = files[0]?.generatedByModel ?? null; const configHash = files[0]?.generationConfigHash ?? sha256("{}"); - db.exec("BEGIN IMMEDIATE"); - let committed = false; - try { + return runWriteTransaction(db, () => { db.prepare("DELETE FROM project_key_files WHERE project_path = ?").run(resolvedProjectPath); insertResolvedKeyFiles( db, @@ -152,19 +151,8 @@ export function replaceProjectKeyFiles( generatedByModel, configHash, ); - const version = bumpKeyFilesVersion(db, resolvedProjectPath); - db.exec("COMMIT"); - committed = true; - return version; - } finally { - if (!committed) { - try { - db.exec("ROLLBACK"); - } catch { - // no active transaction - } - } - } + return bumpKeyFilesVersion(db, resolvedProjectPath); + }); } export function insertResolvedKeyFiles( diff --git a/packages/plugin/src/features/magic-context/message-index.ts b/packages/plugin/src/features/magic-context/message-index.ts index 37887773a..966090bec 100644 --- a/packages/plugin/src/features/magic-context/message-index.ts +++ b/packages/plugin/src/features/magic-context/message-index.ts @@ -7,6 +7,7 @@ import type { RawMessage } from "../../hooks/magic-context/read-session-raw"; import { getHarness } from "../../shared/harness"; import type { Database, Statement as PreparedStatement } from "../../shared/sqlite"; import { removeSystemReminders } from "../../shared/system-directive"; +import { runWriteTransaction } from "../../shared/write-transaction"; import { clearCompressionDepth } from "./compression-depth-storage"; interface MessageHistoryIndexRow { @@ -218,9 +219,7 @@ export function indexMessagesAfterOrdinal( // is reflected, so the second skips those ordinals and inserts nothing // duplicate. The bulk SELECT of existing message-ids is still avoided (it // held the writer lock too long on ~30k-row sessions). - db.exec("BEGIN IMMEDIATE"); - let committed = false; - try { + return runWriteTransaction(db, () => { // Re-read under the lock: another process may have advanced the // watermark between the caller's out-of-transaction read and now. const effectiveWatermark = Math.max( @@ -245,18 +244,8 @@ export function indexMessagesAfterOrdinal( // Never regress a higher watermark a concurrent writer may have set. const newWatermark = Math.max(effectiveWatermark, finalWatermark); getUpsertIndexStatement(db).run(sessionId, newWatermark, now, getHarness()); - db.exec("COMMIT"); - committed = true; - } finally { - if (!committed) { - try { - db.exec("ROLLBACK"); - } catch { - // already rolled back / no active transaction - } - } - } - return inserted; + return inserted; + }); } export function ensureMessagesIndexed( diff --git a/packages/plugin/src/features/magic-context/workspaces.ts b/packages/plugin/src/features/magic-context/workspaces.ts index 412ad7d8b..cddf9822a 100644 --- a/packages/plugin/src/features/magic-context/workspaces.ts +++ b/packages/plugin/src/features/magic-context/workspaces.ts @@ -1,5 +1,6 @@ import { createHash } from "node:crypto"; import type { Database } from "../../shared/sqlite"; +import { runWriteTransaction } from "../../shared/write-transaction"; import { V2_MEMORY_CATEGORIES } from "./memory/constants"; import { normalizeStoredProjectPath, storedPathBelongsToIdentity } from "./project-identity"; @@ -271,11 +272,6 @@ export function computeWorkspaceEpochFingerprint( return hash.digest("hex"); } -function isInTransaction(db: Database): boolean { - const candidate = db as unknown as { inTransaction?: unknown; isTransaction?: unknown }; - return candidate.inTransaction === true || candidate.isTransaction === true; -} - function workspaceMembersForIdentity(db: Database, identity: string): string[] { if (!tableExists(db, "workspace_members")) return [identity]; const rows = db @@ -313,22 +309,7 @@ export function bumpEpochsForWorkspaceMembers( now = Date.now(), ): void { const run = () => bumpEpochRows(db, workspaceMembersForIdentity(db, identity), now); - if (isInTransaction(db)) { - run(); - return; - } - db.exec("BEGIN IMMEDIATE"); - try { - run(); - db.exec("COMMIT"); - } catch (error) { - try { - db.exec("ROLLBACK"); - } catch { - // ignore rollback failures from an already-closed transaction - } - throw error; - } + runWriteTransaction(db, run); } export function bumpEpochsForWorkspaceMemberSet( @@ -337,20 +318,5 @@ export function bumpEpochsForWorkspaceMemberSet( now = Date.now(), ): void { const run = () => bumpEpochRows(db, identities, now); - if (isInTransaction(db)) { - run(); - return; - } - db.exec("BEGIN IMMEDIATE"); - try { - run(); - db.exec("COMMIT"); - } catch (error) { - try { - db.exec("ROLLBACK"); - } catch { - // ignore rollback failures from an already-closed transaction - } - throw error; - } + runWriteTransaction(db, run); } diff --git a/packages/plugin/src/hooks/magic-context/compartment-runner-incremental.ts b/packages/plugin/src/hooks/magic-context/compartment-runner-incremental.ts index e3713a34b..c334fbbcc 100644 --- a/packages/plugin/src/hooks/magic-context/compartment-runner-incremental.ts +++ b/packages/plugin/src/hooks/magic-context/compartment-runner-incremental.ts @@ -47,6 +47,7 @@ import { insertUserMemoryCandidates } from "../../features/magic-context/user-me import { normalizeSDKResponse } from "../../shared"; import { describeError } from "../../shared/error-message"; import { sessionLog } from "../../shared/logger"; +import { runWriteTransaction } from "../../shared/write-transaction"; import { updateCompactionMarkerAfterPublication } from "./compaction-marker-manager"; import { buildCompartmentAgentPrompt } from "./compartment-prompt"; import { queueDropsForCompartmentalizedMessages } from "./compartment-runner-drop-queue"; @@ -554,17 +555,9 @@ export async function runCompartmentAgent(deps: CompartmentRunnerDeps): Promise< rollbackDrainReservation(); return; } - let published = false; - db.exec("BEGIN IMMEDIATE"); - try { + const publishOutcome = runWriteTransaction(db, (): "published" | "lease-lost" => { if (!isCompartmentLeaseHeld(db, sessionId, holderId)) { - db.exec("ROLLBACK"); - rollbackDrainReservation(); - sessionLog( - sessionId, - "historian publish skipped: compartment lease no longer held", - ); - return; + return "lease-lost"; } appendCompartments(db, sessionId, persistedCompartments); // v2 faithful fact lifecycle: facts are NOT a REPLACE-the-whole-list @@ -593,16 +586,12 @@ export async function runCompartmentAgent(deps: CompartmentRunnerDeps): Promise< publishedAt: Date.now(), }); } - db.exec("COMMIT"); - published = true; - } finally { - if (!published) { - try { - db.exec("ROLLBACK"); - } catch { - // Transaction may already be closed by an early rollback. - } - } + return "published"; + }); + if (publishOutcome === "lease-lost") { + rollbackDrainReservation(); + sessionLog(sessionId, "historian publish skipped: compartment lease no longer held"); + return; } // Background publication normally preserves the injection cache until // a materializing pass can rebuild history and apply queued drops diff --git a/packages/plugin/src/hooks/magic-context/compartment-runner-recomp.ts b/packages/plugin/src/hooks/magic-context/compartment-runner-recomp.ts index 605869858..ae8ebf67f 100644 --- a/packages/plugin/src/hooks/magic-context/compartment-runner-recomp.ts +++ b/packages/plugin/src/hooks/magic-context/compartment-runner-recomp.ts @@ -25,6 +25,7 @@ import { getErrorMessage } from "../../shared/error-message"; import { getHarness } from "../../shared/harness"; import { sessionLog } from "../../shared/logger"; import type { Database } from "../../shared/sqlite"; +import { runWriteTransaction } from "../../shared/write-transaction"; import { updateCompactionMarkerAfterPublication } from "./compaction-marker-manager"; import { buildCompartmentAgentPrompt } from "./compartment-prompt"; import { queueDropsForCompartmentalizedMessages } from "./compartment-runner-drop-queue"; @@ -90,19 +91,13 @@ export function promoteRecompStagingWithM0Mutation( facts: Array<{ category: string; content: string }>; } | null { const now = Date.now(); - db.exec("BEGIN IMMEDIATE"); - let finished = false; - try { + return runWriteTransaction(db, () => { if (!isCompartmentLeaseHeld(db, sessionId, holderId)) { - db.exec("ROLLBACK"); - finished = true; return null; } const staging = getRecompStaging(db, sessionId); if (!staging || staging.compartments.length === 0) { - db.exec("ROLLBACK"); - finished = true; return null; } @@ -124,18 +119,8 @@ export function promoteRecompStagingWithM0Mutation( db.prepare("DELETE FROM recomp_facts WHERE session_id = ?").run(sessionId); clearCachedM0M1(db, sessionId); - db.exec("COMMIT"); - finished = true; return { compartments: staging.compartments, facts: staging.facts }; - } finally { - if (!finished) { - try { - db.exec("ROLLBACK"); - } catch { - // Transaction may already be closed by SQLite after an error. - } - } - } + }); } export async function executeContextRecompInternal(deps: CompartmentRunnerDeps): Promise { diff --git a/packages/plugin/src/shared/write-transaction.test.ts b/packages/plugin/src/shared/write-transaction.test.ts new file mode 100644 index 000000000..ee3c7f194 --- /dev/null +++ b/packages/plugin/src/shared/write-transaction.test.ts @@ -0,0 +1,150 @@ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; +import { runWriteTransaction, runWriteTransactionAsync } from "./write-transaction.js"; + +/** + * Minimal fake Database that records exec calls and simulates SQLITE_BUSY on + * the first N attempts. Only the surface area write-transaction.ts uses. + */ +function makeFakeDb(opts: { + inTransaction?: boolean; + isTransaction?: boolean; + busyTimes?: number; +}) { + const calls: string[] = []; + let busyRemaining = opts.busyTimes ?? 0; + const db = { + inTransaction: opts.inTransaction, + isTransaction: opts.isTransaction, + exec(sql: string) { + calls.push(sql); + if (sql === "BEGIN IMMEDIATE" && busyRemaining > 0) { + busyRemaining -= 1; + const err: Error & { code?: string } = new Error("database is locked"); + err.code = "SQLITE_BUSY"; + throw err; + } + }, + _calls: calls, + }; + return db; +} + +describe("runWriteTransaction", () => { + beforeEach(() => { + // node:test doesn't have a global mock restore; we rely on per-test fakes. + }); + + it("issues BEGIN IMMEDIATE / COMMIT around the body", () => { + const db = makeFakeDb({}); + const order: string[] = []; + runWriteTransaction(db as any, () => { + order.push("body"); + return 42; + }); + assert.deepEqual(db._calls, ["BEGIN IMMEDIATE", "COMMIT"]); + assert.deepEqual(order, ["body"]); + }); + + it("returns the body's return value", () => { + const db = makeFakeDb({}); + const result = runWriteTransaction(db as any, () => "ok"); + assert.equal(result, "ok"); + }); + + it("rolls back when the body throws", () => { + const db = makeFakeDb({}); + assert.throws( + () => + runWriteTransaction(db as any, () => { + throw new Error("body failed"); + }), + /body failed/, + ); + assert.deepEqual(db._calls, ["BEGIN IMMEDIATE", "ROLLBACK"]); + }); + + it("retries on transient SQLITE_BUSY and succeeds", () => { + const db = makeFakeDb({ busyTimes: 1 }); + const result = runWriteTransaction(db as any, () => "recovered"); + // First attempt: BEGIN IMMEDIATE throws SQLITE_BUSY (no ROLLBACK — the + // BEGIN itself failed, so there's no transaction to roll back). + // Second attempt: BEGIN IMMEDIATE + COMMIT succeed. + assert.deepEqual(db._calls, ["BEGIN IMMEDIATE", "BEGIN IMMEDIATE", "COMMIT"]); + assert.equal(result, "recovered"); + }); + + it("gives up after WRITE_RETRY_MAX_ATTEMPTS transient failures", () => { + const db = makeFakeDb({ busyTimes: 99 }); + assert.throws(() => runWriteTransaction(db as any, () => "never"), /database is locked/); + // 4 attempts, all on BEGIN IMMEDIATE. + assert.deepEqual(db._calls, [ + "BEGIN IMMEDIATE", + "BEGIN IMMEDIATE", + "BEGIN IMMEDIATE", + "BEGIN IMMEDIATE", + ]); + }); + + it("does NOT issue nested BEGIN when already in a transaction", () => { + const db = makeFakeDb({ inTransaction: true }); + const result = runWriteTransaction(db as any, () => "inline"); + assert.deepEqual(db._calls, []); + assert.equal(result, "inline"); + }); + + it("does NOT issue nested BEGIN when node:sqlite isTransaction flag is set", () => { + const db = makeFakeDb({ isTransaction: true }); + const result = runWriteTransaction(db as any, () => "inline"); + assert.deepEqual(db._calls, []); + assert.equal(result, "inline"); + }); + + it("non-transient errors are not retried", () => { + let attempts = 0; + const db = { + exec() { + attempts += 1; + throw new Error("disk I/O error"); + }, + }; + assert.throws(() => runWriteTransaction(db as any, () => "never"), /disk I\/O error/); + assert.equal(attempts, 1); + }); +}); + +describe("runWriteTransactionAsync", () => { + it("issues BEGIN IMMEDIATE / COMMIT around an async body", async () => { + const db = makeFakeDb({}); + const result = await runWriteTransactionAsync(db as any, async () => { + return "async-ok"; + }); + assert.deepEqual(db._calls, ["BEGIN IMMEDIATE", "COMMIT"]); + assert.equal(result, "async-ok"); + }); + + it("rolls back when the async body throws", async () => { + const db = makeFakeDb({}); + await assert.rejects( + runWriteTransactionAsync(db as any, async () => { + throw new Error("async body failed"); + }), + /async body failed/, + ); + assert.deepEqual(db._calls, ["BEGIN IMMEDIATE", "ROLLBACK"]); + }); + + it("retries on transient SQLITE_BUSY", async () => { + const db = makeFakeDb({ busyTimes: 1 }); + const result = await runWriteTransactionAsync(db as any, async () => "recovered"); + assert.deepEqual(db._calls, ["BEGIN IMMEDIATE", "BEGIN IMMEDIATE", "COMMIT"]); + assert.equal(result, "recovered"); + }); + + it("does NOT issue nested BEGIN when already in a transaction", async () => { + const db = makeFakeDb({ inTransaction: true }); + const result = await runWriteTransactionAsync(db as any, async () => "inline"); + assert.deepEqual(db._calls, []); + assert.equal(result, "inline"); + }); +}); diff --git a/packages/plugin/src/shared/write-transaction.ts b/packages/plugin/src/shared/write-transaction.ts new file mode 100644 index 000000000..eee7b63f9 --- /dev/null +++ b/packages/plugin/src/shared/write-transaction.ts @@ -0,0 +1,202 @@ +/** + * Centralized write-transaction helper for the shared context.db. + * + * Why: the plugin issues 100+ write transactions across 40+ files against ONE + * shared SQLite file, opened by multiple processes (OpenCode + Pi, or two + * OpenCode instances). Each `BEGIN IMMEDIATE` contends for the single WAL + * writer lock. Three problems arose: + * + * 1. SQLITE_BUSY propagation: when a sibling process holds the writer lock + * past busy_timeout, the thrown SQLITE_BUSY propagated up and surfaced + * as "failed to load plugin ... database is locked" or "Hit a transient + * issue comparting history this turn". The plugin would disable itself + * for the run instead of waiting. + * + * 2. Duplicated transaction plumbing: every BEGIN IMMEDIATE site + * reimplemented the same try/commit/finally-rollback pattern, slightly + * differently, sometimes without the rollback path. Bugs leaked in at + * the edges. + * + * 3. Inconsistent busy-retry behavior: some sites retried, some didn't, + * some swallowed, some propagated. The result was unpredictable under + * real multi-process load. + * + * Solution: a single `runWriteTransaction(db, body)` helper that wraps the + * BEGIN IMMEDIATE / COMMIT in a bounded SQLITE_BUSY retry loop, so a + * long-running sibling transaction (large migration, dreamer run, bulk + * Channel-2 delivery) makes us wait-and-retry instead of throwing. This + * centralizes the transaction plumbing and the retry policy in one place. + * + * Cross-process fairness is still handled by SQLite's own WAL writer lock + * plus busy_timeout (set in storage-db.ts initializeDatabase); this helper + * just makes every call site a well-behaved writer that absorbs transient + * BUSY errors instead of propagating them into plugin-disable paths. + * + * Usage (replaces hand-rolled BEGIN IMMEDIATE / COMMIT blocks): + * + * const result = runWriteTransaction(db, () => { + * db.prepare("INSERT ...").run(...); + * return computeResult(db); + * }); + * + * For bodies that must do async work between writes, use the async form: + * + * const result = await runWriteTransactionAsync(db, async () => { + * await someAsyncThing(); + * db.prepare("INSERT ...").run(...); + * }); + * + * Composition: if called inside an existing transaction (db.transaction or + * another runWriteTransaction), the body runs inline WITHOUT issuing a nested + * BEGIN (SQLite doesn't allow nested BEGIN; the outer transaction already + * holds the writer lock). This makes the helper safe to compose. + */ + +import { getErrorMessage } from "./error-message"; +import { log } from "./logger"; +import type { Database } from "./sqlite"; + +/** Bounded retry on transient lock errors. */ +const WRITE_RETRY_MAX_ATTEMPTS = 4; +const WRITE_RETRY_BACKOFF_MS = 100; + +function isTransientBusy(error: unknown): boolean { + if (!error || typeof error !== "object") return false; + const e = error as { code?: unknown; message?: unknown }; + if (typeof e.code === "string") { + if ( + e.code === "SQLITE_BUSY" || + e.code === "SQLITE_LOCKED" || + e.code === "SQLITE_BUSY_SNAPSHOT" || + e.code === "SQLITE_BUSY_RECOVERY" + ) { + return true; + } + } + if (typeof e.message === "string") { + return /database is locked/i.test(e.message) || /sqlite_(busy|locked)/i.test(e.message); + } + return false; +} + +function sleepSync(ms: number): void { + // Synchronous SQLite backends can't await; the durations are tiny (100ms + // cap) and only happen on transient cross-process contention. + const end = Date.now() + ms; + while (Date.now() < end) { + // spin + } +} + +function isInTransaction(db: Database): boolean { + // bun:sqlite and better-sqlite3 expose `inTransaction`; the node:sqlite + // shim in sqlite.ts sets `isTransaction` during savepoint-wrapper + // transactions. Respect either so this helper composes correctly when + // called from inside an existing db.transaction(() => { ... })() block. + const candidate = db as unknown as { inTransaction?: unknown; isTransaction?: unknown }; + return candidate.inTransaction === true || candidate.isTransaction === true; +} + +/** + * Run a synchronous write body inside a BEGIN IMMEDIATE transaction, retried + * on transient SQLITE_BUSY. + * + * - If called inside an existing transaction, the body runs inline WITHOUT + * issuing a nested BEGIN (SQLite doesn't allow nested BEGIN; the outer + * transaction already holds the writer lock). + * - Otherwise: issue BEGIN IMMEDIATE, run the body, COMMIT. On a transient + * SQLITE_BUSY the whole sequence is retried up to WRITE_RETRY_MAX_ATTEMPTS + * times with backoff. + */ +export function runWriteTransaction(db: Database, body: () => T): T { + if (isInTransaction(db)) { + return body(); + } + return runWithBusyRetry(() => { + db.exec("BEGIN IMMEDIATE"); + let committed = false; + try { + const result = body(); + db.exec("COMMIT"); + committed = true; + return result; + } finally { + if (!committed) { + try { + db.exec("ROLLBACK"); + } catch { + // transaction may already be closed by SQLite after an error + } + } + } + }); +} + +/** + * Async form of runWriteTransaction. Use this from any async call site; the + * body may itself be async. Same retry and composition semantics. + */ +export async function runWriteTransactionAsync( + db: Database, + body: () => T | Promise, +): Promise { + if (isInTransaction(db)) { + return body(); + } + return runWithBusyRetryAsync(async () => { + db.exec("BEGIN IMMEDIATE"); + let committed = false; + try { + const result = await body(); + db.exec("COMMIT"); + committed = true; + return result; + } finally { + if (!committed) { + try { + db.exec("ROLLBACK"); + } catch { + // already closed + } + } + } + }); +} + +function runWithBusyRetry(fn: () => T): T { + let lastError: unknown; + for (let attempt = 0; attempt < WRITE_RETRY_MAX_ATTEMPTS; attempt += 1) { + try { + return fn(); + } catch (error) { + lastError = error; + if (!isTransientBusy(error) || attempt === WRITE_RETRY_MAX_ATTEMPTS - 1) { + throw error; + } + log( + `[magic-context] write txn attempt ${attempt + 1}/${WRITE_RETRY_MAX_ATTEMPTS} hit transient lock; retrying in ${WRITE_RETRY_BACKOFF_MS}ms: ${getErrorMessage(error)}`, + ); + sleepSync(WRITE_RETRY_BACKOFF_MS); + } + } + throw lastError; +} + +async function runWithBusyRetryAsync(fn: () => T | Promise): Promise { + let lastError: unknown; + for (let attempt = 0; attempt < WRITE_RETRY_MAX_ATTEMPTS; attempt += 1) { + try { + return await fn(); + } catch (error) { + lastError = error; + if (!isTransientBusy(error) || attempt === WRITE_RETRY_MAX_ATTEMPTS - 1) { + throw error; + } + log( + `[magic-context] write txn attempt ${attempt + 1}/${WRITE_RETRY_MAX_ATTEMPTS} hit transient lock; retrying in ${WRITE_RETRY_BACKOFF_MS}ms: ${getErrorMessage(error)}`, + ); + await new Promise((resolve) => setTimeout(resolve, WRITE_RETRY_BACKOFF_MS)); + } + } + throw lastError; +} From 3688e25973fa42f88107384c48b531c2f0bf11e4 Mon Sep 17 00:00:00 2001 From: codeo1io Date: Sat, 20 Jun 2026 05:11:45 +0000 Subject: [PATCH 2/2] fix(dashboard): classify memory.external in ConfigEditor coverage manifest The external-memory-backend branch added the memory.external schema subtree (provider/endpoint/banks/retain_sources/tags + recall sub-block) but did not add a corresponding entry to the dashboard's config-field-coverage manifest. The config-parity guard (config-parity.test.ts) enforces that every schema leaf is either RENDERED by the ConfigEditor form or listed in OMITTED_BY_DESIGN, so CI failed with 16 uncovered leaves. memory.external is USER-config-only and operator-configured (no form widgets exist for it), so the correct classification is OMITTED_BY_DESIGN with the whole subtree covered by the 'memory.external' prefix. --- .../src/components/ConfigEditor/config-field-coverage.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/packages/dashboard/src/components/ConfigEditor/config-field-coverage.ts b/packages/dashboard/src/components/ConfigEditor/config-field-coverage.ts index 8f529079f..d079c2320 100644 --- a/packages/dashboard/src/components/ConfigEditor/config-field-coverage.ts +++ b/packages/dashboard/src/components/ConfigEditor/config-field-coverage.ts @@ -83,6 +83,11 @@ export const OMITTED_BY_DESIGN: Readonly> = { "sidekick.system_prompt": "free-form prompt override; raw JSONC", "system_prompt_injection.skip_signatures": "free-form substring array; raw JSONC (no array widget in the form yet)", + // External memory backend (Hindsight tee + recall). USER config only and + // operator-configured (endpoint/api_key, bank routing, recall tuning); no + // form widgets exist. Whole subtree is raw-JSONC by design. + "memory.external": + "external long-term memory backend (provider/endpoint/banks/retain_sources/tags + recall sub-block); USER config only, operator-configured, raw JSONC", }; /**