fix(core): retry locked SQLite statements cooperatively

packages/core/src/database/database.ts

@@ -1,4 +1,4 @@
-export * as Database from "./database.js"
+export * as Database from "./database.js"
 
 import { EffectDrizzleSqlite } from "./drizzle.js"
 import { sqliteLayer, supportsForeignKeyToggle, supportsTuningPragmas } from "#sqlite"
@@ -31,7 +31,7 @@ const databaseLayer = Layer.effect(
     if (supportsTuningPragmas) {
       yield* db.run("PRAGMA journal_mode = WAL")
       yield* db.run("PRAGMA synchronous = NORMAL")
-      yield* db.run("PRAGMA busy_timeout = 5000")
+      yield* db.run("PRAGMA busy_timeout = 0")
       yield* db.run("PRAGMA cache_size = -64000")
       yield* db.run("PRAGMA wal_checkpoint(PASSIVE)")
     }
@@ -75,3 +75,4 @@ export function configuredClient(client: Layer.Layer<SqlClient.SqlClient>) {
 }
 
 export const node = configured({ path: ":memory:" })
+

packages/core/src/database/sqlite.node.ts

@@ -81,7 +81,9 @@ const nativeLayer = (config: Config) =>
     Effect.gen(function* () {
       const native = new DatabaseSync(config.filename, {
         readOnly: config.readonly,
-        timeout: config.timeout,
+        // Node's native busy wait would block the event loop; locked statements
+        // fail immediately and retry cooperatively in Sqlite.makeConnection.
+        timeout: config.timeout ?? 0,
         allowExtension: config.allowExtension,
         enableForeignKeyConstraints: true,
         open: true,

packages/core/src/database/sqlite.ts

@@ -1,10 +1,10 @@
 export * as Sqlite from "./sqlite.js"
 
-import { Context, Effect, Fiber, Scope, Semaphore, Stream } from "effect"
+import { Context, Duration, Effect, Fiber, Schedule, Scope, Semaphore, Stream } from "effect"
 import { identity } from "effect/Function"
 import { SqlClient, Statement } from "effect/unstable/sql"
 import type { Connection } from "effect/unstable/sql/SqlConnection"
-import type { SqlError } from "effect/unstable/sql/SqlError"
+import { SqlError } from "effect/unstable/sql/SqlError"
 
 export class Native extends Context.Service<Native, unknown>()("@opencode-ai/core/database/SqliteNative") {}
 
@@ -24,19 +24,36 @@ type RunValues = (
   params?: ReadonlyArray<unknown>,
 ) => Effect.Effect<ReadonlyArray<ReadonlyArray<unknown>>, SqlError>
 
+// SQLITE_BUSY/SQLITE_LOCKED mean the statement never executed, so retrying it
+// is side-effect free. Native busy waiting (busy_timeout) blocks the whole
+// event loop, so locked statements retry cooperatively with a bounded,
+// jittered exponential schedule instead (roughly 25ms..800ms per attempt).
+const lockedStatementSchedule = Schedule.exponential(Duration.millis(25)).pipe(
+  Schedule.jittered,
+  Schedule.upTo({ times: 6 }),
+)
+
+const retryLocked = <A>(effect: Effect.Effect<A, SqlError>) =>
+  effect.pipe(
+    Effect.retry({
+      schedule: lockedStatementSchedule,
+      while: (error) => error.reason._tag === "LockTimeoutError",
+    }),
+  )
+
 export const makeConnection = <Extensions extends object>(run: Run, runValues: RunValues, extensions: Extensions) =>
   identity<Connection & Extensions>({
     execute(query, params, transformRows) {
-      return transformRows ? Effect.map(run(query, params), transformRows) : run(query, params)
+      return transformRows ? Effect.map(retryLocked(run(query, params)), transformRows) : retryLocked(run(query, params))
     },
     executeRaw(query, params) {
-      return run(query, params)
+      return retryLocked(run(query, params))
     },
     executeValues(query, params) {
-      return runValues(query, params)
+      return retryLocked(runValues(query, params))
     },
     executeValuesUnprepared(query, params) {
-      return runValues(query, params)
+      return retryLocked(runValues(query, params))
     },
     executeUnprepared(query, params, transformRows) {
       return this.execute(query, params, transformRows)
@@ -95,3 +112,4 @@ export const makeClient = <
         readonly updateValues: never
       } & Extensions
   })
+

packages/core/test/database-drizzle.test.ts

@@ -4,10 +4,11 @@ import { join } from "node:path"
 import { Database } from "bun:sqlite"
 import { expect, test } from "bun:test"
 import { SqliteClient } from "@effect/sql-sqlite-bun"
+import { sqliteLayer } from "@opencode-ai/core/database/sqlite.bun"
 import { eq, sql } from "drizzle-orm"
 import { integer, sqliteTable, text } from "drizzle-orm/sqlite-core"
-import { Effect } from "effect"
-import type { SqlClient as SqlClientService } from "effect/unstable/sql/SqlClient"
+import { Deferred, Effect, Fiber } from "effect"
+import { SqlClient as SqlClientService } from "effect/unstable/sql/SqlClient"
 import { isSqlError } from "effect/unstable/sql/SqlError"
 import { EffectDrizzleSqlite } from "@opencode-ai/core/database/drizzle"
 
@@ -130,6 +131,106 @@ test("preserves failed transaction begin errors", async () => {
   }
 })
 
+test("retries locked writes cooperatively once the holding transaction releases", async () => {
+  const dir = await mkdtemp(join(tmpdir(), "effect-drizzle-sqlite-"))
+  const filename = join(dir, "locked.db")
+  const holder = new Database(filename)
+
+  try {
+    holder.run("create table users (id integer primary key autoincrement, name text not null)")
+    holder.run("pragma busy_timeout = 0")
+    holder.run("begin immediate")
+
+    await Effect.runPromise(
+      Effect.gen(function* () {
+        const client = yield* SqlClientService
+        const done = yield* Deferred.make<void>()
+        const insert = yield* client
+          .unsafe("insert into users (name) values (?)", ["Ada"])
+          .values.pipe(Effect.ensuring(Deferred.succeed(done, undefined)), Effect.forkChild)
+        yield* Effect.yieldNow
+
+        // The write is blocked by the holder's write lock and must retry
+        // cooperatively instead of native busy waiting.
+        expect(yield* Deferred.isDone(done)).toBe(false)
+
+        holder.run("rollback")
+        yield* Fiber.join(insert)
+        expect(yield* client.unsafe("select * from users").values).toEqual([[1, "Ada"]])
+      }).pipe(Effect.provide(sqliteLayer({ filename, disableWAL: true })), Effect.scoped),
+    )
+  } finally {
+    if (holder.inTransaction) holder.run("rollback")
+    holder.close()
+    await rm(dir, { recursive: true, force: true })
+  }
+})
+
+test("retries locked transaction begins cooperatively once the holding transaction releases", async () => {
+  const dir = await mkdtemp(join(tmpdir(), "effect-drizzle-sqlite-"))
+  const filename = join(dir, "locked.db")
+  const holder = new Database(filename)
+
+  try {
+    holder.run("create table users (id integer primary key autoincrement, name text not null)")
+    holder.run("pragma busy_timeout = 0")
+    holder.run("begin immediate")
+
+    await Effect.runPromise(
+      Effect.gen(function* () {
+        const db = yield* EffectDrizzleSqlite.makeWithDefaults()
+        yield* db.run(sql`pragma busy_timeout = 0`)
+
+        const done = yield* Deferred.make<void>()
+        const transaction = yield* db
+          .transaction((tx) => tx.insert(users).values({ name: "Ada" }), { behavior: "immediate" })
+          .pipe(Effect.ensuring(Deferred.succeed(done, undefined)), Effect.forkChild)
+        yield* Effect.yieldNow
+
+        // The BEGIN IMMEDIATE fails fast and the statement-level retry re-runs
+        // it until the holder releases the write lock.
+        expect(yield* Deferred.isDone(done)).toBe(false)
+
+        holder.run("rollback")
+        yield* Fiber.join(transaction)
+        expect(yield* db.select().from(users)).toEqual([{ id: 1, name: "Ada" }])
+      }).pipe(Effect.provide(sqliteLayer({ filename, disableWAL: true })), Effect.scoped),
+    )
+  } finally {
+    if (holder.inTransaction) holder.run("rollback")
+    holder.close()
+    await rm(dir, { recursive: true, force: true })
+  }
+})
+
+test("reports LockTimeoutError once the bounded statement retry is exhausted", async () => {
+  const dir = await mkdtemp(join(tmpdir(), "effect-drizzle-sqlite-"))
+  const filename = join(dir, "locked.db")
+  const holder = new Database(filename)
+
+  try {
+    holder.run("create table users (id integer primary key autoincrement, name text not null)")
+    holder.run("pragma busy_timeout = 0")
+    holder.run("begin immediate")
+
+    await Effect.runPromise(
+      Effect.gen(function* () {
+        const client = yield* SqlClientService
+        const error = yield* client
+          .unsafe("insert into users (name) values (?)", ["Blocked"])
+          .values.pipe(Effect.flip)
+
+        if (!isSqlError(error)) throw new Error("Expected SqlError")
+        expect(error.reason._tag).toBe("LockTimeoutError")
+      }).pipe(Effect.provide(sqliteLayer({ filename, disableWAL: true })), Effect.scoped),
+    )
+  } finally {
+    if (holder.inTransaction) holder.run("rollback")
+    holder.close()
+    await rm(dir, { recursive: true, force: true })
+  }
+})
+
 test("supports returning and rejects empty update sets", async () => {
   await run(
     Effect.gen(function* () {

packages/core/test/session-runner.test.ts

@@ -1,4 +1,4 @@
-import { describe, expect, test } from "bun:test"
+import { describe, expect, test } from "bun:test"
 import {
   AIError,
   LLMEvent,
@@ -3820,6 +3820,98 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
+    it.effect("waits for unrelated database transactions before interrupted settlement", () =>
+    Effect.gen(function* () {
+      const session = yield* setup
+      const { db } = yield* Database.Service
+      const transactionStarted = yield* Deferred.make<void>()
+      const releaseTransaction = yield* Deferred.make<void>()
+      const interruptSettled = yield* Deferred.make<void>()
+      yield* admit(session, "Interrupt during database contention")
+      const stream = yield* TestLLM.gate
+
+      const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
+      yield* stream.started
+      const transaction = yield* db
+        .transaction(() =>
+          Deferred.succeed(transactionStarted, undefined).pipe(Effect.andThen(Deferred.await(releaseTransaction))),
+        )
+        .pipe(Effect.forkChild)
+      yield* Deferred.await(transactionStarted)
+      const interrupt = yield* session
+        .interrupt(sessionID)
+        .pipe(Effect.ensuring(Deferred.succeed(interruptSettled, undefined)), Effect.forkChild)
+      yield* Effect.yieldNow
+
+      expect(yield* Deferred.isDone(interruptSettled)).toBe(false)
+
+      yield* Deferred.succeed(releaseTransaction, undefined)
+      yield* Fiber.join(transaction)
+      yield* Fiber.join(interrupt)
+      expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" })
+      expect(requireAssistant(yield* session.context(sessionID))).toMatchObject({
+        finish: "error",
+        error: { type: "aborted", message: "Step interrupted" },
+      })
+    }),
+  )
+
+  it.effect("queues interrupted settlement behind an unrelated transaction", () =>
+    Effect.gen(function* () {
+      const session = yield* setup
+      const { db } = yield* Database.Service
+      const transactionStarted = yield* Deferred.make<void>()
+      const interruptSettled = yield* Deferred.make<void>()
+      yield* admit(session, "Interrupt during long transaction")
+      const stream = yield* TestLLM.gate
+
+      const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
+      yield* stream.started
+      // An unrelated session's work holds the shared SQLite transaction permit for a
+      // deterministic amount of work; the interrupted step's terminal publications
+      // queue behind it instead of settling immediately.
+      const transaction = yield* db
+        .transaction(() =>
+          Effect.gen(function* () {
+            yield* Deferred.succeed(transactionStarted, undefined)
+            for (let index = 0; index < 3000; index++) {
+              const id = Session.ID.make(`ses_contention_${index}`)
+              yield* db
+                .insert(SessionTable)
+                .values({
+                  id,
+                  project_id: Project.ID.global,
+                  slug: id,
+                  directory: "/project",
+                  title: "test",
+                  version: "test",
+                })
+                .onConflictDoNothing()
+                .run()
+                .pipe(Effect.orDie)
+            }
+          }),
+        )
+        .pipe(Effect.forkChild)
+      yield* Deferred.await(transactionStarted)
+      const interrupt = yield* session
+        .interrupt(sessionID)
+        .pipe(Effect.ensuring(Deferred.succeed(interruptSettled, undefined)), Effect.forkChild)
+      yield* Effect.yieldNow
+
+      expect(yield* Deferred.isDone(interruptSettled)).toBe(false)
+
+      yield* Fiber.join(transaction)
+      yield* Fiber.join(interrupt)
+      expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" })
+      expect(requireAssistant(yield* session.context(sessionID))).toMatchObject({
+        finish: "error",
+        error: { type: "aborted", message: "Step interrupted" },
+      })
+    }),
+  )
+
+
   it.effect("durably fails blocked local tools when interrupted while awaiting settlement", () =>
     Effect.gen(function* () {
       const session = yield* setup
@@ -5111,3 +5203,8 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 })
+
+
+
+
+