diff --git a/packages/core/src/database/sqlite.node.ts b/packages/core/src/database/sqlite.node.ts index 9053a0def663..3fc493e628cc 100644 --- a/packages/core/src/database/sqlite.node.ts +++ b/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 => + 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 = { ...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>) } 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>, ) } 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") })) } }) diff --git a/packages/core/src/database/sqlite.ts b/packages/core/src/database/sqlite.ts index 186274fe3e31..14cb4ccec12a 100644 --- a/packages/core/src/database/sqlite.ts +++ b/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()("@opencode/core/database/SqliteNative") {} @@ -24,6 +24,54 @@ type RunValues = ( params?: ReadonlyArray, ) => Effect.Effect>, 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 = (effect: Effect.Effect) => + effect.pipe( + Effect.retry({ + while: (error) => isSqlError(error) && error.reason._tag === "LockTimeoutError", + schedule: lockRetrySchedule, + }), + ) + export const makeConnection = (run: Run, runValues: RunValues, extensions: Extensions) => identity({ execute(query, params, transformRows) { @@ -47,6 +95,9 @@ export const makeConnection = (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 = (attempt: Effect.Effect): Effect.Effect => + 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 = (attempt: Effect.Effect): Effect.Effect => + Effect.withFiber((fiber) => { + if (Context.getOption(fiber.context, transactionService)._tag === "Some") return attempt + return retryLockTimeout(attempt) + }) + const gateConnection = ( + gate: (attempt: Effect.Effect) => Effect.Effect, + ): 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: [ diff --git a/packages/core/test/sqlite-lock-retry.test.ts b/packages/core/test/sqlite-lock-retry.test.ts new file mode 100644 index 000000000000..d5d18546c21d --- /dev/null +++ b/packages/core/test/sqlite-lock-retry.test.ts @@ -0,0 +1,369 @@ +import { mkdtemp, rm } from "node:fs/promises" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { Database, type SQLQueryBindings } from "bun:sqlite" +import { expect, test } from "bun:test" +import { EffectDrizzleSqlite } from "@opencode/core/database/drizzle" +import { Deferred, Effect, Fiber, Layer } from "effect" +import { Reactivity } from "effect/reactivity" +import { sql } from "drizzle-orm" +import { SqlClient } from "effect/sql" +import { classifySqliteError, SqlError } from "effect/sql/SqlError" +import { Sqlite } from "@opencode/core/database/sqlite" +import { sqliteLayer } from "@opencode/core/database/sqlite.bun" + +// Lock-retry contract of the sqlite client. Foreign write locks are held by a +// second connection and released by the test, never by a timer, and fiber +// order is set with Deferreds. Where a sleep remains it is a wide margin +// ("long enough that everything before it has certainly happened"), not a +// tuned window: effect sleeps never fire early, so a first attempt always +// precedes a release scheduled a second later. + +const withDatabase = (prefix: string, run: (filename: string) => Promise) => + mkdtemp(join(tmpdir(), prefix)).then(async (dir) => { + try { + await run(join(dir, "test.db")) + } finally { + await rm(dir, { recursive: true, force: true }).catch(() => {}) + } + }) + +// Schema plus WAL (a database-level setting) through one throwaway client. +// busy_timeout is per connection, so each test sets it on its own. +const initFile = (filename: string, exec: (native: unknown, text: string) => void, layer = sqliteLayer({ filename })) => + Effect.runPromise( + Effect.gen(function* () { + const native = yield* Sqlite.Native + for (const text of [ + "create table t (id integer primary key, v text)", + "insert into t values (1, 'before')", + "pragma journal_mode = WAL", + ]) { + exec(native, text) + } + }).pipe(Effect.provide(layer), Effect.scoped), + ) + +const readValue = (filename: string) => + Effect.runPromise( + Effect.gen(function* () { + const db = yield* EffectDrizzleSqlite.makeWithDefaults() + const rows = (yield* db.all(sql`select v from t where id = 1`)) as Array<{ v: string }> + return rows[0]?.v + }).pipe(Effect.provide(sqliteLayer({ filename })), Effect.scoped), + ) + +// Fork `write` against a held lock, hold the lock well past its first +// (immediately failing, busy_timeout = 0) attempt so the retry loop is +// entered for sure, then release from the test and let a later attempt land. +const releaseAfterRetries = (write: Effect.Effect, release: () => void) => + Effect.gen(function* () { + const fiber = yield* Effect.forkChild(write) + yield* Effect.sleep("1 second") + release() + yield* Fiber.join(fiber) + }) + +// drizzle wraps the raw SqlError inside EffectDrizzleQueryError.cause.reasons[0].error. +const surfaceReason = (error: unknown) => + ( + (error as { cause?: { reasons?: ReadonlyArray<{ error?: { reason?: unknown } }> } })?.cause?.reasons ?? [] + )[0]?.error?.reason as { _tag?: string; cause?: { code?: string } } | undefined + +// Regression: a statement blocked by another connection's write lock +// (SQLITE_BUSY past busy_timeout) must retry until the lock frees, not fail +// the prompt with LockTimeoutError. +test("retries a statement until another connection releases the write lock", async () => { + await withDatabase("sqlite-retry-", async (filename) => { + await initFile(filename, (native, text) => (native as Database).run(text)) + const lock = new Database(filename) + lock.run("begin immediate") + + await Effect.runPromise( + Effect.gen(function* () { + const db = yield* EffectDrizzleSqlite.makeWithDefaults() + ;((yield* Sqlite.Native) as Database).run("pragma busy_timeout = 0") + yield* releaseAfterRetries(db.run(sql`update t set v = 'after' where id = 1`), () => { + lock.run("commit") + lock.close() + }) + }).pipe(Effect.provide(sqliteLayer({ filename })), Effect.scoped), + ) + expect(await readValue(filename)).toBe("after") + }) +}) + +// Regression: executeValues routes through runValues, a separate wire in the +// driver; dropping the retry there would leave this green-broken. +test("retries on the values path too", async () => { + await withDatabase("sqlite-retry-values-", async (filename) => { + await initFile(filename, (native, text) => (native as Database).run(text)) + const lock = new Database(filename) + lock.run("begin immediate") + + await Effect.runPromise( + Effect.gen(function* () { + const db = yield* EffectDrizzleSqlite.makeWithDefaults() + ;((yield* Sqlite.Native) as Database).run("pragma busy_timeout = 0") + yield* releaseAfterRetries(db.values(sql`update t set v = 'after' where id = 1`), () => { + lock.run("commit") + lock.close() + }) + }).pipe(Effect.provide(sqliteLayer({ filename })), Effect.scoped), + ) + expect(await readValue(filename)).toBe("after") + }) +}) + +// Regression: node:sqlite reports busy as { code: "ERR_SQLITE_ERROR", errcode: +// 5 } — no errno and no SQLITE_* string code, which is all effect's +// classifier reads — so lock timeouts filed as UnknownError and the retry +// never fired on that driver. +test("node driver classifies busy as a lock timeout and retries", async () => { + const { DatabaseSync } = await import("node:sqlite") + const { sqliteLayer: nodeSqliteLayer } = await import("@opencode/core/database/sqlite.node") + type NodeDatabase = InstanceType + + await withDatabase("sqlite-node-retry-", async (filename) => { + await initFile(filename, (native, text) => (native as NodeDatabase).exec(text), nodeSqliteLayer({ filename })) + const lock = new DatabaseSync(filename) + lock.exec("begin immediate") + + await Effect.runPromise( + Effect.gen(function* () { + const db = yield* EffectDrizzleSqlite.makeWithDefaults() + ;((yield* Sqlite.Native) as NodeDatabase).exec("pragma busy_timeout = 0") + yield* releaseAfterRetries(db.run(sql`update t set v = 'after' where id = 1`), () => { + lock.exec("commit") + lock.close() + }) + }).pipe(Effect.provide(nodeSqliteLayer({ filename })), Effect.scoped), + ) + expect(await readValue(filename)).toBe("after") + }) +}) + +// Regression: a DEFERRED WAL transaction that has read cannot upgrade to a +// write after another connection commits — the snapshot is stale and the +// upgrade fails with SQLITE_BUSY_SNAPSHOT. Retrying the statement cannot fix +// a stale snapshot; it must surface at once so the caller can roll back and +// rerun the whole transaction. The foreign write is synchronous between the +// read and the write, so this test has no timing at all. +test("fails fast inside a deferred transaction on a stale snapshot", async () => { + await withDatabase("sqlite-snapshot-", async (filename) => { + await initFile(filename, (native, text) => (native as Database).run(text)) + + const start = Date.now() + let error: unknown + await Effect.runPromise( + Effect.gen(function* () { + const db = yield* EffectDrizzleSqlite.makeWithDefaults() + yield* db.transaction( + (tx) => + Effect.gen(function* () { + yield* tx.get(sql`select v from t where id = 1`) + yield* Effect.sync(() => { + const other = new Database(filename) + other.run("update t set v = 'gone' where id = 1") + other.close() + }) + yield* tx.run(sql`update t set v = 'after' where id = 1`) + }), + { behavior: "deferred" }, + ) + }).pipe(Effect.provide(sqliteLayer({ filename })), Effect.scoped), + ).catch((e) => { + error = e + }) + + expect(error).toBeDefined() + expect(surfaceReason(error)?._tag).toBe("LockTimeoutError") + expect(surfaceReason(error)?.cause?.code).toBe("SQLITE_BUSY_SNAPSHOT") + expect(Date.now() - start).toBeLessThan(2000) + }) +}) + +// Regression: only lock timeouts retry. A constraint violation must surface +// immediately; a widened predicate would show up here as slowness. +test("fails fast on a non-lock error", async () => { + await withDatabase("sqlite-constraint-", async (filename) => { + await initFile(filename, (native, text) => (native as Database).run(text)) + + const start = Date.now() + let error: unknown + await Effect.runPromise( + Effect.gen(function* () { + const db = yield* EffectDrizzleSqlite.makeWithDefaults() + yield* db.run(sql`insert into t values (1, 'dup')`) + }).pipe(Effect.provide(sqliteLayer({ filename })), Effect.scoped), + ).catch((e) => { + error = e + }) + + expect(error).toBeDefined() + expect(surfaceReason(error)?._tag).not.toBe("LockTimeoutError") + expect(Date.now() - start).toBeLessThan(1000) + }) +}) + +// Regression: while a statement sleeps between lock retries, another fiber +// can open a transaction on the shared connection. An attempt that resumes +// without re-entering the client semaphore runs inside that transaction and +// its writes roll back with it; the fix makes each attempt wait the +// transaction out. The Deferred parks the transaction mid-body so the +// interleaving is ordered by the test, not by sleeps lining up. +test("a retrying statement waits out a transaction opened while it sleeps", async () => { + await withDatabase("sqlite-race-", async (filename) => { + await initFile(filename, (native, text) => (native as Database).run(text)) + const lock = new Database(filename) + lock.run("begin immediate") + + await Effect.runPromise( + Effect.gen(function* () { + const db = yield* EffectDrizzleSqlite.makeWithDefaults() + ;((yield* Sqlite.Native) as Database).run("pragma busy_timeout = 0") + + // A: a plain autocommit write. Its first attempts fail on the foreign + // lock; by one second in, it is sleeping between retries. + const writer = yield* Effect.forkChild(db.run(sql`update t set v = 'after' where id = 1`)) + yield* Effect.sleep("1 second") + + // B: open a transaction on the same client and park inside it on a + // Deferred. It holds the client semaphore until the test lets go. + const inside = yield* Deferred.make() + const transactor = yield* Effect.forkChild( + db + .transaction( + (tx) => + Effect.gen(function* () { + yield* tx.get(sql`select v from t where id = 1`) + yield* Deferred.await(inside) + yield* Effect.fail(new Error("rollback requested")) + }), + { behavior: "deferred" }, + ) + .pipe(Effect.ignore), + ) + yield* Effect.sleep("1 second") + + // Free the write lock while B is parked. A's next attempt must now + // wait for B: if it resumed without the semaphore it would join B's + // transaction, succeed, and be rolled back below. + lock.run("commit") + lock.close() + yield* Effect.sleep("500 millis") + yield* Deferred.succeed(inside, undefined) + yield* Fiber.join(transactor) + yield* Fiber.join(writer) + }).pipe(Effect.provide(sqliteLayer({ filename })), Effect.scoped), + ) + + // B rolled back, so the update only survives if A waited B out and landed + // in autocommit. + expect(await readValue(filename)).toBe("after") + }) +}) + +// A driver like the bun one except that the client's first statement +// suspends until `midflight` is done — parking the fiber that reaches a +// statement exactly in the gap between effect-sql's acquirer returning and +// the statement running, so the test below can order that gap against +// another fiber's BEGIN deterministically. +const yieldingSqliteLayer = (filename: string, midflight: Deferred.Deferred) => + Layer.effect( + SqlClient.SqlClient, + Effect.gen(function* () { + const native = new Database(filename) + yield* Effect.addFinalizer(() => Effect.sync(() => native.close())) + let parked = false + const firstStatementSuspends = (self: Effect.Effect): Effect.Effect => + Effect.suspend(() => { + if (parked) return self + parked = true + return Effect.andThen(Deferred.await(midflight), () => self) + }) + const run = (query: string, params: ReadonlyArray = []) => + firstStatementSuspends( + Effect.try({ + try: () => + native.query, SQLQueryBindings[]>(query).all( + ...(params as SQLQueryBindings[]), + ) ?? [], + catch: (cause) => + new SqlError({ + reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }), + }), + }), + ) + const runValues = (query: string, params: ReadonlyArray = []) => + firstStatementSuspends( + Effect.try({ + try: () => + native.query(query).values(...(params as SQLQueryBindings[])) ?? [], + catch: (cause) => + new SqlError({ + reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }), + }), + }), + ) + return yield* Sqlite.makeClient({}, Sqlite.makeConnection(run, runValues, {}), "~test/SqliteYielding", () => ({})) + }), + ).pipe(Layer.provide(Reactivity.layer)) + +// Regression: the permit has to cover the statement itself, not just handing +// out the connection. effect-sql runs a plain statement after its acquirer +// has already returned, so with the semaphore free in that gap a transaction +// opening there swallows the statement — its write joins the transaction and +// rolls back with it. The driver parks the write exactly in that gap, so the +// interleaving is ordered by the test, not by timing. +test("a plain statement does not join a transaction that opens while it runs", async () => { + await withDatabase("sqlite-midflight-", async (filename) => { + await initFile(filename, (native, text) => (native as Database).run(text)) + + await Effect.runPromise( + Deferred.make().pipe( + Effect.flatMap((midflight) => + Effect.gen(function* () { + const db = yield* EffectDrizzleSqlite.makeWithDefaults() + + // A: a plain autocommit write, parked mid-statement — past its + // acquirer, before the statement executes. + const writer = yield* Effect.forkChild(db.run(sql`update t set v = 'after' where id = 1`)) + yield* Effect.sleep("1 second") + + // B: open a transaction while A is parked and hold it on `inside`. + // With the permit released early, B opens here and A's write lands + // inside it; B's rollback below would erase the write. + const inside = yield* Deferred.make() + const transactor = yield* Effect.forkChild( + db + .transaction( + (tx) => + Effect.gen(function* () { + yield* tx.get(sql`select v from t where id = 1`) + yield* Deferred.await(inside) + yield* Effect.fail(new Error("rollback requested")) + }), + { behavior: "deferred" }, + ) + .pipe(Effect.ignore), + ) + yield* Effect.sleep("1 second") + + // Let A execute, then let B finish. A's write only survives if it + // ran in autocommit. + yield* Deferred.succeed(midflight, undefined) + yield* Effect.sleep("1 second") + yield* Deferred.succeed(inside, undefined) + yield* Fiber.join(transactor) + yield* Fiber.join(writer) + }).pipe(Effect.provide(yieldingSqliteLayer(filename, midflight))), + ), + Effect.scoped, + ), + ) + + // B rolled back, so the update only survives if A ran in autocommit. + expect(await readValue(filename)).toBe("after") + }) +})