diff --git a/packages/core/src/database/database.ts b/packages/core/src/database/database.ts index fb5eb276103a..efc9169e30aa 100644 --- a/packages/core/src/database/database.ts +++ b/packages/core/src/database/database.ts @@ -1,4 +1,4 @@ -export * as Database from "./database.js" +export * as Database from "./database.js" import { EffectDrizzleSqlite } from "./drizzle.js" import { sqliteLayer, supportsForeignKeyToggle, supportsTuningPragmas } from "#sqlite" @@ -31,7 +31,7 @@ const databaseLayer = Layer.effect( if (supportsTuningPragmas) { yield* db.run("PRAGMA journal_mode = WAL") yield* db.run("PRAGMA synchronous = NORMAL") - yield* db.run("PRAGMA busy_timeout = 5000") + yield* db.run("PRAGMA busy_timeout = 0") yield* db.run("PRAGMA cache_size = -64000") yield* db.run("PRAGMA wal_checkpoint(PASSIVE)") } @@ -75,3 +75,4 @@ export function configuredClient(client: Layer.Layer) { } export const node = configured({ path: ":memory:" }) + diff --git a/packages/core/src/database/sqlite.node.ts b/packages/core/src/database/sqlite.node.ts index 837beaa1f88a..5c4d134ed1e6 100644 --- a/packages/core/src/database/sqlite.node.ts +++ b/packages/core/src/database/sqlite.node.ts @@ -81,7 +81,9 @@ const nativeLayer = (config: Config) => Effect.gen(function* () { const native = new DatabaseSync(config.filename, { readOnly: config.readonly, - timeout: config.timeout, + // Node's native busy wait would block the event loop; locked statements + // fail immediately and retry cooperatively in Sqlite.makeConnection. + timeout: config.timeout ?? 0, allowExtension: config.allowExtension, enableForeignKeyConstraints: true, open: true, diff --git a/packages/core/src/database/sqlite.ts b/packages/core/src/database/sqlite.ts index 6ab22de8caf5..6022be80ff50 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, Duration, Effect, Fiber, Schedule, Scope, Semaphore, Stream } from "effect" import { identity } from "effect/Function" import { SqlClient, Statement } from "effect/unstable/sql" import type { Connection } from "effect/unstable/sql/SqlConnection" -import type { SqlError } from "effect/unstable/sql/SqlError" +import { SqlError } from "effect/unstable/sql/SqlError" export class Native extends Context.Service()("@opencode-ai/core/database/SqliteNative") {} @@ -24,19 +24,36 @@ type RunValues = ( params?: ReadonlyArray, ) => Effect.Effect>, SqlError> +// SQLITE_BUSY/SQLITE_LOCKED mean the statement never executed, so retrying it +// is side-effect free. Native busy waiting (busy_timeout) blocks the whole +// event loop, so locked statements retry cooperatively with a bounded, +// jittered exponential schedule instead (roughly 25ms..800ms per attempt). +const lockedStatementSchedule = Schedule.exponential(Duration.millis(25)).pipe( + Schedule.jittered, + Schedule.upTo({ times: 6 }), +) + +const retryLocked = (effect: Effect.Effect) => + effect.pipe( + Effect.retry({ + schedule: lockedStatementSchedule, + while: (error) => error.reason._tag === "LockTimeoutError", + }), + ) + export const makeConnection = (run: Run, runValues: RunValues, extensions: Extensions) => identity({ execute(query, params, transformRows) { - return transformRows ? Effect.map(run(query, params), transformRows) : run(query, params) + return transformRows ? Effect.map(retryLocked(run(query, params)), transformRows) : retryLocked(run(query, params)) }, executeRaw(query, params) { - return run(query, params) + return retryLocked(run(query, params)) }, executeValues(query, params) { - return runValues(query, params) + return retryLocked(runValues(query, params)) }, executeValuesUnprepared(query, params) { - return runValues(query, params) + return retryLocked(runValues(query, params)) }, executeUnprepared(query, params, transformRows) { return this.execute(query, params, transformRows) @@ -95,3 +112,4 @@ export const makeClient = < readonly updateValues: never } & Extensions }) + diff --git a/packages/core/test/database-drizzle.test.ts b/packages/core/test/database-drizzle.test.ts index 0cbf60bf2b80..76b26a5bcc94 100644 --- a/packages/core/test/database-drizzle.test.ts +++ b/packages/core/test/database-drizzle.test.ts @@ -4,10 +4,11 @@ import { join } from "node:path" import { Database } from "bun:sqlite" import { expect, test } from "bun:test" import { SqliteClient } from "@effect/sql-sqlite-bun" +import { sqliteLayer } from "@opencode-ai/core/database/sqlite.bun" import { eq, sql } from "drizzle-orm" import { integer, sqliteTable, text } from "drizzle-orm/sqlite-core" -import { Effect } from "effect" -import type { SqlClient as SqlClientService } from "effect/unstable/sql/SqlClient" +import { Deferred, Effect, Fiber } from "effect" +import { SqlClient as SqlClientService } from "effect/unstable/sql/SqlClient" import { isSqlError } from "effect/unstable/sql/SqlError" import { EffectDrizzleSqlite } from "@opencode-ai/core/database/drizzle" @@ -130,6 +131,106 @@ test("preserves failed transaction begin errors", async () => { } }) +test("retries locked writes cooperatively once the holding transaction releases", async () => { + const dir = await mkdtemp(join(tmpdir(), "effect-drizzle-sqlite-")) + const filename = join(dir, "locked.db") + const holder = new Database(filename) + + try { + holder.run("create table users (id integer primary key autoincrement, name text not null)") + holder.run("pragma busy_timeout = 0") + holder.run("begin immediate") + + await Effect.runPromise( + Effect.gen(function* () { + const client = yield* SqlClientService + const done = yield* Deferred.make() + const insert = yield* client + .unsafe("insert into users (name) values (?)", ["Ada"]) + .values.pipe(Effect.ensuring(Deferred.succeed(done, undefined)), Effect.forkChild) + yield* Effect.yieldNow + + // The write is blocked by the holder's write lock and must retry + // cooperatively instead of native busy waiting. + expect(yield* Deferred.isDone(done)).toBe(false) + + holder.run("rollback") + yield* Fiber.join(insert) + expect(yield* client.unsafe("select * from users").values).toEqual([[1, "Ada"]]) + }).pipe(Effect.provide(sqliteLayer({ filename, disableWAL: true })), Effect.scoped), + ) + } finally { + if (holder.inTransaction) holder.run("rollback") + holder.close() + await rm(dir, { recursive: true, force: true }) + } +}) + +test("retries locked transaction begins cooperatively once the holding transaction releases", async () => { + const dir = await mkdtemp(join(tmpdir(), "effect-drizzle-sqlite-")) + const filename = join(dir, "locked.db") + const holder = new Database(filename) + + try { + holder.run("create table users (id integer primary key autoincrement, name text not null)") + holder.run("pragma busy_timeout = 0") + holder.run("begin immediate") + + await Effect.runPromise( + Effect.gen(function* () { + const db = yield* EffectDrizzleSqlite.makeWithDefaults() + yield* db.run(sql`pragma busy_timeout = 0`) + + const done = yield* Deferred.make() + const transaction = yield* db + .transaction((tx) => tx.insert(users).values({ name: "Ada" }), { behavior: "immediate" }) + .pipe(Effect.ensuring(Deferred.succeed(done, undefined)), Effect.forkChild) + yield* Effect.yieldNow + + // The BEGIN IMMEDIATE fails fast and the statement-level retry re-runs + // it until the holder releases the write lock. + expect(yield* Deferred.isDone(done)).toBe(false) + + holder.run("rollback") + yield* Fiber.join(transaction) + expect(yield* db.select().from(users)).toEqual([{ id: 1, name: "Ada" }]) + }).pipe(Effect.provide(sqliteLayer({ filename, disableWAL: true })), Effect.scoped), + ) + } finally { + if (holder.inTransaction) holder.run("rollback") + holder.close() + await rm(dir, { recursive: true, force: true }) + } +}) + +test("reports LockTimeoutError once the bounded statement retry is exhausted", async () => { + const dir = await mkdtemp(join(tmpdir(), "effect-drizzle-sqlite-")) + const filename = join(dir, "locked.db") + const holder = new Database(filename) + + try { + holder.run("create table users (id integer primary key autoincrement, name text not null)") + holder.run("pragma busy_timeout = 0") + holder.run("begin immediate") + + await Effect.runPromise( + Effect.gen(function* () { + const client = yield* SqlClientService + const error = yield* client + .unsafe("insert into users (name) values (?)", ["Blocked"]) + .values.pipe(Effect.flip) + + if (!isSqlError(error)) throw new Error("Expected SqlError") + expect(error.reason._tag).toBe("LockTimeoutError") + }).pipe(Effect.provide(sqliteLayer({ filename, disableWAL: true })), Effect.scoped), + ) + } finally { + if (holder.inTransaction) holder.run("rollback") + holder.close() + await rm(dir, { recursive: true, force: true }) + } +}) + test("supports returning and rejects empty update sets", async () => { await run( Effect.gen(function* () { diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index f7531621e561..0a597f00aaab 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, test } from "bun:test" +import { describe, expect, test } from "bun:test" import { AIError, LLMEvent, @@ -3820,6 +3820,98 @@ describe("SessionRunnerLLM", () => { }), ) + it.effect("waits for unrelated database transactions before interrupted settlement", () => + Effect.gen(function* () { + const session = yield* setup + const { db } = yield* Database.Service + const transactionStarted = yield* Deferred.make() + const releaseTransaction = yield* Deferred.make() + const interruptSettled = yield* Deferred.make() + yield* admit(session, "Interrupt during database contention") + const stream = yield* TestLLM.gate + + const run = yield* session.resume(sessionID).pipe(Effect.forkChild) + yield* stream.started + const transaction = yield* db + .transaction(() => + Deferred.succeed(transactionStarted, undefined).pipe(Effect.andThen(Deferred.await(releaseTransaction))), + ) + .pipe(Effect.forkChild) + yield* Deferred.await(transactionStarted) + const interrupt = yield* session + .interrupt(sessionID) + .pipe(Effect.ensuring(Deferred.succeed(interruptSettled, undefined)), Effect.forkChild) + yield* Effect.yieldNow + + expect(yield* Deferred.isDone(interruptSettled)).toBe(false) + + yield* Deferred.succeed(releaseTransaction, undefined) + yield* Fiber.join(transaction) + yield* Fiber.join(interrupt) + expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" }) + expect(requireAssistant(yield* session.context(sessionID))).toMatchObject({ + finish: "error", + error: { type: "aborted", message: "Step interrupted" }, + }) + }), + ) + + it.effect("queues interrupted settlement behind an unrelated transaction", () => + Effect.gen(function* () { + const session = yield* setup + const { db } = yield* Database.Service + const transactionStarted = yield* Deferred.make() + const interruptSettled = yield* Deferred.make() + yield* admit(session, "Interrupt during long transaction") + const stream = yield* TestLLM.gate + + const run = yield* session.resume(sessionID).pipe(Effect.forkChild) + yield* stream.started + // An unrelated session's work holds the shared SQLite transaction permit for a + // deterministic amount of work; the interrupted step's terminal publications + // queue behind it instead of settling immediately. + const transaction = yield* db + .transaction(() => + Effect.gen(function* () { + yield* Deferred.succeed(transactionStarted, undefined) + for (let index = 0; index < 3000; index++) { + const id = Session.ID.make(`ses_contention_${index}`) + yield* db + .insert(SessionTable) + .values({ + id, + project_id: Project.ID.global, + slug: id, + directory: "/project", + title: "test", + version: "test", + }) + .onConflictDoNothing() + .run() + .pipe(Effect.orDie) + } + }), + ) + .pipe(Effect.forkChild) + yield* Deferred.await(transactionStarted) + const interrupt = yield* session + .interrupt(sessionID) + .pipe(Effect.ensuring(Deferred.succeed(interruptSettled, undefined)), Effect.forkChild) + yield* Effect.yieldNow + + expect(yield* Deferred.isDone(interruptSettled)).toBe(false) + + yield* Fiber.join(transaction) + yield* Fiber.join(interrupt) + expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" }) + expect(requireAssistant(yield* session.context(sessionID))).toMatchObject({ + finish: "error", + error: { type: "aborted", message: "Step interrupted" }, + }) + }), + ) + + it.effect("durably fails blocked local tools when interrupted while awaiting settlement", () => Effect.gen(function* () { const session = yield* setup @@ -5111,3 +5203,8 @@ describe("SessionRunnerLLM", () => { }), ) }) + + + + +