fix(core): retry sqlite statements on lock timeout

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

@@ -3,6 +3,7 @@ import { Context, Effect, Layer } from "effect"
 import { Reactivity } from "effect/reactivity"
 import { SqlClient } from "effect/sql"
 import { classifySqliteError, SqlError } from "effect/sql/SqlError"
+import { isRecord } from "@opencode/util/record"
 import { Sqlite } from "./sqlite.js"
 
 const TypeId = "~@opencode/core/database/SqliteNode" as const
@@ -22,6 +23,28 @@ interface Config extends Sqlite.ClientConfig {
   readonly allowExtension?: boolean
 }
 
+const isNodeSqliteError = (cause: unknown): cause is Record<string, unknown> =>
+  isRecord(cause) && typeof cause["errcode"] === "number"
+
+// node:sqlite reports SQLITE_BUSY and its kin as
+// { code: "ERR_SQLITE_ERROR", errcode: 5, errstr: "database is locked" } — no
+// errno and no SQLITE_* string code, and those are all classifySqliteError
+// reads, so lock timeouts file as UnknownError and the retry wrapper never
+// fires. node's errcode is the sqlite primary/extended code; copying it onto
+// errno gives the classifier the shape it knows (5 = SQLITE_BUSY,
+// 6 = SQLITE_LOCKED, masked with & 0xff there).
+const classifyNodeSqliteError = (cause: unknown, operation: string) => {
+  if (isNodeSqliteError(cause)) {
+    const normalized: Record<string, unknown> = { ...cause, errno: cause["errcode"] }
+    if (typeof cause["message"] === "string") normalized["message"] = cause["message"]
+    return classifySqliteError(normalized as unknown, {
+      message: "Failed to execute statement",
+      operation,
+    })
+  }
+  return classifySqliteError(cause, { message: "Failed to execute statement", operation })
+}
+
 const make = (options: Config) =>
   Effect.gen(function* () {
     const native = (yield* Sqlite.Native) as DatabaseSync
@@ -33,11 +56,7 @@ const make = (options: Config) =>
         try {
           return Effect.succeed(statement.all(...(params as SQLInputValue[])) as Array<Record<string, unknown>>)
         } catch (cause) {
-          return Effect.fail(
-            new SqlError({
-              reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }),
-            }),
-          )
+          return Effect.fail(new SqlError({ reason: classifyNodeSqliteError(cause, "execute") }))
         }
       })
 
@@ -51,11 +70,7 @@ const make = (options: Config) =>
             statement.all(...(params as SQLInputValue[])) as unknown as ReadonlyArray<ReadonlyArray<unknown>>,
           )
         } catch (cause) {
-          return Effect.fail(
-            new SqlError({
-              reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }),
-            }),
-          )
+          return Effect.fail(new SqlError({ reason: classifyNodeSqliteError(cause, "execute") }))
         }
       })
 

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, Effect, Fiber, Schedule, Scope, Semaphore, Stream } from "effect"
 import { identity } from "effect/Function"
 import { SqlClient, Statement } from "effect/sql"
 import type { Connection } from "effect/sql/SqlConnection"
-import type { SqlError } from "effect/sql/SqlError"
+import { isSqlError, type SqlError } from "effect/sql/SqlError"
 
 export class Native extends Context.Service<Native, unknown>()("@opencode/core/database/SqliteNative") {}
 
@@ -24,6 +24,54 @@ type RunValues = (
   params?: ReadonlyArray<unknown>,
 ) => Effect.Effect<ReadonlyArray<ReadonlyArray<unknown>>, SqlError>
 
