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/2] 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/2] 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;