diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index cb130930372c8..f0f09e199badf 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -16,11 +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"; @@ -379,6 +383,138 @@ 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 firstLockTimeout = yield* Deferred.make(); + 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) { + yield* Deferred.succeed(firstLockTimeout, undefined); + 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)), + ); + + const workerFiber = yield* EffectWorker.OrchestrationEffectWorkerV2.pipe( + Effect.flatMap((worker) => worker.runOnce), + Effect.provide(layerWorker), + 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 d174db2fb3a29..985fff7dfbe62 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -8,7 +8,9 @@ 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"; import { increment, @@ -596,6 +598,16 @@ 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.interruptible( + Effect.retry(settlement, { schedule: lockRetrySchedule, while: isLockTimeout }), + ); + export interface OrchestrationEffectWorkerV2Shape { readonly awaitWork: Effect.Effect; readonly runOnce: Effect.Effect; @@ -654,6 +666,7 @@ export const layerWithOptions = ( delayMs: 0, }) .pipe( + retryWhileLocked, Effect.flatMap((requeued) => requeued ? Effect.logWarning("Requeued effect after unexpected worker failure", { @@ -685,6 +698,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", {