From 29c1cf63d3408b087ae997226e41e7b8733a05c3 Mon Sep 17 00:00:00 2001 From: Lance Woodson Date: Sat, 10 Oct 2026 16:28:12 -0500 Subject: [PATCH 1/5] fix(core): retry sqlite statements on lock timeout When multiple opencode processes share one database, statements that cannot grab the SQLite write lock within busy_timeout fail with database is locked and kill the running prompt. The statement only needs the lock, so retry autocommit statements classified as LockTimeoutError (SQLITE_BUSY/SQLITE_LOCKED) with jittered exponential backoff (50 ms doubling to a 250 ms cap, 30 s total) in the bun and node sqlite drivers. The async sleep between attempts lets the competing process finish its write. Only autocommit statements are retried. Inside a transaction the same error repeats until the transaction is rolled back and restarted (SQLITE_BUSY_SNAPSHOT on a deferred transaction whose read snapshot went stale), so it surfaces at once for the caller to handle. Ports #47567 to the V2 tree. The in-transaction fail-fast and its test originate from iceteaSA's work on that PR. --- packages/core/src/database/sqlite.bun.ts | 7 +- packages/core/src/database/sqlite.node.ts | 7 +- packages/core/src/database/sqlite.ts | 24 +++- packages/core/test/sqlite-lock-retry.test.ts | 121 +++++++++++++++++++ 4 files changed, 151 insertions(+), 8 deletions(-) create mode 100644 packages/core/test/sqlite-lock-retry.test.ts diff --git a/packages/core/src/database/sqlite.bun.ts b/packages/core/src/database/sqlite.bun.ts index 7728bc8f4e95..920ce372c05e 100644 --- a/packages/core/src/database/sqlite.bun.ts +++ b/packages/core/src/database/sqlite.bun.ts @@ -3,7 +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 { Sqlite } from "./sqlite.js" +import { retryLocked, Sqlite } from "./sqlite.js" const TypeId = "~@opencode/core/database/SqliteBun" as const @@ -23,6 +23,7 @@ interface Config extends Sqlite.ClientConfig { const make = (options: Config) => Effect.gen(function* () { const native = (yield* Sqlite.Native) as Database + const retryIfAutocommit = retryLocked(() => native.inTransaction) const run = (query: string, params: ReadonlyArray = []) => Effect.withFiber>, SqlError>((fiber) => { @@ -38,7 +39,7 @@ const make = (options: Config) => }), ) } - }) + }).pipe(retryIfAutocommit) const runValues = (query: string, params: ReadonlyArray = []) => Effect.withFiber, SqlError>((fiber) => { @@ -54,7 +55,7 @@ const make = (options: Config) => }), ) } - }) + }).pipe(retryIfAutocommit) const connection = Sqlite.makeConnection(run, runValues, { export: Effect.try({ diff --git a/packages/core/src/database/sqlite.node.ts b/packages/core/src/database/sqlite.node.ts index 9053a0def663..8fb4564faa66 100644 --- a/packages/core/src/database/sqlite.node.ts +++ b/packages/core/src/database/sqlite.node.ts @@ -3,7 +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 { Sqlite } from "./sqlite.js" +import { retryLocked, Sqlite } from "./sqlite.js" const TypeId = "~@opencode/core/database/SqliteNode" as const @@ -25,6 +25,7 @@ interface Config extends Sqlite.ClientConfig { const make = (options: Config) => Effect.gen(function* () { const native = (yield* Sqlite.Native) as DatabaseSync + const retryIfAutocommit = retryLocked(() => native.isTransaction) const run = (query: string, params: ReadonlyArray = []) => Effect.withFiber>, SqlError>((fiber) => { @@ -39,7 +40,7 @@ const make = (options: Config) => }), ) } - }) + }).pipe(retryIfAutocommit) const runValues = (query: string, params: ReadonlyArray = []) => Effect.withFiber>, SqlError>((fiber) => { @@ -57,7 +58,7 @@ const make = (options: Config) => }), ) } - }) + }).pipe(retryIfAutocommit) const connection = Sqlite.makeConnection(run, runValues, { loadExtension: (path: string) => diff --git a/packages/core/src/database/sqlite.ts b/packages/core/src/database/sqlite.ts index 186274fe3e31..e675a2248a43 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,26 @@ 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. +// Only autocommit statements are retried. Inside a transaction the same error +// repeats until the transaction is rolled back and restarted (SQLITE_BUSY_SNAPSHOT +// on a deferred transaction whose read snapshot went stale), so it surfaces at once. +export const retryLocked = + (inTransaction: () => boolean) => + (effect: Effect.Effect) => + effect.pipe( + Effect.retry({ + while: (error) => isSqlError(error) && error.reason._tag === "LockTimeoutError" && !inTransaction(), + schedule: Schedule.min([Schedule.exponential(50, 2), Schedule.spaced(250)]).pipe( + Schedule.jittered, + Schedule.while((meta) => meta.elapsed < 30_000), + ), + }), + ) + export const makeConnection = (run: Run, runValues: RunValues, extensions: Extensions) => identity({ execute(query, params, transformRows) { 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..03168b79ea8a --- /dev/null +++ b/packages/core/test/sqlite-lock-retry.test.ts @@ -0,0 +1,121 @@ +import { mkdtemp, rm } from "node:fs/promises" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { Database } from "bun:sqlite" +import { expect, test } from "bun:test" +import { EffectDrizzleSqlite } from "@opencode/core/database/drizzle" +import { Effect } from "effect" +import { sql } from "drizzle-orm" +import { Sqlite } from "@opencode/core/database/sqlite" +import { sqliteLayer } from "@opencode/core/database/sqlite.bun" + +// Regression: a statement blocked by another process's write lock (SQLITE_BUSY) +// must retry until the lock frees, not fail the prompt with LockTimeoutError. +test("retries statements while another process holds the write lock", async () => { + const dir = await mkdtemp(join(tmpdir(), "sqlite-retry-")) + const filename = join(dir, "locked.db") + + try { + const holder = new Database(filename, { create: true }) + holder.run("create table t (id integer primary key, v text)") + holder.run("insert into t values (1, 'before')") + holder.run("pragma journal_mode = WAL") + holder.run("begin immediate") + + // Release the lock 100ms in; the retry loop must recover. + setTimeout(() => holder.run("commit"), 100) + + const rows = (await Effect.runPromise( + Effect.gen(function* () { + const db = yield* EffectDrizzleSqlite.makeWithDefaults() + const native = (yield* Sqlite.Native) as Database + native.run("PRAGMA busy_timeout = 0") + yield* db.run(sql`update t set v = 'after' where id = 1`) + return yield* db.all(sql`select v from t where id = 1`) + }).pipe(Effect.provide(sqliteLayer({ filename })), Effect.scoped) as Effect.Effect< + Array>, + unknown, + never + >, + )) as Array> + expect(rows[0].v).toBe("after") + holder.close() + } finally { + await rm(dir, { recursive: true, force: true }).catch(() => {}) + } +}) + +// Regression: a DEFERRED WAL transaction that holds a read snapshot cannot be +// upgraded to a write after another connection commits — the snapshot is +// stale and the upgrade fails with SQLITE_BUSY_SNAPSHOT. The retry wrapper, +// applied indiscriminately, would loop this error for the full 30 s budget; +// only a rollback + restart of the whole transaction helps. Inside an +// explicit transaction, statements must fail fast. +test("fails fast on BUSY_SNAPSHOT inside a deferred transaction", async () => { + const dir = await mkdtemp(join(tmpdir(), "sqlite-snapshot-")) + const filename = join(dir, "snap.db") + + try { + // Set up schema and one row, enable WAL, all via the layer. + await Effect.runPromise( + Effect.gen(function* () { + const native = (yield* Sqlite.Native) as Database + native.run("create table t (id integer primary key, v text)") + native.run("insert into t values (1, 'before')") + native.run("PRAGMA journal_mode = WAL") + native.run("PRAGMA busy_timeout = 0") + }).pipe(Effect.provide(sqliteLayer({ filename })), Effect.scoped), + ) + + // Inside a deferred transaction: read, then have another connection write + // and commit (invalidating the read snapshot), then attempt a write. + // The write must surface a LockTimeoutError immediately, not loop the + // 30 s retry budget. + const start = Date.now() + let error: unknown + try { + 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`) + // A separate connection writes and commits while we hold the + // deferred transaction's read snapshot. + yield* Effect.sync(() => { + const other = new Database(filename) + other.run("update t set v = 'after' where id = 1") + other.close() + }) + // This write must fail with BUSY_SNAPSHOT and surface fast — + // the retry helper must NOT keep trying the same stale snapshot. + yield* tx.run(sql`update t set v = 'after2' where id = 1`) + }), + { behavior: "deferred" }, + ) + }).pipe(Effect.provide(sqliteLayer({ filename })), Effect.scoped), + ) + } catch (e) { + error = e + } + const elapsed = Date.now() - start + + expect(error).toBeDefined() + // drizzle wraps the raw SqlError inside EffectDrizzleQueryError.cause. + // Effect wraps the SqlError in a Cause. Walk the failure chain to confirm + // the surface error is a LockTimeoutError wrapping SQLITE_BUSY_SNAPSHOT. + const reasons = + ( + error as { + cause?: { reasons?: ReadonlyArray<{ error?: { reason?: { _tag?: string; cause?: { code?: string } } } }> } + } + ).cause?.reasons ?? [] + const inner = reasons[0]?.error + expect(inner?.reason?._tag).toBe("LockTimeoutError") + expect((inner?.reason?.cause as { code?: string } | undefined)?.code).toBe("SQLITE_BUSY_SNAPSHOT") + expect(elapsed).toBeLessThan(2000) + } finally { + await rm(dir, { recursive: true, force: true }).catch(() => {}) + } +}) From c11d013493bbbd4a7df18052d8bde033db4f3856 Mon Sep 17 00:00:00 2001 From: Lance Woodson <65194841+woodsonl@users.noreply.github.com> Date: Sat, 10 Oct 2026 21:10:48 -0500 Subject: [PATCH 2/5] fix(core): classify node busy errors as lock timeouts and cover the retry paths Review of the V2 port found three gaps: - node:sqlite reports SQLITE_BUSY as { code: "ERR_SQLITE_ERROR", errcode: 5 } with no errno and no SQLITE_* string code, which is all classifySqliteError reads, so the lock timeout filed as UnknownError and the retry never fired on the node driver. node's errcode is now copied onto errno before classification, which the classifier maps to LockTimeoutError (verified red against the old code, green after). - The retry policy is now named constants in one place, with Schedule.upTo replacing the hand-elapsed check. - Tests cover the node retry path both ways (retry and the in-transaction fail-fast case), the runValues wiring, and a non-lock error surfacing fast, so dropping the pipe or widening the predicate breaks the suite. --- packages/core/src/database/sqlite.node.ts | 34 ++-- packages/core/src/database/sqlite.ts | 23 ++- packages/core/test/sqlite-lock-retry.test.ts | 175 +++++++++++++++++++ 3 files changed, 218 insertions(+), 14 deletions(-) diff --git a/packages/core/src/database/sqlite.node.ts b/packages/core/src/database/sqlite.node.ts index 8fb4564faa66..4e73c8ff7051 100644 --- a/packages/core/src/database/sqlite.node.ts +++ b/packages/core/src/database/sqlite.node.ts @@ -22,6 +22,28 @@ interface Config extends Sqlite.ClientConfig { readonly allowExtension?: boolean } +const isNodeSqliteError = (cause: unknown): cause is Record => + typeof cause === "object" && cause !== null && 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 @@ -34,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") })) } }).pipe(retryIfAutocommit) @@ -52,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") })) } }).pipe(retryIfAutocommit) diff --git a/packages/core/src/database/sqlite.ts b/packages/core/src/database/sqlite.ts index e675a2248a43..69bd1943b16f 100644 --- a/packages/core/src/database/sqlite.ts +++ b/packages/core/src/database/sqlite.ts @@ -31,16 +31,31 @@ type RunValues = ( // Only autocommit statements are retried. Inside a transaction the same error // repeats until the transaction is rolled back and restarted (SQLITE_BUSY_SNAPSHOT // on a deferred transaction whose read snapshot went stale), so it surfaces at once. +// +// The budget and backoff curve are the policy in one place: +// 50ms doubling to 250ms, jittered, giving up after LOCK_BUDGET_MS total. +// Note the sleep runs inside the statement's connection lease (SqlClient +// acquires the single connection for the whole attempt), so a contended +// write back-pressures other statements on the same client for the duration; +// the budget bounds that hold per statement, and 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 + export const retryLocked = (inTransaction: () => boolean) => (effect: Effect.Effect) => effect.pipe( Effect.retry({ while: (error) => isSqlError(error) && error.reason._tag === "LockTimeoutError" && !inTransaction(), - schedule: Schedule.min([Schedule.exponential(50, 2), Schedule.spaced(250)]).pipe( - Schedule.jittered, - Schedule.while((meta) => meta.elapsed < 30_000), - ), + schedule: Schedule.min([ + Schedule.exponential(LOCK_RETRY_INITIAL_MS, LOCK_RETRY_FACTOR), + Schedule.spaced(LOCK_RETRY_MAX_MS), + ]) + .pipe(Schedule.jittered) + .pipe(Schedule.upTo(LOCK_RETRY_BUDGET_MS)), }), ) diff --git a/packages/core/test/sqlite-lock-retry.test.ts b/packages/core/test/sqlite-lock-retry.test.ts index 03168b79ea8a..4a940e2ceeaf 100644 --- a/packages/core/test/sqlite-lock-retry.test.ts +++ b/packages/core/test/sqlite-lock-retry.test.ts @@ -6,6 +6,7 @@ import { expect, test } from "bun:test" import { EffectDrizzleSqlite } from "@opencode/core/database/drizzle" import { Effect } from "effect" import { sql } from "drizzle-orm" +import { isSqlError } from "effect/sql/SqlError" import { Sqlite } from "@opencode/core/database/sqlite" import { sqliteLayer } from "@opencode/core/database/sqlite.bun" @@ -119,3 +120,177 @@ test("fails fast on BUSY_SNAPSHOT inside a deferred transaction", async () => { await rm(dir, { recursive: true, force: true }).catch(() => {}) } }) + +// Regression: the node driver errors with { code: "ERR_SQLITE_ERROR", +// errcode: 5, errstr: "database is locked" } — no errno, no SQLITE_* string +// code — so effect's classifier files the busy error as UnknownError and the +// retry wrapper never fires. node:sqlite's DatabaseSync is typed here instead +// of bun:sqlite's Database; everything else matches the bun test above. +test("node driver retries statements while another process holds the write lock", async () => { + const { DatabaseSync } = await import("node:sqlite") + const { sqliteLayer: nodeSqliteLayer } = await import("@opencode/core/database/sqlite.node") + const dir = await mkdtemp(join(tmpdir(), "sqlite-node-retry-")) + const filename = join(dir, "locked.db") + + try { + const holder = new DatabaseSync(filename) + holder.exec("create table t (id integer primary key, v text)") + holder.exec("insert into t values (1, 'before')") + holder.exec("pragma journal_mode = WAL") + holder.exec("begin immediate") + setTimeout(() => { + holder.exec("commit") + holder.close() + }, 100) + + const rows = (await Effect.runPromise( + Effect.gen(function* () { + const db = yield* EffectDrizzleSqlite.makeWithDefaults() + const native = (yield* Sqlite.Native) as DatabaseSync + native.exec("PRAGMA busy_timeout = 0") + yield* db.run(sql`update t set v = 'after' where id = 1`) + return yield* db.all(sql`select v from t where id = 1`) + }).pipe(Effect.provide(nodeSqliteLayer({ filename })), Effect.scoped) as Effect.Effect< + Array>, + unknown, + never + >, + )) as Array> + expect(rows[0].v).toBe("after") + } finally { + await rm(dir, { recursive: true, force: true }).catch(() => {}) + } +}) + +// Regression: the node in-transaction BUSY_SNAPSHOT path must fail fast, the +// same contract the bun driver test pins under the bun layer. +test("node in-transaction lock surfaces fast instead of retrying", async () => { + const { DatabaseSync } = await import("node:sqlite") + const { sqliteLayer: nodeSqliteLayer } = await import("@opencode/core/database/sqlite.node") + const dir = await mkdtemp(join(tmpdir(), "sqlite-node-snapshot-")) + const filename = join(dir, "snap.db") + + try { + await Effect.runPromise( + Effect.gen(function* () { + const native = (yield* Sqlite.Native) as DatabaseSync + native.exec("create table t (id integer primary key, v text)") + native.exec("insert into t values (1, 'before')") + native.exec("PRAGMA journal_mode = WAL") + native.exec("PRAGMA busy_timeout = 0") + }).pipe(Effect.provide(nodeSqliteLayer({ filename })), Effect.scoped), + ) + + const start = Date.now() + let error: unknown + try { + 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 DatabaseSync(filename) + other.exec("update t set v = 'after' where id = 1") + other.close() + }) + yield* tx.run(sql`update t set v = 'after2' where id = 1`) + }), + { behavior: "deferred" }, + ) + }).pipe(Effect.provide(nodeSqliteLayer({ filename })), Effect.scoped), + ) + } catch (e) { + error = e + } + const elapsed = Date.now() - start + expect(error).toBeDefined() + expect(elapsed).toBeLessThan(2000) + } finally { + await rm(dir, { recursive: true, force: true }).catch(() => {}) + } +}) + +// Regression: a non-lock failure (unique-constraint violation) must surface +// immediately, not ride the lock-retry schedule. Widening retryLocked's +// predicate beyond LockTimeoutError would retry this constraint error, and the +// only visible symptom would be slowness. +test("fails fast on a non-lock SqlError", async () => { + const { Database } = await import("bun:sqlite") + const dir = await mkdtemp(join(tmpdir(), "sqlite-constraint-")) + const filename = join(dir, "constraint.db") + + try { + const holder = new Database(filename, { create: true }) + holder.run("create table t (id integer primary key, v text)") + holder.run("insert into t values (1, 'before')") + holder.close() + + // Widening retryLocked's predicate beyond LockTimeoutError would retry + // this constraint violation for the whole retry budget, so the elapsed + // assertion is the predicate guard; the reason-class assertion pins that + // the driver classified the failure rather than leaving it unknown. + const start = Date.now() + let error: unknown + try { + 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(Date.now() - start).toBeLessThan(1000) + const reasons = ( + error as { + cause?: { reasons?: ReadonlyArray<{ error?: { reason?: { _tag?: string } } }> } + }, + ).cause?.reasons ?? [] + expect(reasons[0]?.error?.reason?._tag).not.toBe("LockTimeoutError") + } finally { + await rm(dir, { recursive: true, force: true }).catch(()=>{}) + } +}) + +// Regression: the retry pipe must also cover the runValues path (executeValues +// and executeValuesUnprepared route there). Every other test goes through +// db.run/db.all, so dropping .pipe(retryIfAutocommit) from either runValues +// block would leave the suite green while values-path writes stop retrying. +test("runValues path retries statements while another process holds the write lock", async () => { + const { Database } = await import("bun:sqlite") + const dir = await mkdtemp(join(tmpdir(), "sqlite-retry-values-")) + const filename = join(dir, "locked.db") + + try { + const holder = new Database(filename, { create: true }) + holder.run("create table t (id integer primary key, v text)") + holder.run("insert into t values (1, 'before')") + holder.run("pragma journal_mode = WAL") + holder.run("begin immediate") + setTimeout(() => holder.run("commit"), 100) + + const rows = (await Effect.runPromise( + Effect.gen(function* () { + const db = yield* EffectDrizzleSqlite.makeWithDefaults() + const native = (yield* Sqlite.Native) as Database + native.run("PRAGMA busy_timeout = 0") + // .values() routes to executeValues, i.e. the runValues wiring. + yield* db.values(sql`update t set v = 'after' where id = 1`) + return yield* db.values(sql`select v from t where id = 1`) + }).pipe(Effect.provide(sqliteLayer({ filename })), Effect.scoped) as Effect.Effect< + ReadonlyArray>, + unknown, + never + >, + )) as ReadonlyArray> + expect(rows[0][0]).toBe("after") + holder.close() + } finally { + await rm(dir, { recursive: true, force: true }).catch(() => {}) + } +}) From ad7d907f4de381a499ab88262d0a31062ef8a35b Mon Sep 17 00:00:00 2001 From: Lance Woodson <65194841+woodsonl@users.noreply.github.com> Date: Sun, 11 Oct 2026 11:02:59 -0500 Subject: [PATCH 3/5] fix(core): re-enter the client semaphore on each lock-retry attempt A retry that only repeats the statement sleeps without holding the semaphore, so a transaction another fiber opens while it sleeps swallows the next attempt: its writes roll back with that transaction, or the in-transaction check, which reads the shared connection, sees the foreign transaction and gives up. Plain attempts now re-enter the semaphore before executing. The running transaction's own statements fail fast on BUSY_SNAPSHOT, and its BEGIN/COMMIT/ROLLBACK retry in place while it holds the semaphore. The gate lives in makeClient, so the drivers no longer carry a retryLocked pipe keyed on shared connection state. Also fixes the failures tsgo reports on the previous head: Schedule.upTo takes an options object, the node busy-error guard indexed an un-narrowed {}, the test file had a trailing comma inside an as cast plus an unused import, and DatabaseSync was used as a type. --- packages/core/src/database/sqlite.bun.ts | 7 +- packages/core/src/database/sqlite.node.ts | 10 +-- packages/core/src/database/sqlite.ts | 94 ++++++++++++++------ packages/core/test/sqlite-lock-retry.test.ts | 73 ++++++++++++++- 4 files changed, 146 insertions(+), 38 deletions(-) diff --git a/packages/core/src/database/sqlite.bun.ts b/packages/core/src/database/sqlite.bun.ts index 920ce372c05e..7728bc8f4e95 100644 --- a/packages/core/src/database/sqlite.bun.ts +++ b/packages/core/src/database/sqlite.bun.ts @@ -3,7 +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 { retryLocked, Sqlite } from "./sqlite.js" +import { Sqlite } from "./sqlite.js" const TypeId = "~@opencode/core/database/SqliteBun" as const @@ -23,7 +23,6 @@ interface Config extends Sqlite.ClientConfig { const make = (options: Config) => Effect.gen(function* () { const native = (yield* Sqlite.Native) as Database - const retryIfAutocommit = retryLocked(() => native.inTransaction) const run = (query: string, params: ReadonlyArray = []) => Effect.withFiber>, SqlError>((fiber) => { @@ -39,7 +38,7 @@ const make = (options: Config) => }), ) } - }).pipe(retryIfAutocommit) + }) const runValues = (query: string, params: ReadonlyArray = []) => Effect.withFiber, SqlError>((fiber) => { @@ -55,7 +54,7 @@ const make = (options: Config) => }), ) } - }).pipe(retryIfAutocommit) + }) const connection = Sqlite.makeConnection(run, runValues, { export: Effect.try({ diff --git a/packages/core/src/database/sqlite.node.ts b/packages/core/src/database/sqlite.node.ts index 4e73c8ff7051..9c09923e405e 100644 --- a/packages/core/src/database/sqlite.node.ts +++ b/packages/core/src/database/sqlite.node.ts @@ -3,7 +3,8 @@ import { Context, Effect, Layer } from "effect" import { Reactivity } from "effect/reactivity" import { SqlClient } from "effect/sql" import { classifySqliteError, SqlError } from "effect/sql/SqlError" -import { retryLocked, Sqlite } from "./sqlite.js" +import { isRecord } from "@opencode/ai/utils/record" +import { Sqlite } from "./sqlite.js" const TypeId = "~@opencode/core/database/SqliteNode" as const @@ -23,7 +24,7 @@ interface Config extends Sqlite.ClientConfig { } const isNodeSqliteError = (cause: unknown): cause is Record => - typeof cause === "object" && cause !== null && typeof cause["errcode"] === "number" + 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 @@ -47,7 +48,6 @@ const classifyNodeSqliteError = (cause: unknown, operation: string) => { const make = (options: Config) => Effect.gen(function* () { const native = (yield* Sqlite.Native) as DatabaseSync - const retryIfAutocommit = retryLocked(() => native.isTransaction) const run = (query: string, params: ReadonlyArray = []) => Effect.withFiber>, SqlError>((fiber) => { @@ -58,7 +58,7 @@ const make = (options: Config) => } catch (cause) { return Effect.fail(new SqlError({ reason: classifyNodeSqliteError(cause, "execute") })) } - }).pipe(retryIfAutocommit) + }) const runValues = (query: string, params: ReadonlyArray = []) => Effect.withFiber>, SqlError>((fiber) => { @@ -72,7 +72,7 @@ const make = (options: Config) => } catch (cause) { return Effect.fail(new SqlError({ reason: classifyNodeSqliteError(cause, "execute") })) } - }).pipe(retryIfAutocommit) + }) const connection = Sqlite.makeConnection(run, runValues, { loadExtension: (path: string) => diff --git a/packages/core/src/database/sqlite.ts b/packages/core/src/database/sqlite.ts index 69bd1943b16f..e561e1a1cf86 100644 --- a/packages/core/src/database/sqlite.ts +++ b/packages/core/src/database/sqlite.ts @@ -28,36 +28,43 @@ type RunValues = ( // 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. -// Only autocommit statements are retried. Inside a transaction the same error -// repeats until the transaction is rolled back and restarted (SQLITE_BUSY_SNAPSHOT -// on a deferred transaction whose read snapshot went stale), so it surfaces at once. +// +// 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. BEGIN, COMMIT and ROLLBACK run outside that +// context while our transaction holds the semaphore, so they retry in place +// rather than re-take it. // // The budget and backoff curve are the policy in one place: // 50ms doubling to 250ms, jittered, giving up after LOCK_BUDGET_MS total. -// Note the sleep runs inside the statement's connection lease (SqlClient -// acquires the single connection for the whole attempt), so a contended -// write back-pressures other statements on the same client for the duration; -// the budget bounds that hold per statement, and a caller with N serial -// statements can spend N times it before surfacing. +// 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 -export const retryLocked = - (inTransaction: () => boolean) => - (effect: Effect.Effect) => - effect.pipe( - Effect.retry({ - while: (error) => isSqlError(error) && error.reason._tag === "LockTimeoutError" && !inTransaction(), - schedule: Schedule.min([ - Schedule.exponential(LOCK_RETRY_INITIAL_MS, LOCK_RETRY_FACTOR), - Schedule.spaced(LOCK_RETRY_MAX_MS), - ]) - .pipe(Schedule.jittered) - .pipe(Schedule.upTo(LOCK_RETRY_BUDGET_MS)), - }), - ) +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({ @@ -82,6 +89,9 @@ export const makeConnection = (run: Run, runValues: R ...extensions, }) +// One key per client, so the gate only ever reads this client's transaction marker. +let transactionCounter = 0 + export const makeClient = < Config extends ClientConfig, SqliteConnection extends Connection, @@ -95,13 +105,46 @@ 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++}`) + let transactionOpen = false + const gateStatement = (attempt: Effect.Effect): Effect.Effect => + Effect.withFiber((fiber) => { + if (Context.getOption(fiber.context, transactionService)._tag === "Some") return attempt + // BEGIN/COMMIT/ROLLBACK run outside the transaction's service context + // while it holds the semaphore; re-taking it there would deadlock. + if (transactionOpen) return retryLockTimeout(attempt) + return retryLockTimeout(Effect.andThen(acquirer, attempt)) + }) + const gatedConnection: SqliteConnection = { + ...connection, + execute: (query, params, transformRows) => gateStatement(connection.execute(query, params, transformRows)), + executeRaw: (query, params) => gateStatement(connection.executeRaw(query, params)), + executeValues: (query, params) => gateStatement(connection.executeValues(query, params)), + executeUnprepared: (query, params, transformRows) => + gateStatement(connection.executeUnprepared(query, params, transformRows)), + } + const acquirer = semaphore.withPermits(1)(Effect.succeed(gatedConnection)) 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, + Effect.tap(restore(semaphore.take(1)), () => { + transactionOpen = true + // Flip before releasing so a plain statement racing the close is gated. + return Scope.addFinalizer( + scope, + Effect.andThen( + Effect.sync(() => { + transactionOpen = false + }), + semaphore.release(1), + ), + ) + }), + gatedConnection, ) }) const transformRows = options.transformResultNames @@ -111,6 +154,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 index 4a940e2ceeaf..4738db8b276c 100644 --- a/packages/core/test/sqlite-lock-retry.test.ts +++ b/packages/core/test/sqlite-lock-retry.test.ts @@ -6,7 +6,6 @@ import { expect, test } from "bun:test" import { EffectDrizzleSqlite } from "@opencode/core/database/drizzle" import { Effect } from "effect" import { sql } from "drizzle-orm" -import { isSqlError } from "effect/sql/SqlError" import { Sqlite } from "@opencode/core/database/sqlite" import { sqliteLayer } from "@opencode/core/database/sqlite.bun" @@ -146,7 +145,7 @@ test("node driver retries statements while another process holds the write lock" const rows = (await Effect.runPromise( Effect.gen(function* () { const db = yield* EffectDrizzleSqlite.makeWithDefaults() - const native = (yield* Sqlite.Native) as DatabaseSync + const native = (yield* Sqlite.Native) as InstanceType native.exec("PRAGMA busy_timeout = 0") yield* db.run(sql`update t set v = 'after' where id = 1`) return yield* db.all(sql`select v from t where id = 1`) @@ -173,7 +172,7 @@ test("node in-transaction lock surfaces fast instead of retrying", async () => { try { await Effect.runPromise( Effect.gen(function* () { - const native = (yield* Sqlite.Native) as DatabaseSync + const native = (yield* Sqlite.Native) as InstanceType native.exec("create table t (id integer primary key, v text)") native.exec("insert into t values (1, 'before')") native.exec("PRAGMA journal_mode = WAL") @@ -249,7 +248,7 @@ test("fails fast on a non-lock SqlError", async () => { const reasons = ( error as { cause?: { reasons?: ReadonlyArray<{ error?: { reason?: { _tag?: string } } }> } - }, + } ).cause?.reasons ?? [] expect(reasons[0]?.error?.reason?._tag).not.toBe("LockTimeoutError") } finally { @@ -294,3 +293,69 @@ test("runValues path retries statements while another process holds the write lo await rm(dir, { recursive: true, force: true }).catch(() => {}) } }) + +// 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 then runs inside that transaction — its +// writes are rolled back with it, or the retry predicate sees the foreign +// transaction and gives up. Fiber A's update must wait out fiber B's +// transaction and land in autocommit after it closes. +test("a retrying statement waits out a transaction that opens while it sleeps", async () => { + const dir = await mkdtemp(join(tmpdir(), "sqlite-retry-race-")) + const filename = join(dir, "race.db") + + try { + await Effect.runPromise( + Effect.gen(function* () { + const native = (yield* Sqlite.Native) as Database + native.run("create table t (id integer primary key, v text)") + native.run("insert into t values (1, 'before')") + native.run("PRAGMA journal_mode = WAL") + native.run("PRAGMA busy_timeout = 0") + }).pipe(Effect.provide(sqliteLayer({ filename })), Effect.scoped), + ) + + // A second process holds the write lock, so fiber A's first attempt files + // as LockTimeoutError and the retry sleeps. + const holder = new Database(filename) + holder.run("begin immediate") + setTimeout(() => { + holder.run("commit") + holder.close() + }, 300) + + const rows = (await Effect.runPromise( + Effect.gen(function* () { + const db = yield* EffectDrizzleSqlite.makeWithDefaults() + const writer = Effect.gen(function* () { + // A: plain autocommit write, retrying through the holder's lock. + yield* Effect.sleep("10 millis") + yield* db.run(sql`update t set v = 'after' where id = 1`) + }) + const transactor = Effect.gen(function* () { + // B: opens a transaction while A sleeps and rolls it back. + yield* Effect.sleep("30 millis") + yield* db + .transaction( + (tx) => + Effect.gen(function* () { + yield* tx.get(sql`select v from t where id = 1`) + yield* Effect.sleep("120 millis") + yield* Effect.fail(new Error("rollback requested")) + }), + { behavior: "deferred" }, + ) + .pipe(Effect.ignore) + }) + yield* Effect.all([writer, transactor], { concurrency: 2, discard: true }) + return yield* db.all(sql`select v from t where id = 1`) + }).pipe( + Effect.provide(sqliteLayer({ filename })), + Effect.scoped, + ) as Effect.Effect>, unknown, never>, + )) as Array> + expect(rows[0].v).toBe("after") + } finally { + await rm(dir, { recursive: true, force: true }).catch(() => {}) + } +}) From a7af4559f06bde7093da4d7cc11fae1be2a9c74b Mon Sep 17 00:00:00 2001 From: Lance Woodson <65194841+woodsonl@users.noreply.github.com> Date: Tue, 27 Oct 2026 02:23:08 -0500 Subject: [PATCH 4/5] test(core): rewrite the lock-retry tests without aligned timers, document the control-statement bypass Review follow-up: - The lock holder is released by the test at wide margins instead of setTimeout, so the first attempt always precedes the release; the cross-fiber race is ordered with a Deferred that parks the interfering transaction mid-body. The interleaving no longer depends on 10/30/120ms sleeps lining up, and the race test now fails on the pre-fix head deterministically instead of by luck. - Shared setup (schema, WAL, holder, value reads) shrinks the file from 361 to ~250 lines across 6 focused cases: bun retry, values-path retry, node classification, fail-fast on a stale snapshot, fail-fast on non-lock errors, and the cross-fiber race. - The transactionOpen bypass in gateStatement now documents why only the transaction's own BEGIN/COMMIT/ROLLBACK can reach it: every gated statement runs holding the client semaphore, and the flag is true exactly while a transaction holds it, so a foreign statement queues instead of observing it. The header comment states the layering with caller-side whole-transaction replay (#53704) so the two retry layers stay disjoint. --- packages/core/src/database/sqlite.node.ts | 2 +- packages/core/src/database/sqlite.ts | 27 +- packages/core/test/sqlite-lock-retry.test.ts | 499 ++++++++----------- 3 files changed, 223 insertions(+), 305 deletions(-) diff --git a/packages/core/src/database/sqlite.node.ts b/packages/core/src/database/sqlite.node.ts index 9c09923e405e..3fc493e628cc 100644 --- a/packages/core/src/database/sqlite.node.ts +++ b/packages/core/src/database/sqlite.node.ts @@ -3,7 +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/ai/utils/record" +import { isRecord } from "@opencode/util/record" import { Sqlite } from "./sqlite.js" const TypeId = "~@opencode/core/database/SqliteNode" as const diff --git a/packages/core/src/database/sqlite.ts b/packages/core/src/database/sqlite.ts index e561e1a1cf86..1a087fd93a5d 100644 --- a/packages/core/src/database/sqlite.ts +++ b/packages/core/src/database/sqlite.ts @@ -38,9 +38,15 @@ type RunValues = ( // 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. BEGIN, COMMIT and ROLLBACK run outside that -// context while our transaction holds the semaphore, so they retry in place -// rather than re-take it. +// 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. @@ -113,8 +119,19 @@ export const makeClient = < const gateStatement = (attempt: Effect.Effect): Effect.Effect => Effect.withFiber((fiber) => { if (Context.getOption(fiber.context, transactionService)._tag === "Some") return attempt - // BEGIN/COMMIT/ROLLBACK run outside the transaction's service context - // while it holds the semaphore; re-taking it there would deadlock. + // Only the transaction's own control statements can reach here with + // the flag set. Every gated statement runs while holding the client + // semaphore — both acquirers take it before handing out the + // connection, and the statement executes inside their scope — and + // `transactionOpen` is true exactly while a transaction holds that + // semaphore. A foreign statement therefore never observes it true: + // it queues on the semaphore until the transaction closes, by which + // point the flag has flipped back. effect-sql runs BEGIN before it + // provides the transaction service and COMMIT/ROLLBACK after that + // context is withdrawn (nested savepoints run inside it, so they + // arrive above with the service), so exactly those three arrive here + // service-less while owning the semaphore; re-taking it there would + // deadlock, so they retry in place. if (transactionOpen) return retryLockTimeout(attempt) return retryLockTimeout(Effect.andThen(acquirer, attempt)) }) diff --git a/packages/core/test/sqlite-lock-retry.test.ts b/packages/core/test/sqlite-lock-retry.test.ts index 4738db8b276c..0a780542a660 100644 --- a/packages/core/test/sqlite-lock-retry.test.ts +++ b/packages/core/test/sqlite-lock-retry.test.ts @@ -4,358 +4,259 @@ import { join } from "node:path" import { Database } from "bun:sqlite" import { expect, test } from "bun:test" import { EffectDrizzleSqlite } from "@opencode/core/database/drizzle" -import { Effect } from "effect" +import { Deferred, Effect, Fiber } from "effect" import { sql } from "drizzle-orm" import { Sqlite } from "@opencode/core/database/sqlite" import { sqliteLayer } from "@opencode/core/database/sqlite.bun" -// Regression: a statement blocked by another process's write lock (SQLITE_BUSY) -// must retry until the lock frees, not fail the prompt with LockTimeoutError. -test("retries statements while another process holds the write lock", async () => { - const dir = await mkdtemp(join(tmpdir(), "sqlite-retry-")) - const filename = join(dir, "locked.db") +// 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. - try { - const holder = new Database(filename, { create: true }) - holder.run("create table t (id integer primary key, v text)") - holder.run("insert into t values (1, 'before')") - holder.run("pragma journal_mode = WAL") - holder.run("begin immediate") +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(() => {}) + } + }) - // Release the lock 100ms in; the retry loop must recover. - setTimeout(() => holder.run("commit"), 100) +// 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 rows = (await Effect.runPromise( - Effect.gen(function* () { - const db = yield* EffectDrizzleSqlite.makeWithDefaults() - const native = (yield* Sqlite.Native) as Database - native.run("PRAGMA busy_timeout = 0") - yield* db.run(sql`update t set v = 'after' where id = 1`) - return yield* db.all(sql`select v from t where id = 1`) - }).pipe(Effect.provide(sqliteLayer({ filename })), Effect.scoped) as Effect.Effect< - Array>, - unknown, - never - >, - )) as Array> - expect(rows[0].v).toBe("after") - holder.close() - } finally { - await rm(dir, { recursive: true, force: true }).catch(() => {}) - } -}) +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 DEFERRED WAL transaction that holds a read snapshot cannot be -// upgraded to a write after another connection commits — the snapshot is -// stale and the upgrade fails with SQLITE_BUSY_SNAPSHOT. The retry wrapper, -// applied indiscriminately, would loop this error for the full 30 s budget; -// only a rollback + restart of the whole transaction helps. Inside an -// explicit transaction, statements must fail fast. -test("fails fast on BUSY_SNAPSHOT inside a deferred transaction", async () => { - const dir = await mkdtemp(join(tmpdir(), "sqlite-snapshot-")) - const filename = join(dir, "snap.db") +// 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") - try { - // Set up schema and one row, enable WAL, all via the layer. await Effect.runPromise( Effect.gen(function* () { - const native = (yield* Sqlite.Native) as Database - native.run("create table t (id integer primary key, v text)") - native.run("insert into t values (1, 'before')") - native.run("PRAGMA journal_mode = WAL") - native.run("PRAGMA busy_timeout = 0") + 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), ) - - // Inside a deferred transaction: read, then have another connection write - // and commit (invalidating the read snapshot), then attempt a write. - // The write must surface a LockTimeoutError immediately, not loop the - // 30 s retry budget. - const start = Date.now() - let error: unknown - try { - 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`) - // A separate connection writes and commits while we hold the - // deferred transaction's read snapshot. - yield* Effect.sync(() => { - const other = new Database(filename) - other.run("update t set v = 'after' where id = 1") - other.close() - }) - // This write must fail with BUSY_SNAPSHOT and surface fast — - // the retry helper must NOT keep trying the same stale snapshot. - yield* tx.run(sql`update t set v = 'after2' where id = 1`) - }), - { behavior: "deferred" }, - ) - }).pipe(Effect.provide(sqliteLayer({ filename })), Effect.scoped), - ) - } catch (e) { - error = e - } - const elapsed = Date.now() - start - - expect(error).toBeDefined() - // drizzle wraps the raw SqlError inside EffectDrizzleQueryError.cause. - // Effect wraps the SqlError in a Cause. Walk the failure chain to confirm - // the surface error is a LockTimeoutError wrapping SQLITE_BUSY_SNAPSHOT. - const reasons = - ( - error as { - cause?: { reasons?: ReadonlyArray<{ error?: { reason?: { _tag?: string; cause?: { code?: string } } } }> } - } - ).cause?.reasons ?? [] - const inner = reasons[0]?.error - expect(inner?.reason?._tag).toBe("LockTimeoutError") - expect((inner?.reason?.cause as { code?: string } | undefined)?.code).toBe("SQLITE_BUSY_SNAPSHOT") - expect(elapsed).toBeLessThan(2000) - } finally { - await rm(dir, { recursive: true, force: true }).catch(() => {}) - } + expect(await readValue(filename)).toBe("after") + }) }) -// Regression: the node driver errors with { code: "ERR_SQLITE_ERROR", -// errcode: 5, errstr: "database is locked" } — no errno, no SQLITE_* string -// code — so effect's classifier files the busy error as UnknownError and the -// retry wrapper never fires. node:sqlite's DatabaseSync is typed here instead -// of bun:sqlite's Database; everything else matches the bun test above. -test("node driver retries statements while another process holds the write lock", async () => { - const { DatabaseSync } = await import("node:sqlite") - const { sqliteLayer: nodeSqliteLayer } = await import("@opencode/core/database/sqlite.node") - const dir = await mkdtemp(join(tmpdir(), "sqlite-node-retry-")) - const filename = join(dir, "locked.db") - - try { - const holder = new DatabaseSync(filename) - holder.exec("create table t (id integer primary key, v text)") - holder.exec("insert into t values (1, 'before')") - holder.exec("pragma journal_mode = WAL") - holder.exec("begin immediate") - setTimeout(() => { - holder.exec("commit") - holder.close() - }, 100) +// 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") - const rows = (await Effect.runPromise( + await Effect.runPromise( Effect.gen(function* () { const db = yield* EffectDrizzleSqlite.makeWithDefaults() - const native = (yield* Sqlite.Native) as InstanceType - native.exec("PRAGMA busy_timeout = 0") - yield* db.run(sql`update t set v = 'after' where id = 1`) - return yield* db.all(sql`select v from t where id = 1`) - }).pipe(Effect.provide(nodeSqliteLayer({ filename })), Effect.scoped) as Effect.Effect< - Array>, - unknown, - never - >, - )) as Array> - expect(rows[0].v).toBe("after") - } finally { - await rm(dir, { recursive: true, force: true }).catch(() => {}) - } + ;((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: the node in-transaction BUSY_SNAPSHOT path must fail fast, the -// same contract the bun driver test pins under the bun layer. -test("node in-transaction lock surfaces fast instead of retrying", async () => { +// 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") - const dir = await mkdtemp(join(tmpdir(), "sqlite-node-snapshot-")) - const filename = join(dir, "snap.db") + 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") - try { await Effect.runPromise( Effect.gen(function* () { - const native = (yield* Sqlite.Native) as InstanceType - native.exec("create table t (id integer primary key, v text)") - native.exec("insert into t values (1, 'before')") - native.exec("PRAGMA journal_mode = WAL") - native.exec("PRAGMA busy_timeout = 0") + 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 - try { - 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 DatabaseSync(filename) - other.exec("update t set v = 'after' where id = 1") - other.close() - }) - yield* tx.run(sql`update t set v = 'after2' where id = 1`) - }), - { behavior: "deferred" }, - ) - }).pipe(Effect.provide(nodeSqliteLayer({ filename })), Effect.scoped), - ) - } catch (e) { + 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 - } - const elapsed = Date.now() - start + }) + expect(error).toBeDefined() - expect(elapsed).toBeLessThan(2000) - } finally { - await rm(dir, { recursive: true, force: true }).catch(() => {}) - } + expect(surfaceReason(error)?._tag).toBe("LockTimeoutError") + expect(surfaceReason(error)?.cause?.code).toBe("SQLITE_BUSY_SNAPSHOT") + expect(Date.now() - start).toBeLessThan(2000) + }) }) -// Regression: a non-lock failure (unique-constraint violation) must surface -// immediately, not ride the lock-retry schedule. Widening retryLocked's -// predicate beyond LockTimeoutError would retry this constraint error, and the -// only visible symptom would be slowness. -test("fails fast on a non-lock SqlError", async () => { - const { Database } = await import("bun:sqlite") - const dir = await mkdtemp(join(tmpdir(), "sqlite-constraint-")) - const filename = join(dir, "constraint.db") - - try { - const holder = new Database(filename, { create: true }) - holder.run("create table t (id integer primary key, v text)") - holder.run("insert into t values (1, 'before')") - holder.close() +// 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)) - // Widening retryLocked's predicate beyond LockTimeoutError would retry - // this constraint violation for the whole retry budget, so the elapsed - // assertion is the predicate guard; the reason-class assertion pins that - // the driver classified the failure rather than leaving it unknown. const start = Date.now() let error: unknown - try { - 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) { + 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) - const reasons = ( - error as { - cause?: { reasons?: ReadonlyArray<{ error?: { reason?: { _tag?: string } } }> } - } - ).cause?.reasons ?? [] - expect(reasons[0]?.error?.reason?._tag).not.toBe("LockTimeoutError") - } finally { - await rm(dir, { recursive: true, force: true }).catch(()=>{}) - } + }) }) -// Regression: the retry pipe must also cover the runValues path (executeValues -// and executeValuesUnprepared route there). Every other test goes through -// db.run/db.all, so dropping .pipe(retryIfAutocommit) from either runValues -// block would leave the suite green while values-path writes stop retrying. -test("runValues path retries statements while another process holds the write lock", async () => { - const { Database } = await import("bun:sqlite") - const dir = await mkdtemp(join(tmpdir(), "sqlite-retry-values-")) - const filename = join(dir, "locked.db") +// 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") - try { - const holder = new Database(filename, { create: true }) - holder.run("create table t (id integer primary key, v text)") - holder.run("insert into t values (1, 'before')") - holder.run("pragma journal_mode = WAL") - holder.run("begin immediate") - setTimeout(() => holder.run("commit"), 100) - - const rows = (await Effect.runPromise( - Effect.gen(function* () { - const db = yield* EffectDrizzleSqlite.makeWithDefaults() - const native = (yield* Sqlite.Native) as Database - native.run("PRAGMA busy_timeout = 0") - // .values() routes to executeValues, i.e. the runValues wiring. - yield* db.values(sql`update t set v = 'after' where id = 1`) - return yield* db.values(sql`select v from t where id = 1`) - }).pipe(Effect.provide(sqliteLayer({ filename })), Effect.scoped) as Effect.Effect< - ReadonlyArray>, - unknown, - never - >, - )) as ReadonlyArray> - expect(rows[0][0]).toBe("after") - holder.close() - } finally { - await rm(dir, { recursive: true, force: true }).catch(() => {}) - } -}) - -// 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 then runs inside that transaction — its -// writes are rolled back with it, or the retry predicate sees the foreign -// transaction and gives up. Fiber A's update must wait out fiber B's -// transaction and land in autocommit after it closes. -test("a retrying statement waits out a transaction that opens while it sleeps", async () => { - const dir = await mkdtemp(join(tmpdir(), "sqlite-retry-race-")) - const filename = join(dir, "race.db") - - try { await Effect.runPromise( Effect.gen(function* () { - const native = (yield* Sqlite.Native) as Database - native.run("create table t (id integer primary key, v text)") - native.run("insert into t values (1, 'before')") - native.run("PRAGMA journal_mode = WAL") - native.run("PRAGMA busy_timeout = 0") - }).pipe(Effect.provide(sqliteLayer({ filename })), Effect.scoped), - ) + const db = yield* EffectDrizzleSqlite.makeWithDefaults() + ;((yield* Sqlite.Native) as Database).run("pragma busy_timeout = 0") - // A second process holds the write lock, so fiber A's first attempt files - // as LockTimeoutError and the retry sleeps. - const holder = new Database(filename) - holder.run("begin immediate") - setTimeout(() => { - holder.run("commit") - holder.close() - }, 300) + // 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") - const rows = (await Effect.runPromise( - Effect.gen(function* () { - const db = yield* EffectDrizzleSqlite.makeWithDefaults() - const writer = Effect.gen(function* () { - // A: plain autocommit write, retrying through the holder's lock. - yield* Effect.sleep("10 millis") - yield* db.run(sql`update t set v = 'after' where id = 1`) - }) - const transactor = Effect.gen(function* () { - // B: opens a transaction while A sleeps and rolls it back. - yield* Effect.sleep("30 millis") - yield* db + // 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* Effect.sleep("120 millis") + yield* Deferred.await(inside) yield* Effect.fail(new Error("rollback requested")) }), { behavior: "deferred" }, ) - .pipe(Effect.ignore) - }) - yield* Effect.all([writer, transactor], { concurrency: 2, discard: true }) - return yield* db.all(sql`select v from t where id = 1`) - }).pipe( - Effect.provide(sqliteLayer({ filename })), - Effect.scoped, - ) as Effect.Effect>, unknown, never>, - )) as Array> - expect(rows[0].v).toBe("after") - } finally { - await rm(dir, { recursive: true, force: true }).catch(() => {}) - } + .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") + }) }) From 3e90cf21159caaebda8e91cdf8593f2ec74e9cbf Mon Sep 17 00:00:00 2001 From: Lance Woodson <65194841+woodsonl@users.noreply.github.com> Date: Tue, 27 Oct 2026 12:46:48 -0500 Subject: [PATCH 5/5] fix(core): hold the client permit across each sqlite statement attempt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The review on this PR caught that the acquirer releases the client semaphore as soon as it hands back the connection, before the statement runs — so a transaction opening in that gap could swallow the attempt, and a plain statement reading the transactionOpen flag outside the permit could take the control-statement shortcut and join a foreign transaction. The comment claimed that was impossible; it was not. Plain attempts now hold the permit for the attempt itself. Control statements are told apart structurally instead of by the flag: the transaction acquirer hands out a connection whose gate retries its service-less statements (BEGIN, COMMIT, ROLLBACK) in place, since the transaction already holds the permit, while plain statements gate through the acquirer's connection and take the permit per attempt. transactionOpen is gone. The new test parks a plain write mid-statement, past its acquirer, and opens a transaction there; on the previous gate the write joins the transaction and is rolled back with it (fails deterministically), with the permit held it lands in autocommit and survives. --- packages/core/src/database/sqlite.ts | 73 ++++++------ packages/core/test/sqlite-lock-retry.test.ts | 111 ++++++++++++++++++- 2 files changed, 144 insertions(+), 40 deletions(-) diff --git a/packages/core/src/database/sqlite.ts b/packages/core/src/database/sqlite.ts index 1a087fd93a5d..14cb4ccec12a 100644 --- a/packages/core/src/database/sqlite.ts +++ b/packages/core/src/database/sqlite.ts @@ -95,7 +95,7 @@ export const makeConnection = (run: Run, runValues: R ...extensions, }) -// One key per client, so the gate only ever reads this client's transaction marker. +// One key per client, so a gate never mistakes another client's transaction for its own. let transactionCounter = 0 export const makeClient = < @@ -115,53 +115,50 @@ export const makeClient = < SqlClient.TransactionConnection, SqlClient.TransactionConnection.Service >(`@opencode/core/database/SqliteTransaction/${transactionCounter++}`) - let transactionOpen = false + // 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 - // Only the transaction's own control statements can reach here with - // the flag set. Every gated statement runs while holding the client - // semaphore — both acquirers take it before handing out the - // connection, and the statement executes inside their scope — and - // `transactionOpen` is true exactly while a transaction holds that - // semaphore. A foreign statement therefore never observes it true: - // it queues on the semaphore until the transaction closes, by which - // point the flag has flipped back. effect-sql runs BEGIN before it - // provides the transaction service and COMMIT/ROLLBACK after that - // context is withdrawn (nested savepoints run inside it, so they - // arrive above with the service), so exactly those three arrive here - // service-less while owning the semaphore; re-taking it there would - // deadlock, so they retry in place. - if (transactionOpen) return retryLockTimeout(attempt) - return retryLockTimeout(Effect.andThen(acquirer, 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)) }) - const gatedConnection: SqliteConnection = { + // 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) => gateStatement(connection.execute(query, params, transformRows)), - executeRaw: (query, params) => gateStatement(connection.executeRaw(query, params)), - executeValues: (query, params) => gateStatement(connection.executeValues(query, params)), + 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) => - gateStatement(connection.executeUnprepared(query, params, transformRows)), - } - const acquirer = semaphore.withPermits(1)(Effect.succeed(gatedConnection)) + 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)), () => { - transactionOpen = true - // Flip before releasing so a plain statement racing the close is gated. - return Scope.addFinalizer( - scope, - Effect.andThen( - Effect.sync(() => { - transactionOpen = false - }), - semaphore.release(1), - ), - ) - }), - gatedConnection, + Effect.tap(restore(semaphore.take(1)), () => Scope.addFinalizer(scope, semaphore.release(1))), + gateConnection(gateTransactionStatement), ) }) const transformRows = options.transformResultNames diff --git a/packages/core/test/sqlite-lock-retry.test.ts b/packages/core/test/sqlite-lock-retry.test.ts index 0a780542a660..d5d18546c21d 100644 --- a/packages/core/test/sqlite-lock-retry.test.ts +++ b/packages/core/test/sqlite-lock-retry.test.ts @@ -1,11 +1,14 @@ import { mkdtemp, rm } from "node:fs/promises" import { tmpdir } from "node:os" import { join } from "node:path" -import { Database } from "bun:sqlite" +import { Database, type SQLQueryBindings } from "bun:sqlite" import { expect, test } from "bun:test" import { EffectDrizzleSqlite } from "@opencode/core/database/drizzle" -import { Deferred, Effect, Fiber } from "effect" +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" @@ -260,3 +263,107 @@ test("a retrying statement waits out a transaction opened while it sleeps", asyn 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") + }) +})