From ef663e2de6b8a58a6e98164b4e870e2a2de111a9 Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 03:30:28 +0000 Subject: [PATCH 01/18] docs(core): plan the SQLite snapshot-loss fix for #162 Plan for fixing forward accepted-prompt loss under concurrent writers on top of the reverted base (60f779b64b). Evidence: data/opencode-fixed-binary-155/ report.md sections 5 and 6, and data/opencode-revert-155/report.md section 7. Mechanism, from the code (not yet measured): - The admission transaction is SessionQueue writable() in packages/opencode/src/session/queue.ts. It reads first (EventV2.assertWritableIn) and then writes (sequence upsert and queue insert). It begins as a deferred transaction, so its read snapshot exists before its write lock is taken. - A plain-busy wait inside a deferred transaction can only end in SQLITE_BUSY_SNAPSHOT when another connection commits during the wait: the snapshot goes stale, and the driver does not restart the transaction. This is the mechanism that predicts the report's section 6.3 observation that the snapshot class grew when the wait was added. It is not measured yet. - The admission and promotion writes are the deferred read-then-write transactions found on the accepted-prompt path by grepping its call sites. The EventV2 append that persists the user message already begins with behavior immediate (event.ts:467). The proof decides whether that is complete. Correction to #162. Its option 2 names Client.make beginTransaction, which the production path never reaches. opencode runs transactions through effect-drizzle-sqlite, whose withTransaction issues "begin " itself (packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts), default deferred. Effect's SqlClient.withTransaction, which reads beginTransaction, has no caller in opencode. The mechanism that works is the existing per-transaction option, behavior immediate, which the codebase already uses in event.ts and project.ts. Choice: (a) BEGIN IMMEDIATE on write-capable transactions, not (b) restart. - (a) takes the write lock before any read snapshot exists, so the failing class cannot occur for that transaction. The lock wait happens at BEGIN, where a retry is safe because nothing has been read yet. - (b) restarts a body that has already run. The admission body is pure SQL, but restart needs a per-site audit of every transaction body for duplicated side effects, and it leaves the stale-snapshot race in place. Steps: 1. Restore the #155 plain-busy wait and the #154 fail-fast on SQLITE_BUSY_SNAPSHOT in packages/core/src/database/sqlite.bun.ts, sqlite.node.ts and packages/effect-sqlite-node/src/index.ts. The three copies stay identical, as #162 requires. Dedupe stays with #165. sqlite.node.ts is outside the brief's named scope; it is named here because it is the third copy of the same rule. 2. Begin writable() with behavior immediate in queue.ts. This covers every queue write: the callers are admit, reserve, withdraw, restore, and update. 3. Restore the #155 test versions that cover the wait and the fail-fast: packages/core/test/sqlite-busy.test.ts, and the opencode lock fixture and consumer bounds, which #155 raised for the 5 s window. 4. Add a regression at the admission path. A second connection holds the write lock across the admission read, then commits. The admission must succeed and persist its row. Red proof: revert writable() to deferred. Also run the restored core tests against the reverted driver for red. 5. Proofs, in the brief's order: focused suites with CI commands; the candidate against the reverted parent, concurrently on separate DBs, same load, several pairs, raw counts; fresh-store proof. The parent is the reverted build (tree equal to bf5180afa0). Not in this change: session queue consumption (#167); the CHANGESET entry (needs the PR number at publication); the snapshot-tracking PATH issue (#164). Claude-Session: https://claude.ai/code/session_01LrS7BTRTnc6eYrSxBFoWAo From e6461df50abc73da69502099babef0a88d6cb199 Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 03:37:20 +0000 Subject: [PATCH 02/18] fix(core): restore the SQLite lock-wait window for busy statements Restores the plain-busy wait that #155 added. A statement that meets a lock retries asynchronously for 5 seconds of elapsed contention. A stale-snapshot failure (SQLITE_BUSY_SNAPSHOT) is still not retried (#154). The three retry copies stay identical in sqlite.bun.ts, sqlite.node.ts and effect-sqlite-node. Red on the reverted driver: the restored sqlite-busy suite fails 14 of 23. The 250 ms window cannot absorb a 1.5 s hold, and a stale-snapshot statement is executed more than once. Green with the window: 23 of 23. The opencode lock-consumer suites pass 36 of 36 with their #155 bounds. Refs swxtchio/swx-opencode#162 Claude-Session: https://claude.ai/code/session_01LrS7BTRTnc6eYrSxBFoWAo --- packages/core/src/database/sqlite.bun.ts | 35 +- packages/core/src/database/sqlite.node.ts | 36 +- packages/core/test/sqlite-busy.test.ts | 339 +++++++++++------- packages/effect-sqlite-node/src/index.ts | 36 +- packages/opencode/test/fixture/sqlite-lock.ts | 9 +- .../server/httpapi-error-middleware.test.ts | 77 ++-- .../test/session/processor-effect.test.ts | 4 +- 7 files changed, 341 insertions(+), 195 deletions(-) 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( From 1948f9d0f95416e4a62c8ff7dec0e2d722584eb6 Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 03:37:25 +0000 Subject: [PATCH 03/18] fix(opencode): admit queue writes under BEGIN IMMEDIATE SessionQueue writable() runs its read (EventV2.assertWritableIn) and then its write in one transaction. Begun deferred, that transaction pins a read snapshot before it takes the write lock. If another writer commits while the write waits on the lock, the snapshot goes stale, and the write fails with SQLITE_BUSY_SNAPSHOT, which is not retried. The prompt was already accepted (204), so it was lost with no user message. With behavior immediate the write lock is taken at BEGIN, before any read. The lock wait then happens where a retry is safe, and no snapshot can go stale inside the transaction. The restored lock-wait window absorbs the wait. The regression holds the write lock from a second connection across the admission read, then commits. Red with the deferred writable: the sequence upsert fails with SQLITE_BUSY_SNAPSHOT after the commit. Green with immediate. Refs swxtchio/swx-opencode#162 Claude-Session: https://claude.ai/code/session_01LrS7BTRTnc6eYrSxBFoWAo --- packages/opencode/src/session/queue.ts | 15 ++- .../test/session/queue-admission.test.ts | 97 +++++++++++++++++++ 2 files changed, 107 insertions(+), 5 deletions(-) create mode 100644 packages/opencode/test/session/queue-admission.test.ts diff --git a/packages/opencode/src/session/queue.ts b/packages/opencode/src/session/queue.ts index 033d13362056..a1a6f60c0b3e 100644 --- a/packages/opencode/src/session/queue.ts +++ b/packages/opencode/src/session/queue.ts @@ -164,12 +164,17 @@ const layer = Layer.effect( return next } + // The write lock is taken at BEGIN. A deferred transaction would pin a read + // snapshot in assertWritableIn, and a commit by another writer while its + // write waits on the lock would leave that snapshot stale. const writable = (sessionID: SessionID, effect: (tx: DatabaseTransaction) => Effect.Effect) => - db.transaction((tx) => - Effect.gen(function* () { - yield* EventV2.assertWritableIn(tx, sessionID) - return yield* effect(tx) - }), + db.transaction( + (tx) => + Effect.gen(function* () { + yield* EventV2.assertWritableIn(tx, sessionID) + return yield* effect(tx) + }), + { behavior: "immediate" }, ) const exclusive: Interface["exclusive"] = (sessionID, effect) => semaphore(locks, sessionID).withPermit(effect) diff --git a/packages/opencode/test/session/queue-admission.test.ts b/packages/opencode/test/session/queue-admission.test.ts new file mode 100644 index 000000000000..18de49d26edc --- /dev/null +++ b/packages/opencode/test/session/queue-admission.test.ts @@ -0,0 +1,97 @@ +import { afterAll, expect } from "bun:test" +import { randomUUID } from "node:crypto" +import { rm } from "node:fs/promises" +import os from "node:os" +import path from "node:path" +import { eq } from "drizzle-orm" +import { Cause, Effect, Exit, Fiber, Layer } from "effect" +import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { SessionPromptQueueTable } from "@opencode-ai/core/session/prompt-queue.sql" +import { Session } from "@/session/session" +import { SessionQueue } from "@/session/queue" +import { EventV2Bridge } from "@/event-v2-bridge" +import { RuntimeFlags } from "@/effect/runtime-flags" +import { InstanceStore } from "@/project/instance-store" +import { InstanceBootstrap } from "@/project/bootstrap" +import { testEffect } from "../lib/effect" + +// A file-backed store, so a second connection can hold the write lock on the +// database the queue writes to. The file is removed after the suite. +const databasePath = path.join(os.tmpdir(), `opencode-queue-admission-${randomUUID()}.db`) + +const it = testEffect( + AppNodeBuilder.build( + LayerNode.group([ + Session.node, + SessionQueue.node, + EventV2Bridge.node, + SessionProjector.node, + Database.node, + CrossSpawnSpawner.node, + InstanceStore.node, + ]), + [ + [Database.node, Database.layerFromPath(databasePath)], + [RuntimeFlags.node, RuntimeFlags.layer({ experimentalWorkspaces: false })], + [ + InstanceBootstrap.node, + Layer.succeed(InstanceBootstrap.Service, InstanceBootstrap.Service.of({ run: Effect.void })), + ], + ], + ), +) + +afterAll(async () => { + await Promise.all( + [databasePath, `${databasePath}-wal`, `${databasePath}-shm`].map((file) => rm(file, { force: true })), + ) +}) + +it.instance( + "admits a prompt whose write waits on another connection's commit instead of failing on a stale read", + () => + Effect.gen(function* () { + const sessions = yield* Session.Service + const queue = yield* SessionQueue.Service + const { db } = yield* Database.Service + const session = yield* sessions.create({ title: "Admission under a concurrent writer" }) + + const sqlite = yield* Effect.promise(() => import("bun:sqlite")) + const holder = new sqlite.Database(databasePath) + yield* Effect.addFinalizer(() => + Effect.sync(() => { + if (holder.inTransaction) holder.run("ROLLBACK") + holder.close() + }), + ) + holder.run("BEGIN IMMEDIATE") + holder.run("CREATE TABLE admission_lock_probe (id INTEGER PRIMARY KEY)") + + const admission = yield* queue + .admit({ + sessionID: session.id, + delivery: "queue", + parts: [{ type: "text", text: "admitted under contention" }], + }) + .pipe(Effect.forkChild) + + // Hold the write lock past the admission's read. The commit then changes + // the database after that read, which is what makes a deferred read stale. + yield* Effect.sleep("1500 millis") + holder.run("COMMIT") + + const exit = yield* Fiber.await(admission) + if (Exit.isFailure(exit)) throw new Error(Cause.pretty(exit.cause)) + const rows = yield* db + .select({ id: SessionPromptQueueTable.id }) + .from(SessionPromptQueueTable) + .where(eq(SessionPromptQueueTable.session_id, session.id)) + .all() + expect(rows).toHaveLength(1) + }), + 30_000, +) From 9e088df64e68420eca7e4b8d900951ccc3972298 Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 04:28:51 +0000 Subject: [PATCH 04/18] refactor(core): give the SQLite busy-retry gate one owner The busy-retry schedule and the stale-snapshot gate were copied into three statement paths: sqlite.bun.ts, sqlite.node.ts and effect-sqlite-node. #155 changed all three, so a rule change had to land three times, and nothing said which copy was authoritative. effect-sqlite-node/src/retry.ts now owns both. It exports retryLocked, which every driver routes its statements through, and it recognizes the stale-snapshot class in both native error shapes: bun:sqlite reports it by code and node:sqlite by errcode. The module imports only effect, so the Bun binary can consume it without loading node:sqlite. The drivers no longer carry a schedule or a gate. Behavior is unchanged. The core busy suite is 23 of 23 across the three clients. Removing the snapshot exclusion from the gate turns the three stale-snapshot tests red. Dropping the node error shape turns two of them red (core-node, standalone-node), and dropping the bun shape turns one red (bun). Removing the retry turns the six lock-wait tests red. statementError, the native error classification, is still copied in the same three files. It is not part of this change, and #87 tracks it. Refs swxtchio/swx-opencode#162 Refs swxtchio/swx-opencode#165 Claude-Session: https://claude.ai/code/session_01LrS7BTRTnc6eYrSxBFoWAo --- packages/core/src/database/sqlite.bun.ts | 29 +++----------------- packages/core/src/database/sqlite.node.ts | 30 +++------------------ packages/effect-sqlite-node/package.json | 3 ++- packages/effect-sqlite-node/src/index.ts | 30 +++------------------ packages/effect-sqlite-node/src/retry.ts | 32 +++++++++++++++++++++++ 5 files changed, 43 insertions(+), 81 deletions(-) create mode 100644 packages/effect-sqlite-node/src/retry.ts diff --git a/packages/core/src/database/sqlite.bun.ts b/packages/core/src/database/sqlite.bun.ts index c7bb9f11f1cb..70f646fb4880 100644 --- a/packages/core/src/database/sqlite.bun.ts +++ b/packages/core/src/database/sqlite.bun.ts @@ -1,40 +1,23 @@ import { Database } from "bun:sqlite" import { drizzle } from "drizzle-orm/bun-sqlite" import * as Context from "effect/Context" -import * as Duration from "effect/Duration" import * as Effect from "effect/Effect" import * as Fiber from "effect/Fiber" import { identity } from "effect/Function" import * as Layer from "effect/Layer" import * as Scope from "effect/Scope" import * as Semaphore from "effect/Semaphore" -import * as Schedule from "effect/Schedule" import * as Stream from "effect/Stream" import * as Reactivity from "effect/unstable/reactivity/Reactivity" import * as Client from "effect/unstable/sql/SqlClient" import type { Connection } from "effect/unstable/sql/SqlConnection" import { classifySqliteError, LockTimeoutError, SqlError } from "effect/unstable/sql/SqlError" import * as Statement from "effect/unstable/sql/Statement" +import { retryLocked } from "@opencode-ai/effect-sqlite-node/retry" import { Sqlite } from "./sqlite" const ATTR_DB_SYSTEM_NAME = "db.system.name" const nativeBusyTimeoutMs = 5 -// 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" }) @@ -100,10 +83,7 @@ const make = (options: Config) => try: () => (statement.all(...(params as any)) ?? []) as Array>, catch: statementError, }) - // 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 retryLocked(execute) }) const runValues = (query: string, params: ReadonlyArray = []) => @@ -115,10 +95,7 @@ const make = (options: Config) => try: () => (statement.values(...(params as any)) ?? []) as Array, catch: statementError, }) - // 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 retryLocked(execute) }) const connection = identity({ diff --git a/packages/core/src/database/sqlite.node.ts b/packages/core/src/database/sqlite.node.ts index 5b3aadebd945..17ca0b66bd8f 100644 --- a/packages/core/src/database/sqlite.node.ts +++ b/packages/core/src/database/sqlite.node.ts @@ -1,41 +1,23 @@ import { DatabaseSync, type SQLInputValue } from "node:sqlite" import { drizzle } from "drizzle-orm/node-sqlite" import * as Context from "effect/Context" -import * as Duration from "effect/Duration" import * as Effect from "effect/Effect" import * as Fiber from "effect/Fiber" import { identity } from "effect/Function" import * as Layer from "effect/Layer" import * as Scope from "effect/Scope" import * as Semaphore from "effect/Semaphore" -import * as Schedule from "effect/Schedule" import * as Stream from "effect/Stream" import * as Reactivity from "effect/unstable/reactivity/Reactivity" import * as Client from "effect/unstable/sql/SqlClient" import type { Connection } from "effect/unstable/sql/SqlConnection" import { classifySqliteError, LockTimeoutError, SqlError } from "effect/unstable/sql/SqlError" import * as Statement from "effect/unstable/sql/Statement" +import { retryLocked } from "@opencode-ai/effect-sqlite-node/retry" import { Sqlite } from "./sqlite" const ATTR_DB_SYSTEM_NAME = "db.system.name" const nativeBusyTimeoutMs = 5 -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 = @@ -109,10 +91,7 @@ const make = (options: Config) => try: () => statement.all(...(params as SQLInputValue[])) as Array>, catch: statementError, }) - // 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 retryLocked(execute) }) const runValues = (query: string, params: ReadonlyArray = []) => @@ -124,10 +103,7 @@ const make = (options: Config) => try: () => statement.all(...(params as SQLInputValue[])) as unknown as ReadonlyArray>, catch: statementError, }) - // 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 retryLocked(execute) }) const connection = identity({ diff --git a/packages/effect-sqlite-node/package.json b/packages/effect-sqlite-node/package.json index 04ced2b0aa90..af8b9c00ba57 100644 --- a/packages/effect-sqlite-node/package.json +++ b/packages/effect-sqlite-node/package.json @@ -9,7 +9,8 @@ "typecheck": "tsgo --noEmit" }, "exports": { - ".": "./src/index.ts" + ".": "./src/index.ts", + "./retry": "./src/retry.ts" }, "devDependencies": { "@tsconfig/bun": "catalog:", diff --git a/packages/effect-sqlite-node/src/index.ts b/packages/effect-sqlite-node/src/index.ts index 55f4d2410d27..937332af8776 100644 --- a/packages/effect-sqlite-node/src/index.ts +++ b/packages/effect-sqlite-node/src/index.ts @@ -3,39 +3,21 @@ export * as NodeSqliteClient from "./index" import { DatabaseSync, type SQLInputValue } from "node:sqlite" import { identity } from "effect/Function" import * as Context from "effect/Context" -import * as Duration from "effect/Duration" import * as Effect from "effect/Effect" import * as Fiber from "effect/Fiber" import * as Layer from "effect/Layer" import * as Scope from "effect/Scope" import * as Semaphore from "effect/Semaphore" -import * as Schedule from "effect/Schedule" import * as Stream from "effect/Stream" import * as Reactivity from "effect/unstable/reactivity/Reactivity" import * as Client from "effect/unstable/sql/SqlClient" import type { Connection } from "effect/unstable/sql/SqlConnection" import { classifySqliteError, LockTimeoutError, SqlError } from "effect/unstable/sql/SqlError" import * as Statement from "effect/unstable/sql/Statement" +import { retryLocked } from "./retry" const ATTR_DB_SYSTEM_NAME = "db.system.name" const nativeBusyTimeoutMs = 5 -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 = @@ -125,10 +107,7 @@ export const make = ( try: () => statement.all(...(params as SQLInputValue[])) as Array>, catch: statementError, }) - // 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 retryLocked(execute) }) const runValues = (sql: string, params: ReadonlyArray = []) => @@ -141,10 +120,7 @@ export const make = ( statement.all(...(params as SQLInputValue[])) as unknown as ReadonlyArray>, catch: statementError, }) - // 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 retryLocked(execute) }) return identity({ diff --git a/packages/effect-sqlite-node/src/retry.ts b/packages/effect-sqlite-node/src/retry.ts new file mode 100644 index 000000000000..f42f9be841a2 --- /dev/null +++ b/packages/effect-sqlite-node/src/retry.ts @@ -0,0 +1,32 @@ +import * as Duration from "effect/Duration" +import * as Effect from "effect/Effect" +import * as Schedule from "effect/Schedule" +import type { SqlError } from "effect/unstable/sql/SqlError" + +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, and only restarting its transaction can recover, +// so that class fails at once. bun:sqlite reports it by code and node:sqlite by its SQLite result code. +const isStaleSnapshot = (cause: unknown) => + typeof cause === "object" && + cause !== null && + (("code" in cause && cause.code === "SQLITE_BUSY_SNAPSHOT") || + ("errcode" in cause && cause.errcode === sqliteBusySnapshot)) + +// Every SQLite client routes its statements through this one gate. +export const retryLocked = (execute: Effect.Effect) => { + const retryable = (error: SqlError) => error.reason.isRetryable && !isStaleSnapshot(error.reason.cause) + // 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 })), + ) +} From 381667f07c1140bba1551dd9b62c134cad756eea Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 13:36:32 +0000 Subject: [PATCH 05/18] test(opencode): wait on the admission's busy attempt, not a sleep Review round 1. The admission test held the write lock for a fixed 1500 ms, then committed. An admission that started late ran after the commit and passed even with a deferred write, so the test did not prove the fix. The test now counts the admission's statement attempts at bun:sqlite's query path, as the core sqlite-busy test does, and records the native codes of the attempts that throw SQLITE_BUSY. It commits only after the admission has met the held lock, asserts the admission is still pending, then checks the persisted row. The timeout is a backstop, not the signal. On failure, the message lists the admission's attempts and busy codes. Checked locally: - writable() immediate: pass, with and without a 1600 ms delay before the fork - writable() reverted to deferred: fail with SQLITE_BUSY_SNAPSHOT after the commit, with and without the 1600 ms delay No product code changes in this commit. Claude-Session: https://claude.ai/code/session_01LrS7BTRTnc6eYrSxBFoWAo --- .../test/session/queue-admission.test.ts | 72 +++++++++++++++++-- 1 file changed, 67 insertions(+), 5 deletions(-) diff --git a/packages/opencode/test/session/queue-admission.test.ts b/packages/opencode/test/session/queue-admission.test.ts index 18de49d26edc..e3e71cd69510 100644 --- a/packages/opencode/test/session/queue-admission.test.ts +++ b/packages/opencode/test/session/queue-admission.test.ts @@ -17,7 +17,7 @@ import { EventV2Bridge } from "@/event-v2-bridge" import { RuntimeFlags } from "@/effect/runtime-flags" import { InstanceStore } from "@/project/instance-store" import { InstanceBootstrap } from "@/project/bootstrap" -import { testEffect } from "../lib/effect" +import { pollWithTimeout, testEffect } from "../lib/effect" // A file-backed store, so a second connection can hold the write lock on the // database the queue writes to. The file is removed after the suite. @@ -51,6 +51,54 @@ afterAll(async () => { ) }) +// The admission's own statements: its transaction begin, and any statement on the queue tables. A busy attempt on +// one of these is the admission meeting the held lock. +const admissionSQL = /^begin\b|session_prompt_queue/i + +// Counts the admission's statement attempts at bun:sqlite's query path, as the core sqlite-busy test does, and the +// native codes of the attempts that throw SQLITE_BUSY. The patch is removed when the test scope closes. +const countAdmissionAttempts = Effect.gen(function* () { + const sqlite = yield* Effect.promise(() => import("bun:sqlite")) + 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 + const counted = new WeakSet() + const attempts = { count: 0, busy: [] as string[] } + const countStatement = (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[]) => { + attempts.count++ + try { + return execute.apply(statement, params) + } catch (error) { + const code = typeof error === "object" && error !== null && "code" in error ? error.code : undefined + if (typeof code === "string" && code.startsWith("SQLITE_BUSY")) attempts.busy.push(code) + throw error + } + }, + }) + } + return statement + } + Object.defineProperty(prototype, "query", { + ...queryDescriptor, + value: function (this: InstanceType, sql: string) { + const statement = query.call(this, sql) + return admissionSQL.test(sql) ? countStatement(statement) : statement + }, + }) + yield* Effect.addFinalizer(() => Effect.sync(() => Object.defineProperty(prototype, "query", queryDescriptor))) + return attempts +}) + it.instance( "admits a prompt whose write waits on another connection's commit instead of failing on a stale read", () => @@ -59,6 +107,7 @@ it.instance( const queue = yield* SessionQueue.Service const { db } = yield* Database.Service const session = yield* sessions.create({ title: "Admission under a concurrent writer" }) + const attempts = yield* countAdmissionAttempts const sqlite = yield* Effect.promise(() => import("bun:sqlite")) const holder = new sqlite.Database(databasePath) @@ -79,13 +128,26 @@ it.instance( }) .pipe(Effect.forkChild) - // Hold the write lock past the admission's read. The commit then changes - // the database after that read, which is what makes a deferred read stale. - yield* Effect.sleep("1500 millis") + // Commit only after the admission has met the held lock. Waiting on its busy + // attempt, not on a clock, means a late-starting admission cannot run after + // the commit. The timeout is only a backstop. + yield* pollWithTimeout( + Effect.sync(() => (attempts.busy.length > 0 ? true : undefined)), + "admission never met SQLITE_BUSY", + "10 seconds", + ) + // It is still pending while the lock is held. + expect(admission.pollUnsafe()).toBeUndefined() + + // The commit then changes the database after the admission's read, which is + // what makes a deferred read stale. holder.run("COMMIT") const exit = yield* Fiber.await(admission) - if (Exit.isFailure(exit)) throw new Error(Cause.pretty(exit.cause)) + if (Exit.isFailure(exit)) + throw new Error( + `${Cause.pretty(exit.cause)}\nadmission attempts: ${attempts.count}; busy codes: ${attempts.busy.join(", ")}`, + ) const rows = yield* db .select({ id: SessionPromptQueueTable.id }) .from(SessionPromptQueueTable) From d71e35e5f0c3730e8add3e0a4e5c1f6ac0a6dc77 Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 14:21:00 +0000 Subject: [PATCH 06/18] test(opencode): read the admission probe's query method through its descriptor Referencing prototype.query unbound tripped the unbound-method lint rule. Reading the descriptor's value keeps the method unbound and adds no behavior change. Claude-Session: https://claude.ai/code/session_01LrS7BTRTnc6eYrSxBFoWAo --- packages/opencode/test/session/queue-admission.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/opencode/test/session/queue-admission.test.ts b/packages/opencode/test/session/queue-admission.test.ts index e3e71cd69510..decd2f7085c2 100644 --- a/packages/opencode/test/session/queue-admission.test.ts +++ b/packages/opencode/test/session/queue-admission.test.ts @@ -62,7 +62,7 @@ const countAdmissionAttempts = Effect.gen(function* () { 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 + const query: typeof prototype.query = queryDescriptor.value const counted = new WeakSet() const attempts = { count: 0, busy: [] as string[] } const countStatement = (statement: S) => { From 53c96a333f5e2b5f57b25e5ec781324f59d2e7b6 Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 15:06:45 +0000 Subject: [PATCH 07/18] docs(opencode): require queue writes to go through writable() The #162 fix makes writable() open every queue write with BEGIN IMMEDIATE. The rule lived only in a code comment, so a future queue write could bypass it. The V1 Prompt Queue section now states it where contributors and agents read queue rules. Claude-Session: https://claude.ai/code/session_01LrS7BTRTnc6eYrSxBFoWAo --- AGENTS.md | 1 + 1 file changed, 1 insertion(+) diff --git a/AGENTS.md b/AGENTS.md index 5edff0ec4721..7226ff0fbbf8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -177,3 +177,4 @@ const table = sqliteTable("session", { - Every V1 prompt except `noReply` is admitted to the fork-owned durable queue in `packages/opencode/src/session/queue.ts` before it becomes a user message; `prompt.ts` only hooks into it. Put new queue behaviour in that module, since `prompt.ts` churns upstream and small hooks re-apply cleanly on a sync. - A `steer` (the default) is promoted at admission so it still reaches the running turn's next step; only an explicit `queue` waits for the would-idle boundary. Holding every steer until the loop top would break senders that expect their message in history mid-run. - Never name a fork table, route or event after upstream v2's `session_input`, `session_pending` or `session_inbox`; a sync bringing v2 in would collide with it. +- Every queue write goes through `writable()` in `queue.ts`, which opens its transaction with `BEGIN IMMEDIATE`. A deferred transaction that reads before it writes pins a read snapshot, and a later commit by another writer then fails the write with `SQLITE_BUSY_SNAPSHOT`, which the driver does not retry (#162). Add new queue writes inside `writable()`; read-only queue transactions may stay deferred. From 8d2e350df25eda8342b4addedd0c06e29a0a002f Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 15:11:40 +0000 Subject: [PATCH 08/18] test(core): one statement-attempt counter for the busy tests The core sqlite-busy test and the opencode admission test each patched bun:sqlite's query method to count statement attempts. That job now has one owner, the fixture-only module packages/core/src/database/sqlite-attempts.ts. Both tests import it, and the core test's node branches count through its countStatement. No runtime module imports the new file. It sits in core's src because opencode tests reach core through its wildcard export, and no shared test-support package exists. Checked: core sqlite-busy (23 pass); admission green with and without a 1600 ms delay before the fork; admission red with writable() reverted to deferred, with and without the delay (busy codes end in SQLITE_BUSY_SNAPSHOT). Claude-Session: https://claude.ai/code/session_01LrS7BTRTnc6eYrSxBFoWAo --- packages/core/src/database/sqlite-attempts.ts | 56 +++++++++++++++++++ packages/core/test/sqlite-busy.test.ts | 50 ++++------------- .../test/session/queue-admission.test.ts | 48 +--------------- 3 files changed, 71 insertions(+), 83 deletions(-) create mode 100644 packages/core/src/database/sqlite-attempts.ts diff --git a/packages/core/src/database/sqlite-attempts.ts b/packages/core/src/database/sqlite-attempts.ts new file mode 100644 index 000000000000..096eb9f53272 --- /dev/null +++ b/packages/core/src/database/sqlite-attempts.ts @@ -0,0 +1,56 @@ +// Fixture-only: counts the statement attempts a SQLite driver makes, for the busy tests in core and opencode. The +// runtime never imports this module. Clients prepare a statement once and retry its execution, so each execution of a +// counted statement is one attempt, and an attempt that throws SQLITE_BUSY is recorded by its native code. +import { Effect } from "effect" + +export type StatementAttempts = { count: number; busy: Array } + +export const newStatementAttempts = (): StatementAttempts => ({ count: 0, busy: [] }) + +const counted = new WeakMap() + +export const countStatement = (statement: S, attempts: StatementAttempts) => { + if (counted.get(statement) === attempts) return statement + counted.set(statement, attempts) + 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[]) => { + attempts.count++ + try { + return execute.apply(statement, params) + } catch (error) { + const code = typeof error === "object" && error !== null && "code" in error ? error.code : undefined + if (typeof code === "string" && code.startsWith("SQLITE_BUSY")) attempts.busy.push(code) + throw error + } + }, + }) + } + return statement +} + +// Patches bun:sqlite's query method, so each statement whose SQL matches counts its attempts. The patch is removed +// when the scope closes. +export const patchBunQuery = Effect.fn("patchBunQuery")(function* ( + match: (sql: string) => boolean, + attempts: StatementAttempts, +) { + const sqlite = yield* Effect.promise(() => import("bun:sqlite")) + const prototype = sqlite.Database.prototype + const descriptor = + Object.getOwnPropertyDescriptor(prototype, "query") ?? + (yield* Effect.die(new Error("bun:sqlite query method was not found"))) + const query: typeof prototype.query = descriptor.value + Object.defineProperty(prototype, "query", { + ...descriptor, + value: function (this: InstanceType, sql: string) { + const statement = query.call(this, sql) + return match(sql) ? countStatement(statement, attempts) : statement + }, + }) + yield* Effect.addFinalizer(() => Effect.sync(() => Object.defineProperty(prototype, "query", descriptor))) +}) diff --git a/packages/core/test/sqlite-busy.test.ts b/packages/core/test/sqlite-busy.test.ts index 2976706ad1b6..6d25128c2911 100644 --- a/packages/core/test/sqlite-busy.test.ts +++ b/packages/core/test/sqlite-busy.test.ts @@ -6,6 +6,12 @@ import { Cause, Context, Deferred, Effect, Exit, Fiber, Layer, Option } from "ef import { isSqlError } from "effect/unstable/sql/SqlError" import { SqlClient } from "effect/unstable/sql/SqlClient" import { Sqlite } from "../src/database/sqlite" +import { + countStatement, + newStatementAttempts, + patchBunQuery, + type StatementAttempts, +} from "../src/database/sqlite-attempts" import path from "path" import { tmpdir } from "./fixture/tmpdir" @@ -177,43 +183,11 @@ const clientWithAttemptCount = (input: { clientType: "bun" | "core-node" | "standalone-node" filename: string statement: string - attempts: { count: number } + attempts: StatementAttempts }) => 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 sqlite = yield* Effect.promise(() => import("bun:sqlite")) - 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 - }, - }) - yield* Effect.addFinalizer(() => Effect.sync(() => Object.defineProperty(prototype, "query", queryDescriptor))) + yield* patchBunQuery((sql) => sql === input.statement, input.attempts) const context = yield* Layer.build(Database.layerFromPath(input.filename)) return Context.get(context, Database.Service).db.$client } @@ -229,7 +203,7 @@ const clientWithAttemptCount = (input: { writable: true, value: function (this: InstanceType, sql: string) { const statement = prepare.call(this, sql) - return sql === input.statement ? countExecutions(statement) : statement + return sql === input.statement ? countStatement(statement, input.attempts) : statement }, }) return Context.get(context, SqlClient) @@ -244,7 +218,7 @@ const clientWithAttemptCount = (input: { ...prepareDescriptor, value: function (this: InstanceType, sql: string) { const statement = prepare.call(this, sql) - return sql === input.statement ? countExecutions(statement) : statement + return sql === input.statement ? countStatement(statement, input.attempts) : statement }, }) yield* Effect.addFinalizer(() => Effect.sync(() => Object.defineProperty(prototype, "prepare", prepareDescriptor))) @@ -296,7 +270,7 @@ describe("SQLite busy timeout and statement retries", () => { const exit = await Effect.runPromiseExit( Effect.scoped( Effect.gen(function* () { - const attempts = { count: 0 } + const attempts = newStatementAttempts() 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 busy_timeout").values).toEqual([[5]]) @@ -416,7 +390,7 @@ describe("SQLite busy timeout and statement retries", () => { const exit = await Effect.runPromiseExit( Effect.scoped( Effect.gen(function* () { - const attempts = { count: 0 } + const attempts = newStatementAttempts() 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"]]) diff --git a/packages/opencode/test/session/queue-admission.test.ts b/packages/opencode/test/session/queue-admission.test.ts index decd2f7085c2..f6869da8f60e 100644 --- a/packages/opencode/test/session/queue-admission.test.ts +++ b/packages/opencode/test/session/queue-admission.test.ts @@ -7,6 +7,7 @@ import { eq } from "drizzle-orm" import { Cause, Effect, Exit, Fiber, Layer } from "effect" import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" import { Database } from "@opencode-ai/core/database/database" +import { newStatementAttempts, patchBunQuery } from "@opencode-ai/core/database/sqlite-attempts" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { SessionProjector } from "@opencode-ai/core/session/projector" @@ -55,50 +56,6 @@ afterAll(async () => { // one of these is the admission meeting the held lock. const admissionSQL = /^begin\b|session_prompt_queue/i -// Counts the admission's statement attempts at bun:sqlite's query path, as the core sqlite-busy test does, and the -// native codes of the attempts that throw SQLITE_BUSY. The patch is removed when the test scope closes. -const countAdmissionAttempts = Effect.gen(function* () { - const sqlite = yield* Effect.promise(() => import("bun:sqlite")) - 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: typeof prototype.query = queryDescriptor.value - const counted = new WeakSet() - const attempts = { count: 0, busy: [] as string[] } - const countStatement = (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[]) => { - attempts.count++ - try { - return execute.apply(statement, params) - } catch (error) { - const code = typeof error === "object" && error !== null && "code" in error ? error.code : undefined - if (typeof code === "string" && code.startsWith("SQLITE_BUSY")) attempts.busy.push(code) - throw error - } - }, - }) - } - return statement - } - Object.defineProperty(prototype, "query", { - ...queryDescriptor, - value: function (this: InstanceType, sql: string) { - const statement = query.call(this, sql) - return admissionSQL.test(sql) ? countStatement(statement) : statement - }, - }) - yield* Effect.addFinalizer(() => Effect.sync(() => Object.defineProperty(prototype, "query", queryDescriptor))) - return attempts -}) - it.instance( "admits a prompt whose write waits on another connection's commit instead of failing on a stale read", () => @@ -107,7 +64,8 @@ it.instance( const queue = yield* SessionQueue.Service const { db } = yield* Database.Service const session = yield* sessions.create({ title: "Admission under a concurrent writer" }) - const attempts = yield* countAdmissionAttempts + const attempts = newStatementAttempts() + yield* patchBunQuery((sql) => admissionSQL.test(sql), attempts) const sqlite = yield* Effect.promise(() => import("bun:sqlite")) const holder = new sqlite.Database(databasePath) From 7b0ea4ad83f46f2c661506e6ca7f904a2f41033f Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 16:15:41 +0000 Subject: [PATCH 09/18] docs: record the #162 lock fix in CHANGESET.md Adds the fork changeset entry for PR #194 (swxtchio/swx-opencode#162 and #165) as the first item under Runtime fixes. Nothing else changes. Claude-Session: https://claude.ai/code/session_01LrS7BTRTnc6eYrSxBFoWAo --- CHANGESET.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGESET.md b/CHANGESET.md index cd758578e617..78555bdcb3ab 100644 --- a/CHANGESET.md +++ b/CHANGESET.md @@ -69,6 +69,7 @@ Newest first. Each entry names the upstream range brought in. ### Runtime fixes +- **#194** Fix dropped accepted prompts under concurrent SQLite writers (swxtchio/swx-opencode#162, #165): queue writes begin with BEGIN IMMEDIATE, so a write waits on the lock instead of failing on a snapshot pinned by an earlier read, and the busy-retry gate has one owner in effect-sqlite-node/src/retry.ts. Queue writes on the consumption path use the same lock, which is beyond #167's scope. _Fork-only._ - **#163** Revert #155 (swxtchio/swx-opencode#153): under concurrent writers it dropped more accepted prompts than its parent #150, so the busy waits return to the pre-#155 schedule. The fix-forward is tracked in swxtchio/swx-opencode#162. _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 From 33237dd216159dc97300ae7ddd2bcd525b9a409e Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 16:49:54 +0000 Subject: [PATCH 10/18] docs(core): plan SQLite lock-hold and stop-site diagnostics (#180) Diagnostic-only logging on candidate 7b0ea4ad83 (fm/opencode-lock-fix-162) so the next SQLITE_BUSY session stop names its lock holder and its call site. Every new line goes through the existing Effect logger; no value, cause, SQL, schedule, timeout or transaction behaviour changes. 1. Write-lock hold timing (effect-drizzle-sqlite session.withTransaction). For a top-level BEGIN IMMEDIATE/EXCLUSIVE that succeeds, time from BEGIN returning to a successful COMMIT or ROLLBACK (including the rollback after a failed commit). Over 250 ms, log "sqlite write lock held" with pid, transaction purpose, current span and duration. Savepoints, deferred transactions and failed BEGINs are not holds. Purpose comes from a new TransactionPurpose context reference, falling back to the current span. event.ts commitDurableEvent provides the event definition type. 2. Retry exhaustion (effect-sqlite-node retryLocked, given the statement SQL by all three drivers). Per execution, count attempts and elapsed time; when a lock failure leaves the gate, log once: statement head, attempts, elapsed ms, and the native cause code and full errcode (item 4), with separate messages for an exhausted window and a stale snapshot that fails without retry. Recovering statements log nothing. 3. Stop call site. A core side table keyed by the failure object records where a failure came from without changing it: commitDurableEvent records the event type and the span chain at the failed commit; the processor records the stream case (handler) or the provider stream (stream). halt's existing "process" error line adds those fields. Tests (focused, producer-driven, removal-to-red proven): - packages/core/test/sqlite-lock-diagnostics.test.ts: long vs short holds, commit and rollback, wait-then-short-hold, nested savepoints; one exhaustion summary with observed attempt count; none on recovery; BUSY vs BUSY_SNAPSHOT native codes on bun, core-node, standalone-node. - packages/opencode/test/session/processor-effect.test.ts: halt line names event, case and call site for reasoning-end (finishReasoning helper) and step-finish (message.updated); provider-stream failure has no event. Preservation: core sqlite-busy and opencode queue-admission suites unchanged. Live proof: env -i scratch store run whose file log shows the hold line. Claude-Session: https://claude.ai/code/session_01QuyVZfg7zCF6ZJmDUf8BnB From db87d48ef72f03d352fa89269765e914faa108df Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 16:56:24 +0000 Subject: [PATCH 11/18] feat(core): log long SQLite write-lock holds and lock retry give-ups Diagnostic logging only (#180); every value, cause, SQL statement, retry schedule and transaction step is unchanged. - effect-drizzle-sqlite withTransaction times a top-level immediate or exclusive BEGIN from its success to the successful COMMIT or ROLLBACK (including the rollback after a failed commit) and logs "sqlite write lock held" with pid, purpose, span, outcome and duration over 250 ms. A new TransactionPurpose reference names the purpose, else the caller's span. - retryLocked takes the statement SQL from all three drivers and, per execution, logs one summary when a lock failure leaves it: statement head, attempts, elapsed ms, and the native code and full errcode, so SQLITE_BUSY and SQLITE_BUSY_SNAPSHOT stay distinguishable. - A core FailureOrigin side table records a failure's event type and span chain at the durable-event commit without changing the failure. - commitDurableEvent supplies its definition type as the purpose. - SessionProcessor records the stream case (or the provider stream) on a failure, and halt's "process" line adds the recorded event, case and call site. Claude-Session: https://claude.ai/code/session_01QuyVZfg7zCF6ZJmDUf8BnB --- packages/core/src/database/sqlite.bun.ts | 4 +- packages/core/src/database/sqlite.node.ts | 4 +- packages/core/src/event.ts | 8 +- .../core/src/observability/failure-origin.ts | 39 ++ .../core/test/sqlite-lock-diagnostics.test.ts | 485 ++++++++++++++++++ .../src/effect-sqlite/session.ts | 44 +- packages/effect-sqlite-node/src/index.ts | 4 +- packages/effect-sqlite-node/src/retry.ts | 38 +- packages/opencode/src/session/processor.ts | 14 +- 9 files changed, 617 insertions(+), 23 deletions(-) create mode 100644 packages/core/src/observability/failure-origin.ts create mode 100644 packages/core/test/sqlite-lock-diagnostics.test.ts diff --git a/packages/core/src/database/sqlite.bun.ts b/packages/core/src/database/sqlite.bun.ts index 70f646fb4880..614321833bb1 100644 --- a/packages/core/src/database/sqlite.bun.ts +++ b/packages/core/src/database/sqlite.bun.ts @@ -83,7 +83,7 @@ const make = (options: Config) => try: () => (statement.all(...(params as any)) ?? []) as Array>, catch: statementError, }) - return retryLocked(execute) + return retryLocked(execute, query) }) const runValues = (query: string, params: ReadonlyArray = []) => @@ -95,7 +95,7 @@ const make = (options: Config) => try: () => (statement.values(...(params as any)) ?? []) as Array, catch: statementError, }) - return retryLocked(execute) + return retryLocked(execute, query) }) const connection = identity({ diff --git a/packages/core/src/database/sqlite.node.ts b/packages/core/src/database/sqlite.node.ts index 17ca0b66bd8f..3ddd0d3947e4 100644 --- a/packages/core/src/database/sqlite.node.ts +++ b/packages/core/src/database/sqlite.node.ts @@ -91,7 +91,7 @@ const make = (options: Config) => try: () => statement.all(...(params as SQLInputValue[])) as Array>, catch: statementError, }) - return retryLocked(execute) + return retryLocked(execute, query) }) const runValues = (query: string, params: ReadonlyArray = []) => @@ -103,7 +103,7 @@ const make = (options: Config) => try: () => statement.all(...(params as SQLInputValue[])) as unknown as ReadonlyArray>, catch: statementError, }) - return retryLocked(execute) + return retryLocked(execute, query) }) const connection = identity({ diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index f0b92fd16b89..7faa855cbeb7 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -5,6 +5,8 @@ import { Event } from "@opencode-ai/schema/event" import type { Data, Definition, Payload } from "@opencode-ai/schema/event" import { and, asc, eq, gt, inArray } from "drizzle-orm" import { Database } from "./database/database" +import { TransactionPurpose } from "@opencode-ai/effect-drizzle-sqlite" +import { FailureOrigin } from "./observability/failure-origin" import { EventRetentionTable, EventSequenceTable, EventTable } from "./event/sql" import { Location } from "./location" import { makeGlobalNode } from "./effect/app-node" @@ -466,7 +468,11 @@ export const layerWith = (options?: LayerOptions) => }), { behavior: "immediate" }, ) - .pipe(Effect.orDie) + .pipe( + Effect.provideService(TransactionPurpose, definition.type), + Effect.tapCause((cause) => FailureOrigin.record(cause, { event: definition.type })), + Effect.orDie, + ) if (committed) { yield* Effect.forEach( pubsub.durable.get(committed.aggregateID) ?? [], diff --git a/packages/core/src/observability/failure-origin.ts b/packages/core/src/observability/failure-origin.ts new file mode 100644 index 000000000000..a081e5e19a43 --- /dev/null +++ b/packages/core/src/observability/failure-origin.ts @@ -0,0 +1,39 @@ +import { Cause, Context, Effect, Tracer } from "effect" + +// Where a failure came from, for diagnostics only. Keyed by the failure object, so the attribution follows that +// failure through Cause squashing and later work cannot overwrite it; the failure itself is never changed. +export type Origin = { + readonly event?: string + readonly stage?: "handler" | "stream" + readonly case?: string + readonly callSite?: string +} + +const origins = new WeakMap() + +// The innermost recording wins each field, and the call site is the span chain where the failure was first seen. +export const record = (cause: Cause.Cause, origin: Omit) => + Effect.withFiber((fiber) => { + if (Cause.hasInterruptsOnly(cause)) return Effect.void + const failure = Cause.squash(cause) + if (typeof failure !== "object" || failure === null) return Effect.void + const callSite = spanChain(fiber.context) + origins.set(failure, { ...origin, ...(callSite && { callSite }), ...origins.get(failure) }) + return Effect.void + }) + +export const get = (failure: unknown) => + typeof failure === "object" && failure !== null ? origins.get(failure) : undefined + +function spanChain(context: Context.Context) { + const names: string[] = [] + const parent = Context.getOption(context, Tracer.ParentSpan) + let span = parent._tag === "Some" ? parent.value : undefined + while (span?._tag === "Span" && names.length < 6) { + names.push(span.name) + span = span.parent._tag === "Some" ? span.parent.value : undefined + } + return names.length ? names.join(" < ") : undefined +} + +export * as FailureOrigin from "./failure-origin" diff --git a/packages/core/test/sqlite-lock-diagnostics.test.ts b/packages/core/test/sqlite-lock-diagnostics.test.ts new file mode 100644 index 000000000000..16d38fe5d461 --- /dev/null +++ b/packages/core/test/sqlite-lock-diagnostics.test.ts @@ -0,0 +1,485 @@ +import { describe, expect, test } from "bun:test" +import { Database } from "@opencode-ai/core/database/database" +import { TransactionPurpose } from "@opencode-ai/effect-drizzle-sqlite" +import { Cause, Context, Effect, Exit, Fiber, Layer, Logger } from "effect" +import { SqlClient } from "effect/unstable/sql/SqlClient" +import { isSqlError } from "effect/unstable/sql/SqlError" +import { Sqlite } from "../src/database/sqlite" +import { + countStatement, + newStatementAttempts, + patchBunQuery, + type StatementAttempts, +} from "../src/database/sqlite-attempts" +import path from "path" +import { tmpdir } from "./fixture/tmpdir" + +const nodeSqliteDrivers = await import("node:sqlite").then( + async (nodeSqlite) => ({ + core: await import("../src/database/sqlite.node"), + standalone: await import("@opencode-ai/effect-sqlite-node"), + DatabaseSync: nodeSqlite.DatabaseSync, + }), + () => undefined, +) +// Above the lock-hold threshold by a margin that scheduling noise cannot close. +const longHoldMs = 400 +// Outlasts the driver's own retry window, so it only catches a retry that never stops. +const lockBackstop = "10 seconds" + +type LogLine = { readonly text: unknown; readonly fields: Record } + +// Collects what the instrumented code sends to the Effect logger, in place of the file sink. +const captured = (effect: (lines: LogLine[]) => Effect.Effect) => + Effect.suspend(() => { + const lines: LogLine[] = [] + const capture = Logger.make((options) => { + const [text, fields] = Array.isArray(options.message) ? options.message : [options.message] + lines.push({ text, fields: typeof fields === "object" && fields !== null ? fields : {} }) + }) + return effect(lines).pipe(Effect.provideService(Logger.CurrentLoggers, new Set([capture]))) + }) + +const holds = (lines: LogLine[]) => lines.filter((line) => line.text === "sqlite write lock held") +const exhausted = (lines: LogLine[]) => lines.filter((line) => line.text === "sqlite lock retries exhausted") +const unretried = (lines: LogLine[]) => lines.filter((line) => line.text === "sqlite lock failed without retry") + +const run = async (effect: Effect.Effect) => { + const exit = await Effect.runPromiseExit(Effect.scoped(effect)) + if (Exit.isFailure(exit)) throw new Error(Cause.pretty(exit.cause)) + return exit.value +} + +const productionDatabase = (filename: string) => + Layer.build(Database.layerFromPath(filename)).pipe(Effect.map((context) => Context.get(context, Database.Service).db)) + +// Holds the write lock from a second connection in this process until the returned release runs. +const inProcessHolder = Effect.fn("inProcessHolder")(function* (filename: string) { + const sqlite = yield* Effect.promise(() => import("bun:sqlite")) + const holder = new sqlite.Database(filename) + yield* Effect.addFinalizer(() => + Effect.sync(() => { + if (holder.inTransaction) holder.run("ROLLBACK") + holder.close() + }), + ) + holder.run("BEGIN IMMEDIATE") + return () => { + if (holder.inTransaction) holder.run("COMMIT") + } +}) + +// A holder in another process takes a real file lock, which every SQLite library in this process must respect. +const childHolder = Effect.fn("childHolder")(function* (filename: string) { + const child = Bun.spawn( + [ + "bun", + "-e", + `import { Database } from "bun:sqlite" +const native = new Database(process.env.SQLITE_LOCK_DB) +native.run("BEGIN IMMEDIATE") +console.log("ready") +await new Promise((resolve) => process.stdin.once("data", resolve)) +native.run("COMMIT") +native.close()`, + ], + { env: { ...process.env, SQLITE_LOCK_DB: filename }, stdin: "pipe", stdout: "pipe", stderr: "inherit" }, + ) + yield* Effect.addFinalizer(() => + Effect.promise(async () => { + child.stdin.write("release\n") + child.stdin.end() + await child.exited + }), + ) + yield* Effect.promise(async () => { + const reader = child.stdout.getReader() + const decoder = new TextDecoder() + let output = "" + while (!output.includes("ready\n")) { + const chunk = await reader.read() + if (chunk.done) throw new Error(`SQLite holder exited before ready: ${output}`) + output += decoder.decode(chunk.value) + } + reader.releaseLock() + }) +}) + +const nodeClientWithAttemptCount = (input: { + clientType: "core-node" | "standalone-node" + filename: string + statement: string + attempts: StatementAttempts +}) => + Effect.gen(function* () { + const drivers = nodeSqliteDrivers + if (!drivers) return yield* Effect.die(new Error("node:sqlite tests ran without node:sqlite")) + const counting = (prepare: (sql: string) => object) => + function (this: unknown, sql: string) { + const statement = prepare.call(this, sql) + return sql === input.statement ? countStatement(statement, input.attempts) : statement + } + if (input.clientType === "core-node") { + const context = yield* Layer.build(drivers.core.layer({ filename: input.filename })) + const native = Context.get(context, Sqlite.Native) as InstanceType + Object.defineProperty(native, "prepare", { configurable: true, writable: true, value: counting(native.prepare) }) + return Context.get(context, SqlClient) + } + const context = yield* Layer.build(drivers.standalone.NodeSqliteClient.layer({ filename: input.filename })) + const prototype = drivers.DatabaseSync.prototype + const descriptor = Object.getOwnPropertyDescriptor(prototype, "prepare") + if (!descriptor) return yield* Effect.die(new Error("node:sqlite prepare method was not found")) + Object.defineProperty(prototype, "prepare", { ...descriptor, value: counting(prototype.prepare) }) + yield* Effect.addFinalizer(() => Effect.sync(() => Object.defineProperty(prototype, "prepare", descriptor))) + return Context.get(context, SqlClient) + }) + +describe("SQLite write-lock hold diagnostics", () => { + test("logs an immediate transaction held past the threshold, by commit and by rollback, and not a short one", () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const db = yield* productionDatabase(path.join(tmp.path, "holds.sqlite")) + yield* db.run("CREATE TABLE hold_test (id INTEGER PRIMARY KEY)") + + const committedStartedAt = performance.now() + yield* db + .transaction( + (tx) => + tx.run("INSERT INTO hold_test (id) VALUES (1)").pipe(Effect.andThen(Effect.sleep(`${longHoldMs} millis`))), + { behavior: "immediate" }, + ) + .pipe(Effect.provideService(TransactionPurpose, "test.commit")) + const committedElapsedMs = performance.now() - committedStartedAt + + const rolledBack = yield* db + .transaction( + (tx) => + tx.run("INSERT INTO hold_test (id) VALUES (2)").pipe( + Effect.andThen(Effect.sleep(`${longHoldMs} millis`)), + Effect.andThen(Effect.fail("abandoned" as const)), + ), + { behavior: "immediate" }, + ) + .pipe(Effect.provideService(TransactionPurpose, "test.rollback"), Effect.flip) + expect(rolledBack).toBe("abandoned") + + yield* db.transaction((tx) => tx.run("INSERT INTO hold_test (id) VALUES (3)"), { behavior: "immediate" }).pipe( + Effect.provideService(TransactionPurpose, "test.short"), + ) + // A deferred transaction does not take the write lock at BEGIN, so it is not a hold. + yield* db.transaction(() => Effect.sleep(`${longHoldMs} millis`)) + + const logged = holds(lines) + expect(logged.map((line) => [line.fields.purpose, line.fields.outcome])).toEqual([ + ["test.commit", "commit"], + ["test.rollback", "rollback"], + ]) + for (const line of logged) { + expect(line.fields.pid).toBe(process.pid) + expect(line.fields.durationMs).toBeGreaterThanOrEqual(longHoldMs) + } + expect(logged[0].fields.durationMs).toBeLessThanOrEqual(Math.ceil(committedElapsedMs)) + expect(yield* db.values("SELECT id FROM hold_test ORDER BY id")).toEqual([[1], [3]]) + }), + ), + )) + + test("names an unlabelled hold by its span and keeps nested savepoints inside the outer hold", () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const db = yield* productionDatabase(path.join(tmp.path, "nested.sqlite")) + yield* db.run("CREATE TABLE hold_test (id INTEGER PRIMARY KEY)") + + // Each step is under the threshold; only the outer hold that spans them all exceeds it. + const stepMs = 150 + yield* db + .transaction( + (tx) => + Effect.gen(function* () { + yield* Effect.sleep(`${stepMs} millis`) + yield* tx.transaction((inner) => + inner.run("INSERT INTO hold_test (id) VALUES (1)").pipe(Effect.andThen(Effect.sleep(`${stepMs} millis`))), + ) + yield* Effect.sleep(`${stepMs} millis`) + }), + { behavior: "immediate" }, + ) + .pipe(Effect.withSpan("test.nestedCaller")) + + const logged = holds(lines) + expect(logged).toHaveLength(1) + expect(logged[0].fields).toMatchObject({ purpose: "test.nestedCaller", span: "test.nestedCaller", outcome: "commit" }) + expect(logged[0].fields.durationMs).toBeGreaterThanOrEqual(stepMs * 3) + }), + ), + )) + + test("attributes a failed commit and its rollback to the same hold", () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const db = yield* productionDatabase(path.join(tmp.path, "commit-failure.sqlite")) + yield* db.run("CREATE TABLE parent (id INTEGER PRIMARY KEY)") + yield* db.run( + "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id) DEFERRABLE INITIALLY DEFERRED)", + ) + + // The deferred foreign key is checked at COMMIT, which fails and is followed by the rollback. + const failure = yield* db + .transaction( + (tx) => + tx + .run("INSERT INTO child (id, parent_id) VALUES (1, 99)") + .pipe(Effect.andThen(Effect.sleep(`${longHoldMs} millis`))), + { behavior: "immediate" }, + ) + .pipe(Effect.provideService(TransactionPurpose, "test.failedCommit"), Effect.flip) + expect(isSqlError(failure) && failure.reason.cause).toMatchObject({ code: "SQLITE_CONSTRAINT_FOREIGNKEY" }) + + const logged = holds(lines) + expect(logged).toHaveLength(1) + expect(logged[0].fields).toMatchObject({ purpose: "test.failedCommit", outcome: "rollback" }) + expect(logged[0].fields.durationMs).toBeGreaterThanOrEqual(longHoldMs) + expect(yield* db.values("SELECT COUNT(*) FROM child")).toEqual([[0]]) + }), + ), + )) + + test("does not count a long wait for the lock as a hold, nor log a statement that recovers", () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const filename = path.join(tmp.path, "wait.sqlite") + const attempts = newStatementAttempts() + yield* patchBunQuery((sql) => sql === "begin immediate", attempts) + const db = yield* productionDatabase(filename) + yield* db.run("CREATE TABLE hold_test (id INTEGER PRIMARY KEY)") + const release = yield* inProcessHolder(filename) + + const waitStartedAt = performance.now() + const writer = yield* db + .transaction((tx) => tx.run("INSERT INTO hold_test (id) VALUES (1)"), { behavior: "immediate" }) + .pipe(Effect.provideService(TransactionPurpose, "test.waiter"), Effect.forkChild({ startImmediately: true })) + // The BEGIN has met the held lock; keep holding well past the hold threshold before releasing it. + while (attempts.busy.length === 0 || performance.now() - waitStartedAt < longHoldMs) + yield* Effect.sleep("10 millis") + release() + yield* Fiber.join(writer).pipe( + Effect.timeoutOrElse({ duration: lockBackstop, orElse: () => Effect.die(new Error("writer never finished")) }), + ) + const waitedMs = performance.now() - waitStartedAt + + expect(waitedMs).toBeGreaterThan(longHoldMs) + expect(attempts.busy.length).toBeGreaterThan(0) + expect(attempts.count).toBeGreaterThan(attempts.busy.length) + expect(holds(lines)).toEqual([]) + expect(exhausted(lines)).toEqual([]) + expect(unretried(lines)).toEqual([]) + expect(yield* db.values("SELECT id FROM hold_test")).toEqual([[1]]) + }), + ), + )) +}) + +describe("SQLite lock retry exhaustion diagnostics", () => { + test( + "summarizes an exhausted BEGIN IMMEDIATE once, with its observed attempts and native code, and logs no hold", + () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const filename = path.join(tmp.path, "begin-exhausted.sqlite") + const attempts = newStatementAttempts() + yield* patchBunQuery((sql) => sql === "begin immediate", attempts) + const db = yield* productionDatabase(filename) + yield* db.run("CREATE TABLE hold_test (id INTEGER PRIMARY KEY)") + yield* inProcessHolder(filename) + + const startedAt = performance.now() + yield* db + .transaction((tx) => tx.run("INSERT INTO hold_test (id) VALUES (1)"), { behavior: "immediate" }) + .pipe( + Effect.provideService(TransactionPurpose, "test.exhausted"), + Effect.timeoutOrElse({ duration: lockBackstop, orElse: () => Effect.die(new Error("retries never stopped")) }), + Effect.flip, + ) + const elapsedMs = performance.now() - startedAt + + const summaries = exhausted(lines) + expect(summaries).toHaveLength(1) + expect(summaries[0].fields).toMatchObject({ + pid: process.pid, + statement: "begin immediate", + attempts: attempts.count, + "sqlite.code": "SQLITE_BUSY", + }) + expect(attempts.count).toBeGreaterThan(1) + expect(summaries[0].fields.elapsedMs).toBeGreaterThanOrEqual(5_000) + expect(summaries[0].fields.elapsedMs).toBeLessThanOrEqual(Math.ceil(elapsedMs)) + expect(holds(lines)).toEqual([]) + expect(unretried(lines)).toEqual([]) + }), + ), + ), + 20_000, + ) + + test( + "keeps attempt counts local to each execution of one reused statement effect", + () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const filename = path.join(tmp.path, "reuse.sqlite") + const writerSql = "INSERT INTO hold_test (id) VALUES (1)" + const attempts = newStatementAttempts() + yield* patchBunQuery((sql) => sql === writerSql, attempts) + const db = yield* productionDatabase(filename) + yield* db.run("CREATE TABLE hold_test (id INTEGER PRIMARY KEY)") + yield* inProcessHolder(filename) + + // One statement effect, run twice at once; the connection serializes them, so each meets its own window. + const write = db.$client.unsafe(writerSql).raw.pipe(Effect.flip) + yield* Effect.all([write, write], { concurrency: 2 }).pipe( + Effect.timeoutOrElse({ duration: "20 seconds", orElse: () => Effect.die(new Error("retries never stopped")) }), + ) + + const summaries = exhausted(lines) + expect(summaries).toHaveLength(2) + const counts = summaries.map((line) => line.fields.attempts as number) + expect(counts.every((count) => count > 1)).toBe(true) + expect(counts[0] + counts[1]).toBe(attempts.count) + for (const line of summaries) { + expect(line.fields.statement).toBe("INSERT INTO hold_test") + expect(line.fields.elapsedMs).toBeLessThan(10_000) + } + }), + ), + ), + 30_000, + ) + + for (const clientType of ["core-node", "standalone-node"] as const) { + test.skipIf(!nodeSqliteDrivers)( + `${clientType} summarizes an exhausted statement once with its full native result code`, + () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const filename = path.join(tmp.path, `${clientType}-exhausted.sqlite`) + const writerSql = "INSERT INTO hold_test (id) VALUES (1)" + const attempts = newStatementAttempts() + const client = yield* nodeClientWithAttemptCount({ clientType, filename, statement: writerSql, attempts }) + yield* client.unsafe("CREATE TABLE hold_test (id INTEGER PRIMARY KEY)").raw + yield* childHolder(filename) + + yield* client.unsafe(writerSql).raw.pipe( + Effect.timeoutOrElse({ duration: lockBackstop, orElse: () => Effect.die(new Error("retries never stopped")) }), + Effect.flip, + ) + + const summaries = exhausted(lines) + expect(summaries).toHaveLength(1) + expect(summaries[0].fields).toMatchObject({ + statement: "INSERT INTO hold_test", + attempts: attempts.count, + "sqlite.code": "ERR_SQLITE_ERROR", + "sqlite.errcode": 5, + }) + expect(attempts.count).toBeGreaterThan(1) + expect(unretried(lines)).toEqual([]) + }), + ), + ), + 20_000, + ) + } + + for (const clientType of ["bun", "core-node", "standalone-node"] as const) { + test.skipIf(clientType !== "bun" && !nodeSqliteDrivers)( + `${clientType} logs a stale-snapshot write once by its own native code, apart from plain SQLITE_BUSY`, + () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const filename = path.join(tmp.path, `${clientType}-snapshot.sqlite`) + const writerSql = "INSERT INTO hold_test (id) VALUES (1)" + const attempts = newStatementAttempts() + const client: SqlClient = + clientType === "bun" + ? yield* patchBunQuery((sql) => sql === writerSql, attempts).pipe( + Effect.andThen(productionDatabase(filename)), + Effect.map((db) => db.$client), + ) + : yield* nodeClientWithAttemptCount({ clientType, filename, statement: writerSql, attempts }) + yield* client.unsafe("CREATE TABLE hold_test (id INTEGER PRIMARY KEY)").raw + yield* client.unsafe("PRAGMA journal_mode = WAL").raw + const sqlite = yield* Effect.promise(() => import("bun:sqlite")) + const other = new sqlite.Database(filename) + yield* Effect.addFinalizer(() => Effect.sync(() => other.close())) + + yield* client + .withTransaction( + Effect.gen(function* () { + yield* client.unsafe("SELECT COUNT(*) FROM hold_test").values + // Another connection commits after this transaction's read, so its snapshot is stale. + other.run("INSERT INTO hold_test (id) VALUES (2)") + yield* client.unsafe(writerSql).raw + }), + ) + .pipe( + Effect.timeoutOrElse({ duration: lockBackstop, orElse: () => Effect.die(new Error("retries never stopped")) }), + Effect.flip, + ) + + const summaries = unretried(lines) + expect(summaries).toHaveLength(1) + expect(summaries[0].fields).toMatchObject({ statement: "INSERT INTO hold_test", attempts: 1 }) + expect(attempts.count).toBe(1) + expect(summaries[0].fields).toMatchObject( + clientType === "bun" + ? { "sqlite.code": "SQLITE_BUSY_SNAPSHOT" } + : { "sqlite.code": "ERR_SQLITE_ERROR", "sqlite.errcode": 517 }, + ) + expect(exhausted(lines)).toEqual([]) + }), + ), + ), + 20_000, + ) + } +}) diff --git a/packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts b/packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts index cba4c11bd424..32fcd6784294 100644 --- a/packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts +++ b/packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts @@ -3,6 +3,7 @@ import * as Context from "effect/Context" import * as Effect from "effect/Effect" import * as Exit from "effect/Exit" import * as Scope from "effect/Scope" +import * as Tracer from "effect/Tracer" import type { SqlClient } from "effect/unstable/sql/SqlClient" import type { SqlError } from "effect/unstable/sql/SqlError" import type { EffectCacheShape } from "drizzle-orm/cache/core/cache-effect" @@ -26,6 +27,15 @@ export interface EffectSQLiteQueryEffectHKT extends QueryEffectHKTBase { export type EffectSQLiteRunResult = readonly never[] +// Names what a write transaction is for in its lock-hold diagnostic. Unset, the diagnostic names the current span. +export const TransactionPurpose = Context.Reference( + "@opencode-ai/effect-drizzle-sqlite/TransactionPurpose", + { defaultValue: () => undefined }, +) + +// A write lock held longer than this is logged as a likely cause of another connection's lock wait. +const longWriteLockHoldMs = 250 + export interface EffectSQLiteSessionOptions { logger: EffectLoggerShape cache: EffectCacheShape @@ -137,15 +147,33 @@ export class EffectSQLiteSession extends SQLite ), ) const id = connectionOption._tag === "Some" ? connectionOption.value[1] + 1 : 0 + const behavior = config?.behavior ?? "deferred" return connection.pipe( Effect.flatMap(([scope, connection]) => { const transaction = this.executeTransactionStatement( connection, - id === 0 ? `begin ${config?.behavior ?? "deferred"}` : `savepoint effect_sql_${id}`, + id === 0 ? `begin ${behavior}` : `savepoint effect_sql_${id}`, ).pipe( - Effect.flatMap(() => - Effect.provideContext( + Effect.flatMap(() => { + // A top-level immediate or exclusive BEGIN owns the write lock from here until COMMIT or ROLLBACK + // succeeds; time spent waiting for the lock is the retry gate's, not this hold's. + const acquiredAt = performance.now() + const released = (outcome: "commit" | "rollback") => + Effect.suspend(() => { + const heldMs = performance.now() - acquiredAt + if (id !== 0 || behavior === "deferred" || heldMs <= longWriteLockHoldMs) return Effect.void + const parent = Context.getOption(services, Tracer.ParentSpan) + const span = parent._tag === "Some" && parent.value._tag === "Span" ? parent.value.name : undefined + return Effect.logWarning("sqlite write lock held", { + pid: process.pid, + purpose: Context.get(services, TransactionPurpose) ?? span ?? "unknown", + span, + outcome, + durationMs: Math.round(heldMs), + }) + }) + return Effect.provideContext( restore(effect), Context.add(services, this.client.transactionService, [connection, id]), ).pipe( @@ -154,9 +182,11 @@ export class EffectSQLiteSession extends SQLite const finalize = Exit.isSuccess(exit) ? id === 0 ? this.executeTransactionStatement(connection, "commit").pipe( + Effect.tap(() => released("commit")), // SQLite keeps the transaction open after deferred constraint commit failures. Effect.catch((error) => this.executeTransactionStatement(connection, "rollback").pipe( + Effect.tap(() => released("rollback")), Effect.catch(() => Effect.void), Effect.andThen(Effect.fail(error)), ), @@ -164,7 +194,9 @@ export class EffectSQLiteSession extends SQLite ) : this.executeTransactionStatement(connection, `release savepoint effect_sql_${id}`) : id === 0 - ? this.executeTransactionStatement(connection, "rollback") + ? this.executeTransactionStatement(connection, "rollback").pipe( + Effect.tap(() => released("rollback")), + ) : this.executeTransactionStatement(connection, `rollback to savepoint effect_sql_${id}`).pipe( Effect.andThen( this.executeTransactionStatement(connection, `release savepoint effect_sql_${id}`), @@ -173,8 +205,8 @@ export class EffectSQLiteSession extends SQLite return finalize.pipe(Effect.flatMap(() => exit)) }), - ), - ), + ) + }), ) return scope === undefined diff --git a/packages/effect-sqlite-node/src/index.ts b/packages/effect-sqlite-node/src/index.ts index 937332af8776..339125972f6c 100644 --- a/packages/effect-sqlite-node/src/index.ts +++ b/packages/effect-sqlite-node/src/index.ts @@ -107,7 +107,7 @@ export const make = ( try: () => statement.all(...(params as SQLInputValue[])) as Array>, catch: statementError, }) - return retryLocked(execute) + return retryLocked(execute, sql) }) const runValues = (sql: string, params: ReadonlyArray = []) => @@ -120,7 +120,7 @@ export const make = ( statement.all(...(params as SQLInputValue[])) as unknown as ReadonlyArray>, catch: statementError, }) - return retryLocked(execute) + return retryLocked(execute, sql) }) return identity({ diff --git a/packages/effect-sqlite-node/src/retry.ts b/packages/effect-sqlite-node/src/retry.ts index f42f9be841a2..a99352257028 100644 --- a/packages/effect-sqlite-node/src/retry.ts +++ b/packages/effect-sqlite-node/src/retry.ts @@ -22,11 +22,33 @@ const isStaleSnapshot = (cause: unknown) => (("code" in cause && cause.code === "SQLITE_BUSY_SNAPSHOT") || ("errcode" in cause && cause.errcode === sqliteBusySnapshot)) -// Every SQLite client routes its statements through this one gate. -export const retryLocked = (execute: Effect.Effect) => { - const retryable = (error: SqlError) => error.reason.isRetryable && !isStaleSnapshot(error.reason.cause) - // 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 })), - ) -} +// Every SQLite client routes its statements through this one gate. A lock failure that leaves it logs one summary, +// never one line per attempt, so a contended statement cannot flood the log. +export const retryLocked = (execute: Effect.Effect, sql: string) => + Effect.suspend(() => { + const startedAt = performance.now() + let attempts = 0 + const attempt = Effect.suspend(() => { + attempts++ + return execute + }) + const retryable = (error: SqlError) => error.reason.isRetryable && !isStaleSnapshot(error.reason.cause) + // Only a statement that met the lock builds the retry schedule, which re-attempts it at once before backing off. + return attempt.pipe( + Effect.catchIf(retryable, () => Effect.retry(attempt, { schedule: retrySchedule, while: retryable })), + Effect.tapError((error) => { + if (error.reason._tag !== "LockTimeoutError") return Effect.void + const cause: { code?: unknown; errcode?: unknown; errno?: unknown } = + typeof error.reason.cause === "object" && error.reason.cause !== null ? error.reason.cause : {} + return Effect.logWarning(retryable(error) ? "sqlite lock retries exhausted" : "sqlite lock failed without retry", { + pid: process.pid, + statement: sql.trim().split(/\s+/).slice(0, 3).join(" ").slice(0, 64), + attempts, + elapsedMs: Math.round(performance.now() - startedAt), + // node:sqlite's code is generic, so its full extended result code tells BUSY from BUSY_SNAPSHOT. + "sqlite.code": cause.code, + "sqlite.errcode": cause.errcode ?? cause.errno, + }) + }), + ) + }) diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 23972f669fdd..ac6aa7ee2726 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -23,6 +23,7 @@ import { ModelV2 } from "@opencode-ai/core/model" import { Question } from "@/question" import { errorMessage } from "@/util/error" import { sqliteLockMessage } from "@/util/sqlite-error" +import { FailureOrigin } from "@opencode-ai/core/observability/failure-origin" import { isRecord } from "@/util/record" import { EventV2Bridge } from "@/event-v2-bridge" import { Database } from "@opencode-ai/core/database/database" @@ -837,11 +838,14 @@ const layer = Layer.effect( }) const halt = Effect.fn("SessionProcessor.halt")(function* (e: unknown) { + // Names the failed operation's event, stream case and call site, so a stop is traceable past its stack. + const origin = FailureOrigin.get(e) yield* Effect.logError("process", { "session.id": input.sessionID, messageID: input.assistantMessage.id, error: errorMessage(e), stack: e instanceof Error ? e.stack : undefined, + ...(origin && { failure: origin }), }) if (ctx.assistantMessage.error) { yield* status.set(ctx.sessionID, { type: "idle" }) @@ -884,10 +888,16 @@ const layer = Layer.effect( type: "busy", activeAssistantMessageID: ctx.assistantMessage.id, }) - const stream = llm.stream(streamInput) + const stream = llm.stream(streamInput).pipe( + Stream.tapCause((cause) => FailureOrigin.record(cause, { stage: "stream" })), + ) yield* stream.pipe( - Stream.tap((event) => handleEvent(event)), + Stream.tap((event) => + handleEvent(event).pipe( + Effect.tapCause((cause) => FailureOrigin.record(cause, { stage: "handler", case: event.type })), + ), + ), Stream.takeUntil(() => ctx.needsCompaction), Stream.runDrain, ) From 3010fba2d4022a4b05fc4a0384ba0669244dda42 Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 17:11:00 +0000 Subject: [PATCH 12/18] test(opencode): prove halt names the failed write's event, case and call site Adds focused processor tests driving real lock failures into halt (#180): - reasoning-end: the finishReasoning helper's message.part.updated commit meets a held lock; halt names the event, the case and a call-site chain running Session.updatePart < SessionProcessor.finishReasoning. - step-finish: the message.updated write meets the lock instead, so the same case path reports a different event identity. - a provider stream that dies after its handled cases succeed is attributed to the stream, with no invented event or case. The held-lock fixture gains a trigger choosing which write takes the lock; its default keeps the existing tests' reasoning-end behaviour. The core nested-savepoint hold test now nests an immediate transaction long enough to log on its own, and withTransaction drops a guard that was dead: only the top-level COMMIT and ROLLBACK report a release. Claude-Session: https://claude.ai/code/session_01QuyVZfg7zCF6ZJmDUf8BnB --- .../core/test/sqlite-lock-diagnostics.test.ts | 13 +- .../src/effect-sqlite/session.ts | 5 +- .../test/session/processor-effect.test.ts | 166 +++++++++++++++++- 3 files changed, 174 insertions(+), 10 deletions(-) diff --git a/packages/core/test/sqlite-lock-diagnostics.test.ts b/packages/core/test/sqlite-lock-diagnostics.test.ts index 16d38fe5d461..811ea8e48443 100644 --- a/packages/core/test/sqlite-lock-diagnostics.test.ts +++ b/packages/core/test/sqlite-lock-diagnostics.test.ts @@ -200,15 +200,20 @@ describe("SQLite write-lock hold diagnostics", () => { const db = yield* productionDatabase(path.join(tmp.path, "nested.sqlite")) yield* db.run("CREATE TABLE hold_test (id INTEGER PRIMARY KEY)") - // Each step is under the threshold; only the outer hold that spans them all exceeds it. + // The savepoint alone outlasts the threshold, so only its exclusion keeps it from logging a hold of its own. const stepMs = 150 yield* db .transaction( (tx) => Effect.gen(function* () { yield* Effect.sleep(`${stepMs} millis`) - yield* tx.transaction((inner) => - inner.run("INSERT INTO hold_test (id) VALUES (1)").pipe(Effect.andThen(Effect.sleep(`${stepMs} millis`))), + // A nested immediate transaction, as a durable-event commit inside an open one makes, is a savepoint. + yield* db.transaction( + (inner) => + inner + .run("INSERT INTO hold_test (id) VALUES (1)") + .pipe(Effect.andThen(Effect.sleep(`${longHoldMs} millis`))), + { behavior: "immediate" }, ) yield* Effect.sleep(`${stepMs} millis`) }), @@ -219,7 +224,7 @@ describe("SQLite write-lock hold diagnostics", () => { const logged = holds(lines) expect(logged).toHaveLength(1) expect(logged[0].fields).toMatchObject({ purpose: "test.nestedCaller", span: "test.nestedCaller", outcome: "commit" }) - expect(logged[0].fields.durationMs).toBeGreaterThanOrEqual(stepMs * 3) + expect(logged[0].fields.durationMs).toBeGreaterThanOrEqual(stepMs * 2 + longHoldMs) }), ), )) diff --git a/packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts b/packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts index 32fcd6784294..a8c4b3e9dff9 100644 --- a/packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts +++ b/packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts @@ -157,12 +157,13 @@ export class EffectSQLiteSession extends SQLite ).pipe( Effect.flatMap(() => { // A top-level immediate or exclusive BEGIN owns the write lock from here until COMMIT or ROLLBACK - // succeeds; time spent waiting for the lock is the retry gate's, not this hold's. + // succeeds; time spent waiting for the lock is the retry gate's, not this hold's. Only those top-level + // statements report a release, so a savepoint neither starts nor ends a hold. const acquiredAt = performance.now() const released = (outcome: "commit" | "rollback") => Effect.suspend(() => { const heldMs = performance.now() - acquiredAt - if (id !== 0 || behavior === "deferred" || heldMs <= longWriteLockHoldMs) return Effect.void + if (behavior === "deferred" || heldMs <= longWriteLockHoldMs) return Effect.void const parent = Context.getOption(services, Tracer.ParentSpan) const span = parent._tag === "Some" && parent.value._tag === "Span" ? parent.value.name : undefined return Effect.logWarning("sqlite write lock held", { diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index 7e49dbf58757..a24107453f86 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -419,6 +419,7 @@ const sqliteTerminalDbPaths = ([2, 3] as const).map((failures) => ({ path: path.join(import.meta.dir, `.opencode-sqlite-terminal-${failures}-${crypto.randomUUID()}.db`), })) const heldLockDbPath = path.join(import.meta.dir, `.opencode-sqlite-held-lock-${crypto.randomUUID()}.db`) +const streamLockDbPath = path.join(import.meta.dir, `.opencode-sqlite-stream-lock-${crypto.randomUUID()}.db`) const routedEnv = LayerNode.compile(root, [ ...replacements, [LLM.node, routedLLM], @@ -445,6 +446,9 @@ afterAll(async () => { heldLockDbPath, `${heldLockDbPath}-wal`, `${heldLockDbPath}-shm`, + streamLockDbPath, + `${streamLockDbPath}-wal`, + `${streamLockDbPath}-shm`, routerLabelDbPath, `${routerLabelDbPath}-wal`, `${routerLabelDbPath}-shm`, @@ -552,6 +556,8 @@ const lockTerminalLLM = Layer.succeed( ) type HeldLock = { messageID: MessageID + // The write the second connection takes the lock just before: the reasoning-end part or the step-finish message. + trigger: "reasoning-end" | "step-finish" holder?: Sqlite streaming: ReturnType> resume: ReturnType> @@ -588,14 +594,26 @@ const heldLockSession = LayerNode.make({ Effect.suspend(() => { const lock = lockFor(part.sessionID, part.messageID) // A second connection takes the write lock just before the turn's reasoning-end write. - if (lock && !lock.holder && part.type === "reasoning" && part.time.end !== undefined) { + if ( + lock?.trigger === "reasoning-end" && + !lock.holder && + part.type === "reasoning" && + part.time.end !== undefined + ) { lock.holder = new Sqlite(heldLockDbPath) lock.holder.run("BEGIN IMMEDIATE") } return heldLockWrite(lock, real.updatePart(part)) }), updateMessage: (msg: T) => - Effect.suspend(() => heldLockWrite(lockFor(msg.sessionID, msg.id), real.updateMessage(msg))), + Effect.suspend(() => { + const lock = lockFor(msg.sessionID, msg.id) + if (lock?.trigger === "step-finish" && !lock.holder && msg.role === "assistant" && msg.finish) { + lock.holder = new Sqlite(heldLockDbPath) + lock.holder.run("BEGIN IMMEDIATE") + } + return heldLockWrite(lock, real.updateMessage(msg)) + }), updateExistingMessage: (msg: SessionV1.Info) => Effect.suspend(() => { const lock = lockFor(msg.sessionID, msg.id) @@ -659,6 +677,31 @@ const sqliteLockEnv = LayerNode.compile(root, [ [Database.node, Database.layerFromPath(sqliteLockDbPath)], ]) const itSqliteLock = testEffect(sqliteLockEnv) +// The provider stream dies with the session's lock failure after its handled events all succeed. +const streamLockFailures = new Map() +const itStreamLock = testEffect( + LayerNode.compile(root, [ + ...replacements, + [ + LLM.node, + Layer.succeed( + LLM.Service, + LLM.Service.of({ + stream: (input) => + Stream.concat( + Stream.make( + LLMEvent.stepStart({ index: 0 }), + LLMEvent.reasoningStart({ id: "reasoning-1" }), + LLMEvent.reasoningEnd({ id: "reasoning-1" }), + ), + Stream.die(streamLockFailures.get(input.sessionID as SessionID)), + ), + }), + ), + ], + [Database.node, Database.layerFromPath(streamLockDbPath)], + ]), +) const boot = Effect.fn("test.boot")(function* () { const processors = yield* SessionProcessor.Service @@ -1669,7 +1712,10 @@ const itHeldLock = testEffect( ) // Runs a turn whose reasoning-end write takes a real write lock that stays held after the processor exits. -const failTurnUnderHeldLock = Effect.fn("test.failTurnUnderHeldLock")(function* (dir: string) { +const failTurnUnderHeldLock = Effect.fn("test.failTurnUnderHeldLock")(function* ( + dir: string, + trigger: HeldLock["trigger"] = "reasoning-end", +) { const { processors, session, provider } = yield* boot() const events = yield* EventV2Bridge.Service const statuses = yield* SessionStatus.Service @@ -1677,7 +1723,7 @@ const failTurnUnderHeldLock = Effect.fn("test.failTurnUnderHeldLock")(function* const parent = yield* user(chat.id, "hold the write lock") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const model = yield* provider.getModel(ref.providerID, ref.modelID) - const lock: HeldLock = { messageID: msg.id, streaming: defer(), resume: defer(), replayed: 0 } + const lock: HeldLock = { messageID: msg.id, trigger, streaming: defer(), resume: defer(), replayed: 0 } heldLocks.set(chat.id, lock) yield* Effect.addFinalizer(() => Effect.sync(() => { @@ -1861,6 +1907,118 @@ itHeldLock.live( 30_000, ) +// Captures halt's error line, which carries the failed operation's recorded origin. +const haltLines = () => { + const lines: Record[] = [] + const logger = Logger.make((options) => { + const [text, fields] = Array.isArray(options.message) ? options.message : [options.message] + if (options.logLevel === "Error" && text === "process" && typeof fields === "object" && fields !== null) + lines.push(fields) + }) + return { lines, logger } +} + +for (const scenario of [ + { + trigger: "reasoning-end", + event: SessionV1.Event.PartUpdated.type, + callSite: ["Session.updatePart", "SessionProcessor.finishReasoning"], + }, + { trigger: "step-finish", event: SessionV1.Event.MessageUpdated.type, callSite: ["Session.updateMessage"] }, +] as const) { + itHeldLock.live( + `processor halt names the ${scenario.trigger} write that met the held lock`, + () => + provideTmpdirInstance( + (dir) => + Effect.gen(function* () { + const halt = haltLines() + const loggers = yield* Effect.service(Logger.CurrentLoggers) + const { lock, exit, terminal } = yield* failTurnUnderHeldLock(dir, scenario.trigger).pipe( + Effect.provideService(Logger.CurrentLoggers, new Set([...loggers, halt.logger])), + ) + expect(lock.produced && sqliteLockMessage(lock.produced)).toBe("Database is locked (SQLITE_BUSY)") + + expect(halt.lines).toHaveLength(1) + expect(String(halt.lines[0].error).toLowerCase()).toContain("database is locked (sqlite_busy)") + const failure = halt.lines[0].failure as Record | undefined + expect(failure).toMatchObject({ event: scenario.event, stage: "handler", case: scenario.trigger }) + // The span chain runs from the failed commit outward, through the helper the case called. + const frames = String(failure?.callSite).split(" < ") + const positions = scenario.callSite.map((name) => frames.indexOf(name)) + expect(positions.every((position) => position >= 0)).toBe(true) + expect(positions).toEqual(positions.toSorted((a, b) => a - b)) + + // The diagnostics leave the stop, its idle state and the eventual terminal write as they were. + expect(Exit.isSuccess(exit) && exit.value).toBe("stop") + lock.holder?.run("ROLLBACK") + yield* Deferred.await(terminal).pipe( + Effect.timeoutOrElse({ + duration: "10 seconds", + orElse: () => Effect.die(new Error("the failed turn was not persisted after the lock was released")), + }), + ) + }), + { config: cfg }, + ), + 40_000, + ) +} + +itStreamLock.live( + "processor halt attributes a provider stream lock failure to the stream, not the last handled case", + () => + provideTmpdirInstance( + (dir) => + Effect.gen(function* () { + const { processors, session, provider } = yield* boot() + const database = yield* Database.Service + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "stream lock origin") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const model = yield* provider.getModel(ref.providerID, ref.modelID) + const error = yield* produceSqliteBusyError(database.db, streamLockDbPath) + streamLockFailures.set(chat.id, error) + yield* Effect.addFinalizer(() => Effect.sync(() => streamLockFailures.delete(chat.id))) + + const halt = haltLines() + const loggers = yield* Effect.service(Logger.CurrentLoggers) + const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model }) + const result = yield* handle + .process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: { providerID: ref.providerID, modelID: ref.modelID }, + } satisfies SessionV1.User, + sessionID: chat.id, + model, + agent: agent(), + system: [], + messages: [{ role: "user", content: "stream lock origin" }], + tools: {}, + }) + .pipe(Effect.provideService(Logger.CurrentLoggers, new Set([...loggers, halt.logger]))) + + expect(result).toBe("stop") + expect(halt.lines).toHaveLength(1) + expect(String(halt.lines[0].error).toLowerCase()).toContain("database is locked (sqlite_busy)") + const failure = halt.lines[0].failure as Record | undefined + expect(failure).toMatchObject({ stage: "stream" }) + expect(failure).not.toHaveProperty("event") + expect(failure).not.toHaveProperty("case") + // The handled reasoning case wrote its part before the stream failed. + const stored = yield* MessageV2.get({ sessionID: chat.id, messageID: msg.id }) + expect(stored.parts.some((part) => part.type === "reasoning")).toBe(true) + }), + { config: cfg }, + ), + 30_000, +) + itPreludeInterrupt.live("session.processor classifies a prelude interrupt as an aborted assistant", () => provideTmpdirInstance( (dir) => From 3e8dbb592a426679a2009c9964e0f695371cd798 Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 17:12:43 +0000 Subject: [PATCH 13/18] fix(core): omit an absent span from the lock-hold line A hold with no caller span printed span=undefined in the file log. The bun stale-snapshot and BEGIN exhaustion tests now also pin the extended errno bun reports (517 and 5), alongside its code. Claude-Session: https://claude.ai/code/session_01QuyVZfg7zCF6ZJmDUf8BnB --- packages/core/test/sqlite-lock-diagnostics.test.ts | 3 ++- packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/core/test/sqlite-lock-diagnostics.test.ts b/packages/core/test/sqlite-lock-diagnostics.test.ts index 811ea8e48443..ed44b49ae668 100644 --- a/packages/core/test/sqlite-lock-diagnostics.test.ts +++ b/packages/core/test/sqlite-lock-diagnostics.test.ts @@ -339,6 +339,7 @@ describe("SQLite lock retry exhaustion diagnostics", () => { statement: "begin immediate", attempts: attempts.count, "sqlite.code": "SQLITE_BUSY", + "sqlite.errcode": 5, }) expect(attempts.count).toBeGreaterThan(1) expect(summaries[0].fields.elapsedMs).toBeGreaterThanOrEqual(5_000) @@ -477,7 +478,7 @@ describe("SQLite lock retry exhaustion diagnostics", () => { expect(attempts.count).toBe(1) expect(summaries[0].fields).toMatchObject( clientType === "bun" - ? { "sqlite.code": "SQLITE_BUSY_SNAPSHOT" } + ? { "sqlite.code": "SQLITE_BUSY_SNAPSHOT", "sqlite.errcode": 517 } : { "sqlite.code": "ERR_SQLITE_ERROR", "sqlite.errcode": 517 }, ) expect(exhausted(lines)).toEqual([]) diff --git a/packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts b/packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts index a8c4b3e9dff9..36449c53fa8e 100644 --- a/packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts +++ b/packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts @@ -169,7 +169,7 @@ export class EffectSQLiteSession extends SQLite return Effect.logWarning("sqlite write lock held", { pid: process.pid, purpose: Context.get(services, TransactionPurpose) ?? span ?? "unknown", - span, + ...(span && { span }), outcome, durationMs: Math.round(heldMs), }) From 7312e44825338d6ddcbfde08976e743b6f166428 Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 17:53:25 +0000 Subject: [PATCH 14/18] docs: record the stop-instrument diagnostics in CHANGESET.md Adds the fork entry under Runtime fixes now, so review covers its text. The PR number does not exist until publication; script/changeset-check.ts keys the entry on it and fails the PR until `#PR` is replaced with that number. Claude-Session: https://claude.ai/code/session_01QuyVZfg7zCF6ZJmDUf8BnB --- CHANGESET.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/CHANGESET.md b/CHANGESET.md index 0f8ac4551851..612ab4ed7234 100644 --- a/CHANGESET.md +++ b/CHANGESET.md @@ -69,6 +69,11 @@ Newest first. Each entry names the upstream range brought in. ### Runtime fixes +- **#PR** Log SQLite write-lock holds and lock-stop call sites, diagnostics only (firstmate task + opencode-stop-instrument-180, stacked on #194): an immediate transaction holding the write lock over 250 ms logs its + pid, purpose and duration; a lock failure leaving the retry gate logs one summary with the statement head, attempts, + elapsed time and native code; and halt's error line names the failed write's event, stream case and call site. No + behaviour changes. _Fork-only._ - **#194** Fix dropped accepted prompts under concurrent SQLite writers (swxtchio/swx-opencode#162, #165): queue writes begin with BEGIN IMMEDIATE, so a write waits on the lock instead of failing on a snapshot pinned by an earlier read, and the busy-retry gate has one owner in effect-sqlite-node/src/retry.ts. Queue writes on the consumption path use the same lock, which is beyond #167's scope. _Fork-only._ - **#163** Revert #155 (swxtchio/swx-opencode#153): under concurrent writers it dropped more accepted prompts than its parent #150, so the busy waits return to the pre-#155 schedule. The fix-forward is tracked in swxtchio/swx-opencode#162. _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._ From 1cdb1bd2cc6f1442e48e4564650f665ae3a0cf7a Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 17:53:53 +0000 Subject: [PATCH 15/18] docs: say the stop-instrument entry follows #194, which has landed Claude-Session: https://claude.ai/code/session_01QuyVZfg7zCF6ZJmDUf8BnB --- CHANGESET.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGESET.md b/CHANGESET.md index 612ab4ed7234..0822c3a7d814 100644 --- a/CHANGESET.md +++ b/CHANGESET.md @@ -70,7 +70,7 @@ Newest first. Each entry names the upstream range brought in. ### Runtime fixes - **#PR** Log SQLite write-lock holds and lock-stop call sites, diagnostics only (firstmate task - opencode-stop-instrument-180, stacked on #194): an immediate transaction holding the write lock over 250 ms logs its + opencode-stop-instrument-180, follows #194): an immediate transaction holding the write lock over 250 ms logs its pid, purpose and duration; a lock failure leaving the retry gate logs one summary with the statement head, attempts, elapsed time and native code; and halt's error line names the failed write's event, stream case and call site. No behaviour changes. _Fork-only._ From 261a931cfc52540eb23591fbe250dc2888370445 Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 18:31:29 +0000 Subject: [PATCH 16/18] fix(core): name an exhausted statement by its operation only Review round 1, required 1: the retry gate's summary logged the first three SQL tokens, so a leading comment such as /* secret */ or a quoted identifier reached the log and the operation itself could be lost. It now logs only the leading keyword (and a BEGIN's deferred/immediate/exclusive), after skipping leading comments and whitespace; nothing after the operation is read. Returned errors and parameters are unchanged. The new focused test runs real statements through the retry gate on held production connections (leading block and line comments, a string literal, a quoted identifier, mixed case and whitespace, a multi-statement string and a bound parameter) and asserts each summary names its operation and none contains the payload. The first hold test gets an explicit 20 s timeout like its siblings. Claude-Session: https://claude.ai/code/session_01QuyVZfg7zCF6ZJmDUf8BnB --- .../core/test/sqlite-lock-diagnostics.test.ts | 74 ++++++++++++++++++- packages/effect-sqlite-node/src/retry.ts | 13 +++- 2 files changed, 82 insertions(+), 5 deletions(-) diff --git a/packages/core/test/sqlite-lock-diagnostics.test.ts b/packages/core/test/sqlite-lock-diagnostics.test.ts index ed44b49ae668..cf0de7397bc1 100644 --- a/packages/core/test/sqlite-lock-diagnostics.test.ts +++ b/packages/core/test/sqlite-lock-diagnostics.test.ts @@ -187,7 +187,7 @@ describe("SQLite write-lock hold diagnostics", () => { expect(yield* db.values("SELECT id FROM hold_test ORDER BY id")).toEqual([[1], [3]]) }), ), - )) + ), 20_000) test("names an unlabelled hold by its span and keeps nested savepoints inside the outer hold", () => run( @@ -382,7 +382,7 @@ describe("SQLite lock retry exhaustion diagnostics", () => { expect(counts.every((count) => count > 1)).toBe(true) expect(counts[0] + counts[1]).toBe(attempts.count) for (const line of summaries) { - expect(line.fields.statement).toBe("INSERT INTO hold_test") + expect(line.fields.statement).toBe("insert") expect(line.fields.elapsedMs).toBeLessThan(10_000) } }), @@ -391,6 +391,72 @@ describe("SQLite lock retry exhaustion diagnostics", () => { 30_000, ) + test( + "names an exhausted statement by its operation, never its comments, literals, identifiers or parameters", + () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const filename = path.join(tmp.path, "redaction.sqlite") + const setup = yield* productionDatabase(filename) + yield* setup.run("CREATE TABLE redact_test (id INTEGER PRIMARY KEY, v TEXT)") + yield* setup.run(`CREATE TABLE "secret_identifier" (id INTEGER PRIMARY KEY)`) + yield* setup.run("INSERT INTO redact_test (id, v) VALUES (1, NULL)") + const statements = [ + { sql: "/* secret_comment */ INSERT INTO redact_test (id) VALUES (10)", params: [], operation: "insert" }, + { sql: "-- secret_line_comment\nINSERT INTO redact_test (id) VALUES (11)", params: [], operation: "insert" }, + { + sql: "INSERT INTO redact_test (id, v) VALUES (12, 'secret_literal')", + params: [], + operation: "insert", + }, + { sql: `INSERT INTO "secret_identifier" (id) VALUES (13)`, params: [], operation: "insert" }, + { sql: "\n\t bEgIn\r\n ImMeDiAtE ", params: [], operation: "begin immediate" }, + { + sql: "UPDATE redact_test SET v = 'secret_multi' WHERE id = 1; DELETE FROM redact_test", + params: [], + operation: "update", + }, + { + sql: "INSERT INTO redact_test (id, v) VALUES (?, ?)", + params: [14, "secret_parameter"], + operation: "insert", + }, + ] + // One connection per statement, so every statement meets the held lock at once. + const clients = yield* Effect.forEach(statements, () => + productionDatabase(filename).pipe(Effect.map((db) => db.$client)), + ) + yield* inProcessHolder(filename) + + const failures = yield* Effect.forEach( + statements, + (statement, index) => clients[index].unsafe(statement.sql, statement.params).raw.pipe(Effect.flip), + { concurrency: "unbounded" }, + ).pipe( + Effect.timeoutOrElse({ duration: lockBackstop, orElse: () => Effect.die(new Error("retries never stopped")) }), + ) + + // The returned errors keep their existing lock classification and message. + for (const failure of failures) { + expect(isSqlError(failure) && failure.reason._tag).toBe("LockTimeoutError") + expect(failure.message).toBe("Failed to execute statement: database is locked (SQLITE_BUSY)") + } + const summaries = exhausted(lines) + expect(summaries.map((line) => line.fields.statement).toSorted()).toEqual( + statements.map((statement) => statement.operation).toSorted(), + ) + expect(JSON.stringify(summaries)).not.toContain("secret") + }), + ), + ), + 20_000, + ) + for (const clientType of ["core-node", "standalone-node"] as const) { test.skipIf(!nodeSqliteDrivers)( `${clientType} summarizes an exhausted statement once with its full native result code`, @@ -417,7 +483,7 @@ describe("SQLite lock retry exhaustion diagnostics", () => { const summaries = exhausted(lines) expect(summaries).toHaveLength(1) expect(summaries[0].fields).toMatchObject({ - statement: "INSERT INTO hold_test", + statement: "insert", attempts: attempts.count, "sqlite.code": "ERR_SQLITE_ERROR", "sqlite.errcode": 5, @@ -474,7 +540,7 @@ describe("SQLite lock retry exhaustion diagnostics", () => { const summaries = unretried(lines) expect(summaries).toHaveLength(1) - expect(summaries[0].fields).toMatchObject({ statement: "INSERT INTO hold_test", attempts: 1 }) + expect(summaries[0].fields).toMatchObject({ statement: "insert", attempts: 1 }) expect(attempts.count).toBe(1) expect(summaries[0].fields).toMatchObject( clientType === "bun" diff --git a/packages/effect-sqlite-node/src/retry.ts b/packages/effect-sqlite-node/src/retry.ts index a99352257028..01d40ef1575b 100644 --- a/packages/effect-sqlite-node/src/retry.ts +++ b/packages/effect-sqlite-node/src/retry.ts @@ -42,7 +42,7 @@ export const retryLocked = (execute: Effect.Effect, sql: string) typeof error.reason.cause === "object" && error.reason.cause !== null ? error.reason.cause : {} return Effect.logWarning(retryable(error) ? "sqlite lock retries exhausted" : "sqlite lock failed without retry", { pid: process.pid, - statement: sql.trim().split(/\s+/).slice(0, 3).join(" ").slice(0, 64), + statement: statementOperation(sql), attempts, elapsedMs: Math.round(performance.now() - startedAt), // node:sqlite's code is generic, so its full extended result code tells BUSY from BUSY_SNAPSHOT. @@ -52,3 +52,14 @@ export const retryLocked = (execute: Effect.Effect, sql: string) }), ) }) + +// Names a statement by its operation alone: the leading keyword, and a BEGIN's transaction behaviour. Leading comments +// are skipped and nothing after the operation is read, so comments, literals, identifiers and parameters never reach +// the log. +const statementOperation = (sql: string) => { + const statement = sql.replace(/^(?:\s|--[^\n]*|\/\*[\s\S]*?\*\/)*/, "") + const operation = statement.match(/^[a-z]+\b/i)?.[0].toLowerCase() + if (operation !== "begin") return operation ?? "unknown" + const behavior = statement.match(/^begin\s+(deferred|immediate|exclusive)\b/i)?.[1] + return behavior ? `begin ${behavior.toLowerCase()}` : "begin" +} From 4bb7f872aa1b6879c97a750b0d62df81ac66cf57 Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 18:31:29 +0000 Subject: [PATCH 17/18] fix(core): take the stop call site from where the failure was raised Review round 1, required 2: FailureOrigin read the observing fiber's span after the failing scopes had unwound, so a provider-stream failure recorded only SessionProcessor.process. It now reads the stack-frame chain the Effect runtime annotates onto the cause where the failure is raised (Cause.StackTrace), leaving out Effect.fn's definition-site frames, so the call site names the failing callable across the stream and helper boundary. The provider-stream halt test now fails inside a named callable and asserts the call site starts with it; the held-lock tests pin their innermost frame to Session.updatePart and Session.updateMessage. Claude-Session: https://claude.ai/code/session_01QuyVZfg7zCF6ZJmDUf8BnB --- .../core/src/observability/failure-origin.ts | 28 ++++++++++--------- .../test/session/processor-effect.test.ts | 9 +++++- 2 files changed, 23 insertions(+), 14 deletions(-) diff --git a/packages/core/src/observability/failure-origin.ts b/packages/core/src/observability/failure-origin.ts index a081e5e19a43..8bd8b2a1d387 100644 --- a/packages/core/src/observability/failure-origin.ts +++ b/packages/core/src/observability/failure-origin.ts @@ -1,4 +1,4 @@ -import { Cause, Context, Effect, Tracer } from "effect" +import { Cause, Context, Effect } from "effect" // Where a failure came from, for diagnostics only. Keyed by the failure object, so the attribution follows that // failure through Cause squashing and later work cannot overwrite it; the failure itself is never changed. @@ -11,27 +11,29 @@ export type Origin = { const origins = new WeakMap() -// The innermost recording wins each field, and the call site is the span chain where the failure was first seen. +// The innermost recording wins each field. The call site is the stack frame chain the runtime captured where the +// failure was raised, so it names the failing callable even when it is recorded after those scopes have unwound or +// across a stream boundary. export const record = (cause: Cause.Cause, origin: Omit) => - Effect.withFiber((fiber) => { - if (Cause.hasInterruptsOnly(cause)) return Effect.void + Effect.sync(() => { + if (Cause.hasInterruptsOnly(cause)) return const failure = Cause.squash(cause) - if (typeof failure !== "object" || failure === null) return Effect.void - const callSite = spanChain(fiber.context) + if (typeof failure !== "object" || failure === null) return + const callSite = frameChain(cause) origins.set(failure, { ...origin, ...(callSite && { callSite }), ...origins.get(failure) }) - return Effect.void }) export const get = (failure: unknown) => typeof failure === "object" && failure !== null ? origins.get(failure) : undefined -function spanChain(context: Context.Context) { +function frameChain(cause: Cause.Cause) { const names: string[] = [] - const parent = Context.getOption(context, Tracer.ParentSpan) - let span = parent._tag === "Some" ? parent.value : undefined - while (span?._tag === "Span" && names.length < 6) { - names.push(span.name) - span = span.parent._tag === "Some" ? span.parent.value : undefined + const captured = Context.getOption(Cause.annotations(cause), Cause.StackTrace) + let frame = captured._tag === "Some" ? captured.value : undefined + while (frame && names.length < 8) { + // Effect.fn records a definition-site frame beside each call frame; the call frames name the path. + if (!frame.name.endsWith(" (definition)")) names.push(frame.name) + frame = frame.parent } return names.length ? names.join(" < ") : undefined } diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index a24107453f86..9d8111fa15c3 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -679,6 +679,10 @@ const sqliteLockEnv = LayerNode.compile(root, [ const itSqliteLock = testEffect(sqliteLockEnv) // The provider stream dies with the session's lock failure after its handled events all succeed. const streamLockFailures = new Map() +// A named callable inside the provider stream, so the halt line must name it rather than the processor around it. +const failProviderStream = Effect.fn("test.failProviderStream")(function* (sessionID: SessionID) { + return yield* Effect.die(streamLockFailures.get(sessionID)) +}) const itStreamLock = testEffect( LayerNode.compile(root, [ ...replacements, @@ -694,7 +698,7 @@ const itStreamLock = testEffect( LLMEvent.reasoningStart({ id: "reasoning-1" }), LLMEvent.reasoningEnd({ id: "reasoning-1" }), ), - Stream.die(streamLockFailures.get(input.sessionID as SessionID)), + Stream.unwrap(failProviderStream(input.sessionID as SessionID)), ), }), ), @@ -1945,6 +1949,7 @@ for (const scenario of [ expect(failure).toMatchObject({ event: scenario.event, stage: "handler", case: scenario.trigger }) // The span chain runs from the failed commit outward, through the helper the case called. const frames = String(failure?.callSite).split(" < ") + expect(frames[0]).toBe(scenario.callSite[0]) const positions = scenario.callSite.map((name) => frames.indexOf(name)) expect(positions.every((position) => position >= 0)).toBe(true) expect(positions).toEqual(positions.toSorted((a, b) => a - b)) @@ -2008,6 +2013,8 @@ itStreamLock.live( expect(String(halt.lines[0].error).toLowerCase()).toContain("database is locked (sqlite_busy)") const failure = halt.lines[0].failure as Record | undefined expect(failure).toMatchObject({ stage: "stream" }) + // The call site is the failing callable inside the stream, not the processor that observed the failure. + expect(String(failure?.callSite).split(" < ")[0]).toBe("test.failProviderStream") expect(failure).not.toHaveProperty("event") expect(failure).not.toHaveProperty("case") // The handled reasoning case wrote its part before the stream failed. From b8c1bba606ed44604c28e080499bc9eb90f6e3f9 Mon Sep 17 00:00:00 2001 From: Brent Yates Date: Sat, 10 Oct 2026 20:42:21 +0000 Subject: [PATCH 18/18] docs: record the stop-instrument entry as #201 in CHANGESET.md Claude-Session: https://claude.ai/code/session_01QuyVZfg7zCF6ZJmDUf8BnB --- CHANGESET.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGESET.md b/CHANGESET.md index 0822c3a7d814..12f04206c563 100644 --- a/CHANGESET.md +++ b/CHANGESET.md @@ -69,7 +69,7 @@ Newest first. Each entry names the upstream range brought in. ### Runtime fixes -- **#PR** Log SQLite write-lock holds and lock-stop call sites, diagnostics only (firstmate task +- **#201** Log SQLite write-lock holds and lock-stop call sites, diagnostics only (firstmate task opencode-stop-instrument-180, follows #194): an immediate transaction holding the write lock over 250 ms logs its pid, purpose and duration; a lock failure leaving the retry gate logs one summary with the statement head, attempts, elapsed time and native code; and halt's error line names the failed write's event, stream case and call site. No