diff --git a/CHANGESET.md b/CHANGESET.md index 716d9f591645..e45dfd8b8f21 100644 --- a/CHANGESET.md +++ b/CHANGESET.md @@ -69,6 +69,12 @@ Newest first. Each entry names the upstream range brought in. ### Runtime fixes +- **#155** Restore the SQLite lock-wait window that #88 cut to about 250ms (swxtchio/swx-opencode#153): every + runtime client keeps its short native attempt but retries a busy statement asynchronously for 5s of elapsed + contention, the old `busy_timeout` window, so a competing writer on the shared store no longer fails accepted + prompts and the event loop keeps turning. A stale-snapshot write (`SQLITE_BUSY_SNAPSHOT`), which no statement retry + can recover, now fails at once instead of holding the connection for the window (swxtchio/swx-opencode#154). + _Fork-only._ - **#144** Make `opencode db path` print the selected path without initializing `AppRuntime` or SQLite, defer global directory creation to runtime initialization, and preserve explicit and channel-specific database path selection. _Upstreamable._ - **#141** Reconcile `.git/HEAD` after an early or missed native callback through a shared read/change/publish path; keep a closed native watch `unavailable` while polling still delivers changed branch values. Cover root-off switches, native diff --git a/packages/core/src/database/sqlite.bun.ts b/packages/core/src/database/sqlite.bun.ts index ef2f94b341ba..c7bb9f11f1cb 100644 --- a/packages/core/src/database/sqlite.bun.ts +++ b/packages/core/src/database/sqlite.bun.ts @@ -19,12 +19,23 @@ import { Sqlite } from "./sqlite" const ATTR_DB_SYSTEM_NAME = "db.system.name" const nativeBusyTimeoutMs = 5 -// Five 5ms native attempts plus four 25/50/75/75ms backoffs target ~250ms total. -const retrySchedule = Schedule.exponential("25 millis").pipe( - Schedule.modifyDelay((_output, delay) => Effect.succeed(Duration.millis(Math.min(Duration.toMillis(delay), 75)))), - Schedule.take(4), +// Restore the lock tolerance the old busy_timeout gave, but wait asynchronously between short native attempts +// so a contended statement never blocks the event loop. The window runs from the first failed retry, and the +// backoff cap matches SQLite's own busy-handler sleep. +const retrySchedule = Schedule.exponential("10 millis").pipe( + Schedule.modifyDelay((_output, delay) => Effect.succeed(Duration.millis(Math.min(Duration.toMillis(delay), 100)))), + Schedule.jittered, + Schedule.both(Schedule.during("5 seconds")), ) +// A stale read snapshot fails every retry of the same statement; only restarting its transaction can recover. +const retryable = (error: SqlError) => { + const cause = error.reason.cause + const snapshot = + typeof cause === "object" && cause !== null && "code" in cause && cause.code === "SQLITE_BUSY_SNAPSHOT" + return error.reason.isRetryable && !snapshot +} + const statementError = (cause: unknown) => { const reason = classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }) if (reason._tag !== "LockTimeoutError") return new SqlError({ reason }) @@ -85,10 +96,14 @@ const make = (options: Config) => const statement = native.query(query) // @ts-ignore bun-types missing safeIntegers method, fixed in https://github.com/oven-sh/bun/pull/26627 statement.safeIntegers(Context.get(fiber.context, Client.SafeIntegers)) - return Effect.try({ + const execute = Effect.try({ try: () => (statement.all(...(params as any)) ?? []) as Array>, catch: statementError, - }).pipe(Effect.retry({ schedule: retrySchedule, while: (error) => error.reason.isRetryable })) + }) + // Only a statement that met the lock builds the retry schedule, which re-attempts it at once before backing off. + return execute.pipe( + Effect.catchIf(retryable, () => Effect.retry(execute, { schedule: retrySchedule, while: retryable })), + ) }) const runValues = (query: string, params: ReadonlyArray = []) => @@ -96,10 +111,14 @@ const make = (options: Config) => const statement = native.query(query) // @ts-ignore bun-types missing safeIntegers method, fixed in https://github.com/oven-sh/bun/pull/26627 statement.safeIntegers(Context.get(fiber.context, Client.SafeIntegers)) - return Effect.try({ + const execute = Effect.try({ try: () => (statement.values(...(params as any)) ?? []) as Array, catch: statementError, - }).pipe(Effect.retry({ schedule: retrySchedule, while: (error) => error.reason.isRetryable })) + }) + // Only a statement that met the lock builds the retry schedule, which re-attempts it at once before backing off. + return execute.pipe( + Effect.catchIf(retryable, () => Effect.retry(execute, { schedule: retrySchedule, while: retryable })), + ) }) const connection = identity({ diff --git a/packages/core/src/database/sqlite.node.ts b/packages/core/src/database/sqlite.node.ts index 796cebf612d7..5b3aadebd945 100644 --- a/packages/core/src/database/sqlite.node.ts +++ b/packages/core/src/database/sqlite.node.ts @@ -19,12 +19,24 @@ import { Sqlite } from "./sqlite" const ATTR_DB_SYSTEM_NAME = "db.system.name" const nativeBusyTimeoutMs = 5 -// Five 5ms native attempts plus four 25/50/75/75ms backoffs target ~250ms total. -const retrySchedule = Schedule.exponential("25 millis").pipe( - Schedule.modifyDelay((_output, delay) => Effect.succeed(Duration.millis(Math.min(Duration.toMillis(delay), 75)))), - Schedule.take(4), +const sqliteBusySnapshot = 517 +// Restore the lock tolerance the old busy_timeout gave, but wait asynchronously between short native attempts +// so a contended statement never blocks the event loop. The window runs from the first failed retry, and the +// backoff cap matches SQLite's own busy-handler sleep. +const retrySchedule = Schedule.exponential("10 millis").pipe( + Schedule.modifyDelay((_output, delay) => Effect.succeed(Duration.millis(Math.min(Duration.toMillis(delay), 100)))), + Schedule.jittered, + Schedule.both(Schedule.during("5 seconds")), ) +// A stale read snapshot fails every retry of the same statement; only restarting its transaction can recover. +const retryable = (error: SqlError) => { + const cause = error.reason.cause + const snapshot = + typeof cause === "object" && cause !== null && "errcode" in cause && cause.errcode === sqliteBusySnapshot + return error.reason.isRetryable && !snapshot +} + const statementError = (cause: unknown) => { const errcode = typeof cause === "object" && cause !== null && "errcode" in cause && typeof cause.errcode === "number" @@ -93,10 +105,14 @@ const make = (options: Config) => Effect.withFiber>, SqlError>((fiber) => { const statement = native.prepare(query) statement.setReadBigInts(Context.get(fiber.context, Client.SafeIntegers)) - return Effect.try({ + const execute = Effect.try({ try: () => statement.all(...(params as SQLInputValue[])) as Array>, catch: statementError, - }).pipe(Effect.retry({ schedule: retrySchedule, while: (error) => error.reason.isRetryable })) + }) + // Only a statement that met the lock builds the retry schedule, which re-attempts it at once before backing off. + return execute.pipe( + Effect.catchIf(retryable, () => Effect.retry(execute, { schedule: retrySchedule, while: retryable })), + ) }) const runValues = (query: string, params: ReadonlyArray = []) => @@ -104,10 +120,14 @@ const make = (options: Config) => const statement = native.prepare(query) statement.setReadBigInts(Context.get(fiber.context, Client.SafeIntegers)) statement.setReturnArrays(true) - return Effect.try({ + const execute = Effect.try({ try: () => statement.all(...(params as SQLInputValue[])) as unknown as ReadonlyArray>, catch: statementError, - }).pipe(Effect.retry({ schedule: retrySchedule, while: (error) => error.reason.isRetryable })) + }) + // Only a statement that met the lock builds the retry schedule, which re-attempts it at once before backing off. + return execute.pipe( + Effect.catchIf(retryable, () => Effect.retry(execute, { schedule: retrySchedule, while: retryable })), + ) }) const connection = identity({ diff --git a/packages/core/test/sqlite-busy.test.ts b/packages/core/test/sqlite-busy.test.ts index f08253e90b5b..2976706ad1b6 100644 --- a/packages/core/test/sqlite-busy.test.ts +++ b/packages/core/test/sqlite-busy.test.ts @@ -18,6 +18,37 @@ const nodeSqliteDrivers = await import("node:sqlite").then( () => undefined, ) const nodeSqliteTest = test.skipIf(!nodeSqliteDrivers) +// The retry window each statement must give a competing writer: the tolerance the old native busy_timeout gave. +const lockWindowMs = 5_000 +// Exhaustion may overrun the window by one capped backoff, one native attempt and scheduling delay. +const lockWindowSlackMs = 1_500 +// Backstops sit above window plus slack so a statement that never stops fails with its own message. +const lockBackstop = "10 seconds" +// Well past the point where the old four-attempt schedule gave up, and well inside the restored window. +const heldLockMs = 1_500 +// The event loop must keep turning while a statement waits; a long synchronous native wait stalls it for seconds. +const stallBoundMs = 1_000 +// A stale snapshot can never succeed on retry, so its write must fail far inside the retry window. +const snapshotBoundMs = 1_000 + +// Ticks continually from now on and returns a reader for the longest gap between ticks. The reader includes the gap +// still open when it is called, so a synchronous call the ticker has not yet woken from still counts. +const eventLoopStalls = Effect.gen(function* () { + let lastTick = performance.now() + let longestMs = 0 + yield* Effect.sleep("5 millis").pipe( + Effect.andThen( + Effect.sync(() => { + const now = performance.now() + longestMs = Math.max(longestMs, now - lastTick) + lastTick = now + }), + ), + Effect.forever, + Effect.forkScoped, + ) + return () => Math.max(longestMs, performance.now() - lastTick) +}) const sqliteError = (error: unknown) => { if (isSqlError(error)) return error @@ -149,21 +180,42 @@ const clientWithAttemptCount = (input: { attempts: { count: number } }) => Effect.gen(function* () { + // Clients prepare once and retry the execution, so count each execution of the prepared statement. + const counted = new WeakSet() + const countExecutions = (statement: S) => { + if (counted.has(statement)) return statement + counted.add(statement) + for (const method of ["all", "values"]) { + const execute: unknown = Reflect.get(statement, method) + if (typeof execute !== "function") continue + Object.defineProperty(statement, method, { + configurable: true, + writable: true, + value: (...params: unknown[]) => { + input.attempts.count++ + return execute.apply(statement, params) + }, + }) + } + return statement + } + if (input.clientType === "bun") { - const bun = yield* Effect.promise(() => import("../src/database/sqlite.bun")) - const context = yield* Layer.build(bun.layer({ filename: input.filename })) const sqlite = yield* Effect.promise(() => import("bun:sqlite")) - const native = Context.get(context, Sqlite.Native) as InstanceType - const query = native.query - Object.defineProperty(native, "query", { - configurable: true, - writable: true, - value: (sql: string) => { - if (sql === input.statement) input.attempts.count++ - return query.call(native, sql) + const prototype = sqlite.Database.prototype + const queryDescriptor = Object.getOwnPropertyDescriptor(prototype, "query") + if (!queryDescriptor) return yield* Effect.die(new Error("bun:sqlite query method was not found")) + const query = prototype.query + Object.defineProperty(prototype, "query", { + ...queryDescriptor, + value: function (this: InstanceType, sql: string) { + const statement = query.call(this, sql) + return sql === input.statement ? countExecutions(statement) : statement }, }) - return Context.get(context, SqlClient) + yield* Effect.addFinalizer(() => Effect.sync(() => Object.defineProperty(prototype, "query", queryDescriptor))) + const context = yield* Layer.build(Database.layerFromPath(input.filename)) + return Context.get(context, Database.Service).db.$client } const drivers = nodeSqliteDrivers @@ -176,8 +228,8 @@ const clientWithAttemptCount = (input: { configurable: true, writable: true, value: function (this: InstanceType, sql: string) { - if (sql === input.statement) input.attempts.count++ - return prepare.call(this, sql) + const statement = prepare.call(this, sql) + return sql === input.statement ? countExecutions(statement) : statement }, }) return Context.get(context, SqlClient) @@ -191,8 +243,8 @@ const clientWithAttemptCount = (input: { Object.defineProperty(prototype, "prepare", { ...prepareDescriptor, value: function (this: InstanceType, sql: string) { - if (sql === input.statement) input.attempts.count++ - return prepare.call(this, sql) + const statement = prepare.call(this, sql) + return sql === input.statement ? countExecutions(statement) : statement }, }) yield* Effect.addFinalizer(() => Effect.sync(() => Object.defineProperty(prototype, "prepare", prepareDescriptor))) @@ -229,14 +281,13 @@ describe("SQLite busy timeout and statement retries", () => { ) }) - for (const clientType of ["core", "standalone"] as const) { + for (const clientType of ["bun", "core-node", "standalone-node"] as const) { for (const method of ["run", "values"] as const) { - nodeSqliteTest( - `node:sqlite ${clientType} retries ${method} after a second process releases its write lock`, + test.skipIf(clientType !== "bun" && !nodeSqliteDrivers)( + `${clientType} ${method} waits out a write lock held past the old 250ms window`, async () => { - if (!nodeSqliteDrivers) return await using tmp = await tmpdir() - const filename = path.join(tmp.path, "node-busy.sqlite") + const filename = path.join(tmp.path, `${clientType}-held.sqlite`) const writerValue = `${clientType}-${method}-writer` const writerSql = method === "run" @@ -246,42 +297,59 @@ describe("SQLite busy timeout and statement retries", () => { Effect.scoped( Effect.gen(function* () { const attempts = { count: 0 } - const client = yield* clientWithAttemptCount({ - clientType: clientType === "core" ? "core-node" : "standalone-node", - filename, - statement: writerSql, - attempts, - }) + const client = yield* clientWithAttemptCount({ clientType, filename, statement: writerSql, attempts }) yield* client.unsafe("CREATE TABLE busy_retry_test (id INTEGER PRIMARY KEY, value TEXT NOT NULL)").raw - yield* client.unsafe("PRAGMA busy_timeout = 0").raw + expect(yield* client.unsafe("PRAGMA busy_timeout").values).toEqual([[5]]) + const server = Bun.serve({ port: 0, fetch: () => new Response("server-progress") }) + yield* Effect.addFinalizer(() => Effect.sync(() => server.stop(true))) + const holder = yield* Effect.promise(() => + startLockHolder(filename, clientType === "bun" ? holderScript : nodeHolderScript), + ) + yield* Effect.addFinalizer(() => Effect.promise(() => holder.release())) - const holder = yield* Effect.promise(() => startLockHolder(filename, nodeHolderScript)) + const longestStallMs = yield* eventLoopStalls + let writerFinished = false + const startedAt = performance.now() + const writer = yield* ( + method === "run" ? client.unsafe(writerSql).raw : client.unsafe(writerSql).values + ).pipe( + Effect.onExit(() => Effect.sync(() => void (writerFinished = true))), + Effect.forkChild({ startImmediately: true }), + ) yield* Effect.gen(function* () { - const statement = method === "run" ? client.unsafe(writerSql).raw : client.unsafe(writerSql).values - let writerFinished = false - const writer = yield* statement.pipe( - Effect.onExit(() => Effect.sync(() => void (writerFinished = true))), - Effect.forkChild({ startImmediately: true }), - ) + while (attempts.count === 0) yield* Effect.sleep("1 millis") + }).pipe( + Effect.timeoutOrElse({ + duration: "1 second", + orElse: () => Effect.fail(new Error("writer did not attempt the held-lock statement")), + }), + ) + // The holder owns the write lock and the writer has met it; keep holding well past the old schedule. + while (performance.now() - startedAt < heldLockMs) yield* Effect.sleep("10 millis") + expect(writerFinished).toBe(false) - yield* Effect.gen(function* () { - while (attempts.count === 0) yield* Effect.sleep("1 millis") - }).pipe( - Effect.timeoutOrElse({ - duration: "1 second", - orElse: () => Effect.fail(new Error("writer did not attempt the held-lock statement")), - }), - ) - expect(writerFinished).toBe(false) - yield* Effect.promise(() => holder.release()) - const result = yield* Fiber.await(writer) - expect(Exit.isSuccess(result)).toBe(true) - if (!Exit.isSuccess(result)) return - if (method === "values") expect(result.value).toEqual([[1]]) - expect( - yield* client.unsafe(`SELECT COUNT(*) FROM busy_retry_test WHERE value = '${writerValue}'`).values, - ).toEqual([[1]]) - }).pipe(Effect.ensuring(Effect.promise(() => holder.release()))) + yield* Effect.promise(() => holder.probe(server.url.href)).pipe( + Effect.timeoutOrElse({ + duration: "2 seconds", + orElse: () => + Effect.fail(new Error("unrelated request did not progress while SQLite was contended")), + }), + ) + expect(longestStallMs()).toBeLessThan(stallBoundMs) + expect(writerFinished).toBe(false) + + yield* Effect.promise(() => holder.release()) + const result = yield* Fiber.await(writer).pipe( + Effect.timeoutOrElse({ + duration: "2 seconds", + orElse: () => Effect.fail(new Error("writer did not finish after the lock was released")), + }), + ) + if (Exit.isFailure(result)) return yield* Effect.fail(new Error(Cause.pretty(result.cause))) + if (method === "values") expect(result.value).toEqual([[1]]) + expect( + yield* client.unsafe(`SELECT COUNT(*) FROM busy_retry_test WHERE value = '${writerValue}'`).values, + ).toEqual([[1]]) }), ), ) @@ -318,7 +386,7 @@ describe("SQLite busy timeout and statement retries", () => { .unsafe("INSERT INTO busy_retry_test (id, value) VALUES (1, 'node-writer')") .raw.pipe( Effect.timeoutOrElse({ - duration: "5 seconds", + duration: lockBackstop, orElse: () => Effect.fail(new Error("node retry did not stop")), }), ) @@ -334,7 +402,67 @@ describe("SQLite busy timeout and statement retries", () => { ) expect(Exit.isSuccess(exit)).toBe(true) }, - 10_000, + 20_000, + ) + } + + for (const clientType of ["bun", "core-node", "standalone-node"] as const) { + test.skipIf(clientType !== "bun" && !nodeSqliteDrivers)( + `${clientType} fails a stale-snapshot transaction write without retrying it`, + async () => { + await using tmp = await tmpdir() + const filename = path.join(tmp.path, `${clientType}-snapshot.sqlite`) + const writerSql = "INSERT INTO busy_retry_test (id, value) VALUES (1, 'stale-writer')" + const exit = await Effect.runPromiseExit( + Effect.scoped( + Effect.gen(function* () { + const attempts = { count: 0 } + const client = yield* clientWithAttemptCount({ clientType, filename, statement: writerSql, attempts }) + yield* client.unsafe("CREATE TABLE busy_retry_test (id INTEGER PRIMARY KEY, value TEXT NOT NULL)").raw + expect(yield* client.unsafe("PRAGMA journal_mode").values).toEqual([["wal"]]) + const sqlite = yield* Effect.promise(() => import("bun:sqlite")) + const other = new sqlite.Database(filename) + yield* Effect.addFinalizer(() => Effect.sync(() => other.close())) + + let elapsedMs = 0 + const failure = yield* client + .withTransaction( + Effect.gen(function* () { + expect(yield* client.unsafe("SELECT COUNT(*) FROM busy_retry_test").values).toEqual([[0]]) + // Another connection commits after this transaction's read, so its snapshot is stale. + other.run("INSERT INTO busy_retry_test (id, value) VALUES (2, 'other-writer')") + const startedAt = performance.now() + yield* client + .unsafe(writerSql) + .raw.pipe( + Effect.onExit(() => Effect.sync(() => void (elapsedMs = performance.now() - startedAt))), + ) + }), + ) + .pipe( + Effect.timeoutOrElse({ + duration: lockBackstop, + orElse: () => Effect.fail(new Error("stale-snapshot retries did not stop")), + }), + Effect.flip, + ) + expect(isSqlError(failure)).toBe(true) + if (!isSqlError(failure)) return + expect(failure.reason._tag).toBe("LockTimeoutError") + expect(failure.reason.cause).toMatchObject( + clientType === "bun" ? { code: "SQLITE_BUSY_SNAPSHOT" } : { errcode: 517 }, + ) + expect(elapsedMs).toBeLessThan(snapshotBoundMs) + expect(attempts.count).toBe(1) + expect(yield* client.unsafe("SELECT id, value FROM busy_retry_test ORDER BY id").values).toEqual([ + [2, "other-writer"], + ]) + }), + ), + ) + if (Exit.isFailure(exit)) throw new Error(Cause.pretty(exit.cause)) + }, + 20_000, ) } @@ -358,7 +486,7 @@ describe("SQLite busy timeout and statement retries", () => { }) .pipe( Effect.timeoutOrElse({ - duration: "5 seconds", + duration: lockBackstop, orElse: () => Effect.fail(new Error("immediate transaction retries did not stop")), }), ) @@ -374,75 +502,11 @@ describe("SQLite busy timeout and statement retries", () => { ), ) if (Exit.isFailure(exit)) throw new Error(Cause.pretty(exit.cause)) - }, 10_000) - - for (const method of ["run", "values"] as const) { - test(`retries ${method} after a second process releases its write lock`, async () => { - await using tmp = await tmpdir() - const filename = path.join(tmp.path, "busy.sqlite") - const writerSql = - method === "run" - ? "INSERT INTO busy_retry_test (id, value) VALUES (1, 'writer')" - : "INSERT INTO busy_retry_test (id, value) VALUES (1, 'writer') RETURNING id" - const attempts = { count: 0 } - const exit = await Effect.runPromiseExit( - Effect.scoped( - Effect.gen(function* () { - const client = yield* clientWithAttemptCount({ - clientType: "bun", - filename, - statement: writerSql, - attempts, - }) - yield* client.unsafe("CREATE TABLE busy_retry_test (id INTEGER PRIMARY KEY, value TEXT NOT NULL)").raw - yield* client.unsafe("PRAGMA busy_timeout = 0").raw - - const holder = yield* Effect.promise(() => startLockHolder(filename, bunHolderScript)) - yield* Effect.gen(function* () { - const statement = Effect.gen(function* () { - if (method === "run") { - yield* client.unsafe(writerSql).raw - return { method: "run" as const } - } - return { - method: "values" as const, - rows: yield* client.unsafe(writerSql).values, - } - }) - let writerFinished = false - const writer = yield* statement.pipe( - Effect.onExit(() => Effect.sync(() => void (writerFinished = true))), - Effect.forkChild({ startImmediately: true }), - ) - - yield* Effect.gen(function* () { - while (attempts.count === 0) yield* Effect.sleep("1 millis") - }).pipe( - Effect.timeoutOrElse({ - duration: "1 second", - orElse: () => Effect.fail(new Error("writer did not attempt the held-lock statement")), - }), - ) - expect(writerFinished).toBe(false) - yield* Effect.promise(() => holder.release()) - const result = yield* Fiber.await(writer) - expect(Exit.isSuccess(result)).toBe(true) - if (!Exit.isSuccess(result)) return - if (result.value.method === "values") expect(result.value.rows).toEqual([[1]]) - expect( - yield* client.unsafe("SELECT COUNT(*) FROM busy_retry_test WHERE value = 'writer'").values, - ).toEqual([[1]]) - }).pipe(Effect.ensuring(Effect.promise(() => holder.release()))) - }), - ), - ) - expect(Exit.isSuccess(exit)).toBe(true) - }, 10_000) - } + }, 20_000) for (const clientType of ["bun", "core-node", "standalone-node"] as const) { - nodeSqliteTest( - `${clientType} serves unrelated requests and exhausts a persistent lock within budget`, + test.skipIf(clientType !== "bun" && !nodeSqliteDrivers)( + `${clientType} serves unrelated requests and exhausts a persistent lock after the retry window`, async () => { const drivers = nodeSqliteDrivers if (clientType !== "bun" && !drivers) return @@ -473,6 +537,7 @@ describe("SQLite busy timeout and statement retries", () => { ) yield* Effect.addFinalizer(() => Effect.promise(() => holder.release())) + const longestStallMs = yield* eventLoopStalls let writerFinished = false let elapsedMs = 0 const startedAt = performance.now() @@ -494,10 +559,17 @@ describe("SQLite busy timeout and statement retries", () => { }), ) expect(writerFinished).toBe(false) - const result = yield* Fiber.await(writer) + const result = yield* Fiber.await(writer).pipe( + Effect.timeoutOrElse({ + duration: lockBackstop, + orElse: () => Effect.fail(new Error("persistent-lock retries did not stop")), + }), + ) expect(writerFinished).toBe(true) expect(Exit.isFailure(result)).toBe(true) - expect(elapsedMs).toBeLessThan(1_000) + expect(elapsedMs).toBeGreaterThanOrEqual(lockWindowMs) + expect(elapsedMs).toBeLessThan(lockWindowMs + lockWindowSlackMs) + expect(longestStallMs()).toBeLessThan(stallBoundMs) if (!Exit.isFailure(result)) return const error = Option.getOrUndefined(Cause.findErrorOption(result.cause)) expect(isSqlError(error)).toBe(true) @@ -509,12 +581,17 @@ describe("SQLite busy timeout and statement retries", () => { expect( yield* client.unsafe("SELECT COUNT(*) FROM busy_retry_test WHERE value = 'persistent-writer'").values, ).toEqual([[0]]) + yield* client.unsafe("INSERT INTO busy_retry_test (id, value) VALUES (2, 'independent-writer')").raw + expect(yield* client.unsafe("SELECT id, value FROM busy_retry_test ORDER BY id").values).toEqual([ + [2, "independent-writer"], + [100, "holder"], + ]) }), ), ) if (Exit.isFailure(exit)) throw new Error(Cause.pretty(exit.cause)) }, - 10_000, + 20_000, ) } @@ -689,10 +766,11 @@ describe("SQLite busy timeout and statement retries", () => { yield* Effect.gen(function* () { let failure: unknown + const startedAt = performance.now() if (method === "run") { yield* database.run("INSERT INTO busy_retry_test (id, value) VALUES (1, 'writer')").pipe( Effect.timeoutOrElse({ - duration: "5 seconds", + duration: lockBackstop, orElse: () => Effect.fail(new Error("retry did not stop")), }), Effect.catch((error) => Effect.sync(() => (failure = error))), @@ -702,7 +780,7 @@ describe("SQLite busy timeout and statement retries", () => { .values("INSERT INTO busy_retry_test (id, value) VALUES (1, 'writer') RETURNING id") .pipe( Effect.timeoutOrElse({ - duration: "5 seconds", + duration: lockBackstop, orElse: () => Effect.fail(new Error("retry did not stop")), }), Effect.catch((error) => Effect.sync(() => (failure = error))), @@ -710,7 +788,10 @@ describe("SQLite busy timeout and statement retries", () => { } expect(failure).toBeInstanceOf(EffectDrizzleQueryError) if (!(failure instanceof EffectDrizzleQueryError)) return - expect(attempts.lock).toBe(5) + const elapsedMs = performance.now() - startedAt + expect(elapsedMs).toBeGreaterThanOrEqual(lockWindowMs) + expect(elapsedMs).toBeLessThan(lockWindowMs + lockWindowSlackMs) + expect(attempts.lock).toBeGreaterThan(1) expect(failure.query).toBe("Database is locked (SQLITE_BUSY)") expect(failure.params).toEqual([]) expect(failure.message).toContain("Database is locked (SQLITE_BUSY)") @@ -752,7 +833,7 @@ describe("SQLite busy timeout and statement retries", () => { ), ) if (Exit.isFailure(exit)) throw new Error(Cause.pretty(exit.cause)) - }, 10_000) + }, 20_000) } for (const method of ["run", "values"] as const) { diff --git a/packages/effect-sqlite-node/src/index.ts b/packages/effect-sqlite-node/src/index.ts index 28bff2d5acd4..55f4d2410d27 100644 --- a/packages/effect-sqlite-node/src/index.ts +++ b/packages/effect-sqlite-node/src/index.ts @@ -19,12 +19,24 @@ import * as Statement from "effect/unstable/sql/Statement" const ATTR_DB_SYSTEM_NAME = "db.system.name" const nativeBusyTimeoutMs = 5 -// Five 5ms native attempts plus four 25/50/75/75ms backoffs target ~250ms total. -const retrySchedule = Schedule.exponential("25 millis").pipe( - Schedule.modifyDelay((_output, delay) => Effect.succeed(Duration.millis(Math.min(Duration.toMillis(delay), 75)))), - Schedule.take(4), +const sqliteBusySnapshot = 517 +// Restore the lock tolerance the old busy_timeout gave, but wait asynchronously between short native attempts +// so a contended statement never blocks the event loop. The window runs from the first failed retry, and the +// backoff cap matches SQLite's own busy-handler sleep. +const retrySchedule = Schedule.exponential("10 millis").pipe( + Schedule.modifyDelay((_output, delay) => Effect.succeed(Duration.millis(Math.min(Duration.toMillis(delay), 100)))), + Schedule.jittered, + Schedule.both(Schedule.during("5 seconds")), ) +// A stale read snapshot fails every retry of the same statement; only restarting its transaction can recover. +const retryable = (error: SqlError) => { + const cause = error.reason.cause + const snapshot = + typeof cause === "object" && cause !== null && "errcode" in cause && cause.errcode === sqliteBusySnapshot + return error.reason.isRetryable && !snapshot +} + const statementError = (cause: unknown) => { const errcode = typeof cause === "object" && cause !== null && "errcode" in cause && typeof cause.errcode === "number" @@ -109,10 +121,14 @@ export const make = ( Effect.withFiber>, SqlError>((fiber) => { const statement = db.prepare(sql) statement.setReadBigInts(Context.get(fiber.context, Client.SafeIntegers)) - return Effect.try({ + const execute = Effect.try({ try: () => statement.all(...(params as SQLInputValue[])) as Array>, catch: statementError, - }).pipe(Effect.retry({ schedule: retrySchedule, while: (error) => error.reason.isRetryable })) + }) + // Only a statement that met the lock builds the retry schedule, which re-attempts it at once before backing off. + return execute.pipe( + Effect.catchIf(retryable, () => Effect.retry(execute, { schedule: retrySchedule, while: retryable })), + ) }) const runValues = (sql: string, params: ReadonlyArray = []) => @@ -120,11 +136,15 @@ export const make = ( const statement = db.prepare(sql) statement.setReadBigInts(Context.get(fiber.context, Client.SafeIntegers)) statement.setReturnArrays(true) - return Effect.try({ + const execute = Effect.try({ try: () => statement.all(...(params as SQLInputValue[])) as unknown as ReadonlyArray>, catch: statementError, - }).pipe(Effect.retry({ schedule: retrySchedule, while: (error) => error.reason.isRetryable })) + }) + // Only a statement that met the lock builds the retry schedule, which re-attempts it at once before backing off. + return execute.pipe( + Effect.catchIf(retryable, () => Effect.retry(execute, { schedule: retrySchedule, while: retryable })), + ) }) return identity({ diff --git a/packages/opencode/test/fixture/sqlite-lock.ts b/packages/opencode/test/fixture/sqlite-lock.ts index 737a260f68eb..f0010becce18 100644 --- a/packages/opencode/test/fixture/sqlite-lock.ts +++ b/packages/opencode/test/fixture/sqlite-lock.ts @@ -3,6 +3,9 @@ import { sql } from "drizzle-orm" import { Context, Effect, Layer } from "effect" import { SqlClient } from "effect/unstable/sql/SqlClient" +// Outlasts the driver's own retry window, so it only catches a retry that never stops. +const retryBackstop = "10 seconds" + const withImmediateSqliteLock = ( database: Database.Interface["db"], filename: string, @@ -34,7 +37,7 @@ export const produceSqliteBusyError = (database: Database.Interface["db"], filen Effect.flip( database.run(sql`INSERT INTO secret_lock_fixture (id, value) VALUES (${1}, ${"secret_parameter"})`).pipe( Effect.timeoutOrElse({ - duration: "5 seconds", + duration: retryBackstop, orElse: () => Effect.fail(new Error("SQLite busy retries did not stop")), }), ), @@ -51,7 +54,7 @@ export const produceSqliteImmediateError = (database: Database.Interface["db"], ) .pipe( Effect.timeoutOrElse({ - duration: "5 seconds", + duration: retryBackstop, orElse: () => Effect.fail(new Error("SQLite immediate transaction retries did not stop")), }), ), @@ -78,7 +81,7 @@ export const produceSqliteLockedError = (filename: string) => return yield* Effect.flip( client.unsafe("DROP TABLE secret_lock_fixture").raw.pipe( Effect.timeoutOrElse({ - duration: "5 seconds", + duration: retryBackstop, orElse: () => Effect.fail(new Error("SQLite locked retries did not stop")), }), ), diff --git a/packages/opencode/test/server/httpapi-error-middleware.test.ts b/packages/opencode/test/server/httpapi-error-middleware.test.ts index 1a14fde57c81..ad32665a650c 100644 --- a/packages/opencode/test/server/httpapi-error-middleware.test.ts +++ b/packages/opencode/test/server/httpapi-error-middleware.test.ts @@ -111,45 +111,48 @@ describe("HttpApi error middleware", () => { 20_000, ) - it.live("maps a produced direct SQLITE_LOCKED error distinctly", () => - Effect.gen(function* () { - const tmp = yield* Effect.promise(() => tmpdir()) - yield* Effect.addFinalizer(() => Effect.promise(() => tmp[Symbol.asyncDispose]())) - const error = yield* produceSqliteLockedError(path.join(tmp.path, "sqlite-locked.sqlite")) - expect(error).not.toBeInstanceOf(EffectDrizzleQueryError) - expect(isSqlError(error)).toBe(true) - if (!isSqlError(error)) return - expect(error.reason._tag).toBe("LockTimeoutError") - expect(error.reason.cause).toMatchObject({ code: "SQLITE_LOCKED" }) - expect(error.message).not.toContain("secret_lock_fixture") - expect(error.message).not.toContain("secret_parameter") - - const converted = MessageV2.fromError(error, { providerID: ProviderV2.ID.make("test") }) - expect(converted).toMatchObject({ - name: "UnknownError", - data: { message: "Database is locked (SQLITE_LOCKED)" }, - }) - expect(JSON.stringify(converted)).not.toContain("secret_lock_fixture") - expect(JSON.stringify(converted)).not.toContain("secret_parameter") + it.live( + "maps a produced direct SQLITE_LOCKED error distinctly", + () => + Effect.gen(function* () { + const tmp = yield* Effect.promise(() => tmpdir()) + yield* Effect.addFinalizer(() => Effect.promise(() => tmp[Symbol.asyncDispose]())) + const error = yield* produceSqliteLockedError(path.join(tmp.path, "sqlite-locked.sqlite")) + expect(error).not.toBeInstanceOf(EffectDrizzleQueryError) + expect(isSqlError(error)).toBe(true) + if (!isSqlError(error)) return + expect(error.reason._tag).toBe("LockTimeoutError") + expect(error.reason.cause).toMatchObject({ code: "SQLITE_LOCKED" }) + expect(error.message).not.toContain("secret_lock_fixture") + expect(error.message).not.toContain("secret_parameter") - yield* HttpRouter.add("GET", "/sqlite-locked", Effect.die(error)).pipe( - Layer.provide(errorLayer), - HttpRouter.serve, - Layer.build, - ) - const response = yield* HttpClientRequest.get("/sqlite-locked").pipe(HttpClient.execute) - const body = yield* response.json - const serialized = JSON.stringify(body) + const converted = MessageV2.fromError(error, { providerID: ProviderV2.ID.make("test") }) + expect(converted).toMatchObject({ + name: "UnknownError", + data: { message: "Database is locked (SQLITE_LOCKED)" }, + }) + expect(JSON.stringify(converted)).not.toContain("secret_lock_fixture") + expect(JSON.stringify(converted)).not.toContain("secret_parameter") - expect(response.status).toBe(500) - expect(body).toMatchObject({ - name: "UnknownError", - data: { message: "Database is locked (SQLITE_LOCKED)" }, - }) - expect((body as { data?: { ref?: unknown } }).data?.ref).toMatch(/^err_[0-9a-f-]{8}$/) - expect(serialized).not.toContain("secret_lock_fixture") - expect(serialized).not.toContain("secret_parameter") - }), + yield* HttpRouter.add("GET", "/sqlite-locked", Effect.die(error)).pipe( + Layer.provide(errorLayer), + HttpRouter.serve, + Layer.build, + ) + const response = yield* HttpClientRequest.get("/sqlite-locked").pipe(HttpClient.execute) + const body = yield* response.json + const serialized = JSON.stringify(body) + + expect(response.status).toBe(500) + expect(body).toMatchObject({ + name: "UnknownError", + data: { message: "Database is locked (SQLITE_LOCKED)" }, + }) + expect((body as { data?: { ref?: unknown } }).data?.ref).toMatch(/^err_[0-9a-f-]{8}$/) + expect(serialized).not.toContain("secret_lock_fixture") + expect(serialized).not.toContain("secret_parameter") + }), + 20_000, ) it.live("returns invalid config defects as structured client errors", () => diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index 6b2f07d32ce5..7e49dbf58757 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -1790,7 +1790,7 @@ itHeldLock.live( yield* waitFor( Effect.sync(() => (progress.length >= 2 ? true : undefined)), "the refused terminal write reported no progress", - 30_000, + 120_000, ) expect(progress.slice(0, 2)).toEqual([ { "session.id": chat.id, messageID: msg.id, attempts: 4 }, @@ -1813,7 +1813,7 @@ itHeldLock.live( }), { config: cfg }, ), - 60_000, + 180_000, ) itHeldLock.live(