+// SQLITE_BUSY means another process held the write lock past busy_timeout. The
+// statement itself is fine — retrying with an async sleep lets the competing
+// process finish so the next attempt can acquire the lock. Without this, one
+// long write in a concurrent process kills another process's prompt.
+//
+// The sleep runs between attempts, and while it sleeps another fiber can open
+// a transaction on the shared connection; an attempt that only repeated the
+// statement would then resume inside that foreign transaction, because the
+// connection's own in-transaction flag cannot tell whose transaction it is
+// reading. Each plain attempt therefore re-enters the client semaphore before
+// executing: an open transaction holds the semaphore for its whole life, so
+// the next attempt waits it out. Statements of our own transaction carry its
+// service in the fiber context and fail fast instead — a BUSY_SNAPSHOT there
+// repeats until the transaction is rolled back and restarted, so retrying
+// would only burn the budget. Rolling back and rerunning the whole
+// transaction is the only recovery there, and only the caller can do it: the
+// body reads and branches between its statements, so whether a rerun is safe
+// is the caller's call. The durable write paths (Bus, session lifecycle) get
+// that caller-side replay from #53704's StorageRetry; this client stays at
+// statement scope so the two layers never retry the same write — a
+// transaction's body statements never retry here, and its BEGIN, COMMIT and
+// ROLLBACK retry in place while it holds the semaphore, so a caller-side
+// replay only starts after this client's budget is spent.
+//
+// The budget and backoff curve are the policy in one place:
+//   50ms doubling to 250ms, jittered, giving up after LOCK_BUDGET_MS total.
+// The budget bounds how long one statement holds up the client; a caller with
+// N serial statements can spend N times it before surfacing.
+const LOCK_RETRY_INITIAL_MS = 50
+const LOCK_RETRY_FACTOR = 2
+const LOCK_RETRY_MAX_MS = 250
+const LOCK_RETRY_BUDGET_MS = 30_000
+
+const lockRetrySchedule = Schedule.min([
+  Schedule.exponential(LOCK_RETRY_INITIAL_MS, LOCK_RETRY_FACTOR),
+  Schedule.spaced(LOCK_RETRY_MAX_MS),
+])
+  .pipe(Schedule.jittered)
+  .pipe(Schedule.upTo({ duration: LOCK_RETRY_BUDGET_MS }))
+
+const retryLockTimeout = <A, E>(effect: Effect.Effect<A, E>) =>
+  effect.pipe(
+    Effect.retry({
+      while: (error) => isSqlError(error) && error.reason._tag === "LockTimeoutError",
+      schedule: lockRetrySchedule,
+    }),
+  )
+
 export const makeConnection = <Extensions extends object>(run: Run, runValues: RunValues, extensions: Extensions) =>
   identity<Connection & Extensions>({
     execute(query, params, transformRows) {
@@ -47,6 +95,9 @@ export const makeConnection = <Extensions extends object>(run: Run, runValues: R
     ...extensions,
   })
 
+// One key per client, so a gate never mistakes another client's transaction for its own.
+let transactionCounter = 0
+
 export const makeClient = <
   Config extends ClientConfig,
   SqliteConnection extends Connection,
@@ -60,13 +111,54 @@ export const makeClient = <
 ) =>
   Effect.gen(function* () {
     const semaphore = yield* Semaphore.make(1)
-    const acquirer = semaphore.withPermits(1)(Effect.succeed(connection))
+    const transactionService = Context.Service<
+      SqlClient.TransactionConnection,
+      SqlClient.TransactionConnection.Service
+    >(`@opencode/core/database/SqliteTransaction/${transactionCounter++}`)
+    // Body statements of our own transaction carry its service and run while
+    // the transaction holds the semaphore, so both gates pass them through
+    // untouched — a BUSY_SNAPSHOT there repeats until rollback and retrying
+    // would only burn budget (see the block comment above).
+    const gateStatement = <A>(attempt: Effect.Effect<A, SqlError>): Effect.Effect<A, SqlError> =>
+      Effect.withFiber((fiber) => {
+        if (Context.getOption(fiber.context, transactionService)._tag === "Some") return attempt
+        // The permit covers the attempt, not just handing out the connection:
+        // effect-sql runs the statement after its acquirer has already
+        // returned, so a permit dropped there would let a transaction opened
+        // in that gap swallow the attempt. A transaction holds the semaphore
+        // for its whole life, so this take waits out any that opened while a
+        // previous attempt slept.
+        return retryLockTimeout(semaphore.withPermits(1)(attempt))
+      })
+    // Service-less statements on the transaction connection are exactly its
+    // BEGIN, COMMIT and ROLLBACK: effect-sql runs BEGIN after the acquirer
+    // returns and COMMIT/ROLLBACK after the body's service is withdrawn, all
+    // before the transaction's scope releases the semaphore (nested
+    // savepoints run inside the body, so they arrive above with the service).
+    // The transaction already holds the semaphore, so re-taking it would
+    // deadlock; they retry in place.
+    const gateTransactionStatement = <A>(attempt: Effect.Effect<A, SqlError>): Effect.Effect<A, SqlError> =>
+      Effect.withFiber((fiber) => {
+        if (Context.getOption(fiber.context, transactionService)._tag === "Some") return attempt
+        return retryLockTimeout(attempt)
+      })
+    const gateConnection = (
+      gate: <A>(attempt: Effect.Effect<A, SqlError>) => Effect.Effect<A, SqlError>,
+    ): SqliteConnection => ({
+      ...connection,
+      execute: (query, params, transformRows) => gate(connection.execute(query, params, transformRows)),
+      executeRaw: (query, params) => gate(connection.executeRaw(query, params)),
+      executeValues: (query, params) => gate(connection.executeValues(query, params)),
+      executeUnprepared: (query, params, transformRows) =>
+        gate(connection.executeUnprepared(query, params, transformRows)),
+    })
+    const acquirer = semaphore.withPermits(1)(Effect.succeed(gateConnection(gateStatement)))
     const transactionAcquirer = Effect.uninterruptibleMask((restore) => {
       const fiber = Fiber.getCurrent()!
       const scope = Context.getUnsafe(fiber.context, Scope.Scope)
       return Effect.as(
         Effect.tap(restore(semaphore.take(1)), () => Scope.addFinalizer(scope, semaphore.release(1))),
-        connection,
+        gateConnection(gateTransactionStatement),
       )
     })
     const transformRows = options.transformResultNames
@@ -76,6 +168,7 @@ export const makeClient = <
     return Object.assign(
       yield* SqlClient.make({
         acquirer,
+        transactionService,
         compiler: Statement.makeCompilerSqlite(options.transformQueryNames),
         transactionAcquirer,
         spanAttributes: [