From 60511c1384e2db2770afa74a73310e16c5332146 Mon Sep 17 00:00:00 2001 From: Rene Franke <132335976+ReneFr81@users.noreply.github.com> Date: Wed, 28 Oct 2026 20:42:19 +0100 Subject: [PATCH 1/3] fix(server): a locked database no longer strands a claimed effect and stalls its thread When an effect fails and the worker's write to requeue it hits SQLite's lock timeout, the worker logged the error and dropped the claim. The row stayed `running`, and a running row blocks every later effect on its thread, so the next run's provider-turn.start was never claimed and the thread sat in "starting" until the server restarted again. The fallback requeue and terminalize writes now retry while the failure is a lock timeout. Only the claiming worker can settle its row, and each lock timeout has already waited out busy_timeout. --- .../src/orchestration-v2/EffectWorker.test.ts | 69 +++++++++++++++++++ .../src/orchestration-v2/EffectWorker.ts | 10 +++ 2 files changed, 79 insertions(+) diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index cb130930372c8..7f70e498585a1 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -21,6 +21,7 @@ import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; +import * as SqlError from "effect/sql/SqlError"; import * as CheckpointRollbackService from "./CheckpointRollbackService.ts"; import * as CheckpointService from "./CheckpointService.ts"; @@ -379,6 +380,74 @@ it.effect("requeues a claim when a pre-execution worker check fails", () => }), ); +it.effect("keeps requeueing a failed claim while the database is locked", () => + Effect.gen(function* () { + const now = DateTime.formatIso(yield* DateTime.now); + const effectId = "effect:worker-requeue-lock-timeout"; + const workerId = "worker-requeue-lock-timeout"; + const claimedEffect: EffectOutbox.OrchestrationEffectV2 = { + id: effectId, + commandId: CommandId.make("command:worker-requeue-lock-timeout"), + threadId, + request: { type: "provider-runtime.continue", sourceRunId: runId }, + status: "running", + attemptCount: 1, + availableAt: now, + leaseOwner: workerId, + createdAt: now, + updatedAt: now, + completedAt: null, + lastError: null, + }; + const lockedWrites = yield* Ref.make(3); + const requeued = yield* Ref.make(false); + const layerOutbox = Layer.mock(EffectOutbox.EffectOutboxV2)({ + claimNext: () => Effect.succeedSome(claimedEffect), + get: () => Effect.succeedSome(claimedEffect), + awaitCancellation: () => Effect.never, + clearCancellation: () => Effect.void, + retry: () => + Effect.gen(function* () { + if ((yield* Ref.getAndUpdate(lockedWrites, (count) => count - 1)) > 0) { + return yield* new EffectOutbox.EffectOutboxError({ + operation: "retry", + effectId, + cause: new SqlError.SqlError({ + reason: new SqlError.LockTimeoutError({ cause: null }), + }), + }); + } + yield* Ref.set(requeued, true); + return true; + }), + }); + const layerExecutor = Layer.succeed( + EffectWorker.OrchestrationEffectExecutorV2, + EffectWorker.OrchestrationEffectExecutorV2.of({ + execute: () => + Effect.fail( + new EffectWorker.OrchestrationEffectExecutionError({ + effectId, + effectType: "provider-runtime.continue", + cause: "simulated database is locked", + }), + ), + }), + ); + const layerWorker = EffectWorker.layerWithOptions({ workerId }).pipe( + Layer.provide(Layer.merge(layerOutbox, layerExecutor)), + ); + + yield* EffectWorker.OrchestrationEffectWorkerV2.pipe( + Effect.flatMap((worker) => worker.runOnce), + Effect.provide(layerWorker), + Effect.exit, + ); + + assert.isTrue(yield* Ref.get(requeued)); + }), +); + it.effect("arms cancellation before the durable pre-execution check", () => Effect.gen(function* () { const now = DateTime.formatIso(yield* DateTime.now); diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index d174db2fb3a29..e1681c1cb1f0c 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -9,6 +9,7 @@ import * as Layer from "effect/Layer"; import * as Metric from "effect/Metric"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; +import * as SqlError from "effect/sql/SqlError"; import { increment, @@ -596,6 +597,13 @@ export class OrchestrationEffectWorkerError extends Schema.TaggedError + SqlError.isSqlError(error.cause) && error.cause.reason._tag === "LockTimeoutError"; +const retryWhileLocked = (settlement: Effect.Effect) => + Effect.retry(settlement, { while: isLockTimeout }); + export interface OrchestrationEffectWorkerV2Shape { readonly awaitWork: Effect.Effect; readonly runOnce: Effect.Effect; @@ -654,6 +662,7 @@ export const layerWithOptions = ( delayMs: 0, }) .pipe( + retryWhileLocked, Effect.flatMap((requeued) => requeued ? Effect.logWarning("Requeued effect after unexpected worker failure", { @@ -685,6 +694,7 @@ export const layerWithOptions = ( error: `Worker failed to settle a process-bound effect after execution started: ${Cause.pretty(cause)}`, }) .pipe( + retryWhileLocked, Effect.flatMap((failed) => failed ? Effect.logError("Terminalized process-bound effect after settlement failure", { From bcba32c62043a2a0cb65b2e778e77d41e0e6469c Mon Sep 17 00:00:00 2001 From: Rene Franke <132335976+ReneFr81@users.noreply.github.com> Date: Fri, 30 Oct 2026 09:29:52 +0100 Subject: [PATCH 2/3] fix(server): stop lock retries during shutdown Error cleanup masks interruption, so settlement retries could keep the worker group alive until SQLite released its lock. Restore interruption and space lock retries by 100 ms to avoid spinning on immediate errors. --- .../src/orchestration-v2/EffectWorker.test.ts | 71 ++++++++++++++++++- .../src/orchestration-v2/EffectWorker.ts | 8 ++- 2 files changed, 75 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index 7f70e498585a1..f0f09e199badf 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -16,12 +16,15 @@ import { import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import * as SqlError from "effect/sql/SqlError"; +import * as TestClock from "effect/testing/TestClock"; import * as CheckpointRollbackService from "./CheckpointRollbackService.ts"; import * as CheckpointService from "./CheckpointService.ts"; @@ -401,6 +404,7 @@ it.effect("keeps requeueing a failed claim while the database is locked", () => }; const lockedWrites = yield* Ref.make(3); const requeued = yield* Ref.make(false); + const firstLockTimeout = yield* Deferred.make(); const layerOutbox = Layer.mock(EffectOutbox.EffectOutboxV2)({ claimNext: () => Effect.succeedSome(claimedEffect), get: () => Effect.succeedSome(claimedEffect), @@ -409,6 +413,7 @@ it.effect("keeps requeueing a failed claim while the database is locked", () => retry: () => Effect.gen(function* () { if ((yield* Ref.getAndUpdate(lockedWrites, (count) => count - 1)) > 0) { + yield* Deferred.succeed(firstLockTimeout, undefined); return yield* new EffectOutbox.EffectOutboxError({ operation: "retry", effectId, @@ -438,16 +443,78 @@ it.effect("keeps requeueing a failed claim while the database is locked", () => Layer.provide(Layer.merge(layerOutbox, layerExecutor)), ); - yield* EffectWorker.OrchestrationEffectWorkerV2.pipe( + const workerFiber = yield* EffectWorker.OrchestrationEffectWorkerV2.pipe( Effect.flatMap((worker) => worker.runOnce), Effect.provide(layerWorker), - Effect.exit, + Effect.forkChild, ); + yield* Deferred.await(firstLockTimeout); + yield* TestClock.adjust(Duration.millis(99)); + assert.isFalse(yield* Ref.get(requeued)); + yield* TestClock.adjust(Duration.seconds(1)); + yield* Effect.exit(Fiber.join(workerFiber)); assert.isTrue(yield* Ref.get(requeued)); }), ); +it.effect.each([ + { name: "requeue", request: { type: "provider-runtime.continue", sourceRunId: runId } }, + { name: "terminalize", request: { type: "provider-turn.start", runId } }, +] as const)("interrupts $name settlement retries when the worker shuts down", ({ request }) => + Effect.gen(function* () { + const claimedEffect: EffectOutbox.OrchestrationEffectV2 = { + ...restartEffect(yield* DateTime.now, { type: "detach" }), + request, + }; + const settlementStarted = yield* Deferred.make(); + const releaseLock = yield* Ref.make(false); + const lockedWrite = Effect.gen(function* () { + yield* Deferred.succeed(settlementStarted, undefined); + // An in-flight database write can finish its own wait before noticing shutdown. + yield* Effect.sleep(Duration.millis(1)); + if (yield* Ref.get(releaseLock)) return true; + return yield* new EffectOutbox.EffectOutboxError({ + operation: "settlement", + effectId: claimedEffect.id, + cause: new SqlError.SqlError({ reason: new SqlError.LockTimeoutError({ cause: null }) }), + }); + }); + const layerOutbox = Layer.mock(EffectOutbox.EffectOutboxV2)({ + claimNext: () => Effect.succeedSome(claimedEffect), + get: () => Effect.succeedSome(claimedEffect), + awaitCancellation: () => Effect.never, + clearCancellation: () => Effect.void, + succeed: () => Effect.fail(new EffectOutbox.EffectOutboxError({ operation: "succeed" })), + retry: () => lockedWrite, + fail: () => lockedWrite, + }); + const layerExecutor = Layer.succeed( + EffectWorker.OrchestrationEffectExecutorV2, + EffectWorker.OrchestrationEffectExecutorV2.of({ execute: () => Effect.void }), + ); + const layerWorker = EffectWorker.layerWithOptions({ workerId: "test-worker" }).pipe( + Layer.provide(Layer.merge(layerOutbox, layerExecutor)), + ); + + const daemon = yield* EffectWorker.runDaemonWithOptions({ concurrency: 1 }).pipe( + Effect.provide(layerWorker), + Effect.forkChild, + ); + yield* Deferred.await(settlementStarted); + const shutdown = yield* Fiber.interrupt(daemon).pipe( + Effect.timeoutOption(Duration.millis(100)), + Effect.forkChild, + ); + yield* TestClock.adjust(Duration.millis(100)); + const stopped = yield* Fiber.join(shutdown); + yield* Ref.set(releaseLock, true); + yield* TestClock.adjust(Duration.seconds(1)); + yield* Fiber.await(daemon); + assert.isTrue(Option.isSome(stopped)); + }), +); + it.effect("arms cancellation before the durable pre-execution check", () => Effect.gen(function* () { const now = DateTime.formatIso(yield* DateTime.now); diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index e1681c1cb1f0c..985fff7dfbe62 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -8,6 +8,7 @@ import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as Metric from "effect/Metric"; import * as Option from "effect/Option"; +import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; import * as SqlError from "effect/sql/SqlError"; @@ -598,11 +599,14 @@ export class OrchestrationEffectWorkerError extends Schema.TaggedError SqlError.isSqlError(error.cause) && error.cause.reason._tag === "LockTimeoutError"; +const lockRetrySchedule = Schedule.spaced(Duration.millis(100)); const retryWhileLocked = (settlement: Effect.Effect) => - Effect.retry(settlement, { while: isLockTimeout }); + Effect.interruptible( + Effect.retry(settlement, { schedule: lockRetrySchedule, while: isLockTimeout }), + ); export interface OrchestrationEffectWorkerV2Shape { readonly awaitWork: Effect.Effect; From 2b686809eb7988292fd4a250ed67e503e69a7ef8 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 10:18:18 +0000 Subject: [PATCH 3/3] fix(shared): a write that timed out on a lock no longer wedges the SQLite writer When a RETURNING statement threw while stepping in the worker (SQLITE_BUSY after busy_timeout), the statement stayed active on the writer connection. Every later COMMIT then failed with "SQL statements in progress", leaving the transaction open so each next BEGIN failed too. Reset the iterator when it throws. Co-Authored-By: Claude Opus 5.5 --- packages/shared/src/nodeSqliteClient.test.ts | 19 +++++++++++++++++++ packages/shared/src/nodeSqliteClient.ts | 8 +++++++- 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index ac65e7f12daa7..b7b9b7a0866a2 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -240,6 +240,25 @@ it.effect("reuses a worker query after an iterator constraint failure", () => }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), ); +it.effect("commits on the worker writer after a returning write timed out on a lock", () => + Effect.gen(function* () { + const { filename, other } = yield* makeTempDatabase; + yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`PRAGMA busy_timeout = 0`; + yield* Effect.sync(() => other.exec("BEGIN IMMEDIATE")); + const error = yield* sql`UPDATE counters SET value = 1 WHERE id = 1 RETURNING value`.pipe( + Effect.flip, + ); + assert.equal(error.reason._tag, "LockTimeoutError"); + yield* Effect.sync(() => other.exec("COMMIT")); + + yield* sql.withTransaction(sql`UPDATE counters SET value = 2 WHERE id = 1`); + assert.deepEqual(yield* sql`SELECT value FROM counters WHERE id = 1`.values, [[2]]); + }).pipe(Effect.provide(SqliteClient.layer({ filename, writerWorker: true }))); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + it.effect("keeps another connection from committing between a transaction's read and write", () => Effect.gen(function* () { const { filename, other } = yield* makeTempDatabase; diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index fda878a114d9a..ab14675c7bfc0 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -244,6 +244,7 @@ parentPort.on("message", (request) => { } const { id, sql, params, cached, values, raw, safeIntegers, received } = request; let operation = "open"; + let iterator; try { if (db === undefined) throw new Error("database connection is unavailable"); operation = "prepare"; @@ -255,7 +256,8 @@ parentPort.on("message", (request) => { let rows = []; let bytes = 0; let pendingBatch = false; - for (const row of statement.iterate(...params)) { + iterator = statement.iterate(...params); + for (const row of iterator) { let rowBytes = 0; for (const key in row) { const value = row[key]; @@ -281,6 +283,10 @@ parentPort.on("message", (request) => { parentPort.postMessage({ id, rows: raw ? result : [] }); } } catch (error) { + // A step that throws leaves its statement active; an active write blocks every later COMMIT. + try { + iterator?.return(); + } catch {} // A failed iterator can report its previous SQLite error when the statement is reused. statements.delete(sql); parentPort.postMessage({ id, error: { operation, ...describe(error) } });