diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index cc43601543eec..c753e07d4689a 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -16,12 +16,16 @@ 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 Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import * as CheckpointRollbackService from "./CheckpointRollbackService.ts"; import * as CheckpointService from "./CheckpointService.ts"; @@ -396,6 +400,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 9dd5079805745..bbb214542540d 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -9,7 +9,9 @@ import * as Fiber from "effect/Fiber"; 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, @@ -667,6 +669,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 }), + ); + type ClaimFilter = Omit[0], "workerId">; export interface OrchestrationEffectWorkerV2Shape { @@ -728,6 +740,7 @@ export const layerWithOptions = ( delayMs: 0, }) .pipe( + retryWhileLocked, Effect.flatMap((requeued) => requeued ? Effect.logWarning("Requeued effect after unexpected worker failure", { @@ -759,6 +772,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", { 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) } });