diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index e163b5c1eb6bc..01e3eeff57286 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -61,6 +61,7 @@ import * as TextGeneration from "../textGeneration/TextGeneration.ts"; import { ClaudeProviderCapabilitiesV2 } from "./Adapters/ClaudeAdapterV2.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as CommandReceiptStore from "./CommandReceiptStore.ts"; +import * as Orchestrator from "./Orchestrator.ts"; import * as EffectOutbox from "./EffectOutbox.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; @@ -138,6 +139,10 @@ interface HarnessOptions { readonly getLaunchSettings?: ServerSettings.ServerSettingsService["Service"]["getSettings"]; readonly providers?: ReadonlyArray; readonly git?: Partial; + readonly dispatch?: ( + command: Parameters[0], + dispatch: ThreadManagement.ThreadManagementService["Service"]["dispatch"], + ) => ReturnType; readonly isGitRepository?: CheckpointStore.CheckpointStore["Service"]["isGitRepository"]; } @@ -245,11 +250,25 @@ function makeHarness(options: HarnessOptions = {}) { folderForThread: () => Effect.succeedNone, }), ); + const interceptDispatch = options.dispatch; + const layerLaunchThreads = + interceptDispatch === undefined + ? layerThreadManagement + : Layer.effect( + ThreadManagement.ThreadManagementService, + Effect.gen(function* () { + const threads = yield* ThreadManagement.ThreadManagementService; + return ThreadManagement.ThreadManagementService.of({ + ...threads, + dispatch: (command) => interceptDispatch(command, threads.dispatch), + }); + }), + ).pipe(Layer.provide(layerThreadManagement)); const layerLaunch = ThreadLaunch.layer.pipe( Layer.provide( Layer.mergeAll( layerExternalServices, - layerThreadManagement, + layerLaunchThreads, layerReceipts, layerOrchestrator, IdAllocator.layer, @@ -1676,6 +1695,73 @@ it.effect.each(["fails", "is interrupted"] as const)( }), ); +it.effect("fails a run whose preparation failure could not be stored until storage recovers", () => + Effect.gen(function* () { + const storageFull = yield* Deferred.make(); + let rejectedWrites = 0; + let preparations = 0; + const harness = makeHarness({ + fetchWorktreeBaseCommit: () => + Effect.sync(() => preparations++).pipe( + Effect.andThen( + Effect.fail( + new GitCommandError({ + operation: "GitVcsDriver.fetchRemote", + command: "git", + cwd: project.workspaceRoot, + detail: "Git could not reach the remote.", + exitCode: 128, + }), + ), + ), + ), + dispatch: (command, dispatch) => + command.type !== "prepared-run.fail" + ? dispatch(command) + : Effect.suspend(() => { + if (rejectedWrites >= 2) return dispatch(command); + rejectedWrites += 1; + return Deferred.succeed(storageFull, undefined).pipe( + Effect.andThen( + Effect.fail( + new Orchestrator.OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: new Error("SQLITE_FULL: database or disk is full"), + }), + ), + ), + ); + }), + }); + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + const input = launchInput({ + command: "command:launch:storage-full", + thread: "thread:launch:storage-full", + message: "First message", + workspace: { type: "worktree", baseRef: "main", startFromOrigin: true }, + }); + const launched = yield* launches.launch(input); + yield* Deferred.await(storageFull); + yield* TestClock.adjust("0 seconds"); + yield* launches.launch(input); + yield* TestClock.adjust("0 seconds"); + assert.equal(preparations, 1); + yield* TestClock.adjust("1 second"); + yield* threads.streamStoredEventsFrom({ threadId: launched.threadId }).pipe( + Stream.filter( + (stored) => + stored.event.type === "run.updated" && stored.event.payload.status === "failed", + ), + Stream.runHead, + ); + assert.equal(rejectedWrites, 2); + }).pipe(Effect.provide(harness.layer)); + }), +); + it.effect.each(["failed", "interrupted"] as const)( "holds every message that arrives after the setup %s without a worktree", (ending) => { diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index eb07bb1de165a..89179505840f8 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -25,12 +25,14 @@ import * as Cause from "effect/Cause"; import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; +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 Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; @@ -986,36 +988,72 @@ const make = Effect.gen(function* () { threadId: ThreadId, runId: RunId | null, cause: unknown, - ) => - runId === null - ? Effect.logWarning("Thread workspace preparation failed", { - commandId: input.commandId, - threadId, - cause, - }) - : threads - .dispatch({ - type: "prepared-run.fail", - commandId: CommandId.make(`${input.commandId}:fail`), + ) => { + if (runId === null) { + return Effect.logWarning("Thread workspace preparation failed", { + commandId: input.commandId, + threadId, + cause, + }); + } + const persistFailure = threads.dispatch({ + type: "prepared-run.fail", + commandId: CommandId.make(`${input.commandId}:fail`), + threadId, + runId, + failure: makeProviderFailure({ + cause, + message: failureDetail(cause), + class: "validation_error", + retryable: false, + }), + }); + const runIsPreparing = threads.getThreadRecords(threadId, ["runs"], { runIds: [runId] }).pipe( + Effect.map((current) => + current.runs.some((run) => run.id === runId && run.status === "preparing"), + ), + // The read can fail in the same outage, which says nothing about the run. + Effect.catchCause(() => Effect.succeed(true)), + ); + return persistFailure.pipe( + Effect.catchCause((persistCause) => + Effect.gen(function* () { + yield* Effect.logWarning("Failed to persist thread workspace preparation failure", { + commandId: input.commandId, threadId, - runId, - failure: makeProviderFailure({ - cause, - message: failureDetail(cause), - class: "validation_error", - retryable: false, - }), - }) - .pipe( - Effect.catchCause((persistCause) => - Effect.logWarning("Failed to persist thread workspace preparation failure", { - commandId: input.commandId, - threadId, - cause, - persistCause, - }), + cause, + persistCause, + }); + if (Cause.hasInterruptsOnly(persistCause) || !(yield* runIsPreparing)) return; + // A full disk rejects this write as well, and nothing else moves the run out of + // "preparing", so it would hold the thread's queue until a restart. Retry until + // storage recovers or the run settles another way. + yield* persistFailure.pipe( + // Storage failures can be defects; retry both failure channels. + Effect.catchCauseIf( + (retryCause) => !Cause.hasInterruptsOnly(retryCause), + (retryCause) => + runIsPreparing.pipe( + Effect.flatMap((preparing) => + preparing ? Effect.fail(retryCause) : Effect.void, + ), + ), ), + Effect.retry( + Schedule.exponential("1 second").pipe( + Schedule.modifyDelay(({ duration }) => + Effect.succeed(Duration.min(duration, Duration.seconds(30))), + ), + ), + ), + Effect.ignoreCause, + // Callers reach this from finalizers; shutdown must still be able to stop it. + Effect.interruptible, ); + }), + ), + ); + }; const reservePreparation = (commandId: CommandId) => Ref.modify(scheduledLaunches, (scheduled) => {