History

fix OrchestrationEngine pre-transaction error handling

Julius Marminge committed 7 months ago

Commit 563808b1 parent 90f9c9a2 files · +175 −75Browse files

Unified · Split

2 files
  1. apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +100 −1
  2. apps/server/src/orchestration/Layers/OrchestrationEngine.ts +75 −74
apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+100−1
@@ -8,13 +8,17 @@ import { · 7 unmodified lines
88 TurnId,
99 type OrchestrationEvent,
1010 } from "@t3tools/contracts";
11−import { Effect, Layer, ManagedRuntime, Queue, Stream } from "effect";
11+import { Effect, Layer, ManagedRuntime, Option, Queue, Stream } from "effect";
1212 import { describe, expect, it } from "vitest";
1313
1414 import { PersistenceSqlError } from "../../persistence/Errors.ts";
1515 import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";
1616 import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";
1717 import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";
18+import {
19+ OrchestrationCommandReceiptRepository,
20+ type OrchestrationCommandReceiptRepositoryShape,
21+} from "../../persistence/Services/OrchestrationCommandReceipts.ts";
1822 import {
1923 OrchestrationEventStore,
2024 type OrchestrationEventStoreShape,
@@ -488,6 +492,101 @@ describe("OrchestrationEngine", () => { · 467 unmodified lines
488492 await runtime.dispose();
489493 });
490494
495+ it("keeps processing queued commands after a receipt lookup failure", async () => {
496+ const failingReceiptRepository: OrchestrationCommandReceiptRepositoryShape = {
497+ upsert: () => Effect.void,
498+ getByCommandId: ({ commandId }) => {
499+ if (commandId === CommandId.makeUnsafe("cmd-receipt-fail")) {
500+ return Effect.fail(
501+ new PersistenceSqlError({
502+ operation: "test.getByCommandId",
503+ detail: "receipt lookup failed",
504+ }),
505+ );
506+ }
507+ return Effect.succeed(Option.none());
508+ },
509+ };
510+
511+ const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), {
512+ prefix: "t3-orchestration-engine-test-",
513+ });
514+
515+ const runtime = ManagedRuntime.make(
516+ OrchestrationEngineLive.pipe(
517+ Layer.provide(OrchestrationProjectionPipelineLive),
518+ Layer.provide(OrchestrationEventStoreLive),
519+ Layer.provide(
520+ Layer.succeed(OrchestrationCommandReceiptRepository, failingReceiptRepository),
521+ ),
522+ Layer.provide(SqlitePersistenceMemory),
523+ Layer.provideMerge(ServerConfigLayer),
524+ Layer.provideMerge(NodeServices.layer),
525+ ),
526+ );
527+ const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));
528+ const createdAt = now();
529+
530+ await runtime.runPromise(
531+ engine.dispatch({
532+ type: "project.create",
533+ commandId: CommandId.makeUnsafe("cmd-project-receipt-create"),
534+ projectId: asProjectId("project-receipt"),
535+ title: "Receipt Project",
536+ workspaceRoot: "/tmp/project-receipt",
537+ defaultModelSelection: {
538+ provider: "codex",
539+ model: "gpt-5-codex",
540+ },
541+ createdAt,
542+ }),
543+ );
544+
545+ await expect(
546+ runtime.runPromise(
547+ engine.dispatch({
548+ type: "thread.create",
549+ commandId: CommandId.makeUnsafe("cmd-receipt-fail"),
550+ threadId: ThreadId.makeUnsafe("thread-receipt-fail"),
551+ projectId: asProjectId("project-receipt"),
552+ title: "receipt-fail",
553+ modelSelection: {
554+ provider: "codex",
555+ model: "gpt-5-codex",
556+ },
557+ interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
558+ runtimeMode: "approval-required",
559+ branch: null,
560+ worktreePath: null,
561+ createdAt,
562+ }),
563+ ),
564+ ).rejects.toThrow("receipt lookup failed");
565+
566+ const result = await runtime.runPromise(
567+ engine.dispatch({
568+ type: "thread.create",
569+ commandId: CommandId.makeUnsafe("cmd-receipt-ok"),
570+ threadId: ThreadId.makeUnsafe("thread-receipt-ok"),
571+ projectId: asProjectId("project-receipt"),
572+ title: "receipt-ok",
573+ modelSelection: {
574+ provider: "codex",
575+ model: "gpt-5-codex",
576+ },
577+ interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
578+ runtimeMode: "approval-required",
579+ branch: null,
580+ worktreePath: null,
581+ createdAt,
582+ }),
583+ );
584+
585+ expect(result.sequence).toBe(2);
586+ expect((await runtime.runPromise(engine.getReadModel())).snapshotSequence).toBe(2);
587+ await runtime.dispose();
588+ });
589+
491590 it("rolls back all events for a multi-event command when projection fails mid-dispatch", async () => {
492591 let shouldFailRequestedProjection = true;
493592 const flakyProjectionPipeline: OrchestrationProjectionPipelineShape = {
apps/server/src/orchestration/Layers/OrchestrationEngine.ts+75−74
@@ -118,86 +118,87 @@ const makeOrchestrationEngine = Effect.gen(function* () { · 117 unmodified lines
118118 yield* Deferred.fail(envelope.result, error);
119119 });
120120
121− const existingReceipt = yield* commandReceiptRepository.getByCommandId({
122− commandId: envelope.command.commandId,
123− });
124− if (Option.isSome(existingReceipt)) {
125− if (existingReceipt.value.status === "accepted") {
126− yield* Deferred.succeed(envelope.result, {
127− sequence: existingReceipt.value.resultSequence,
128− });
121+ return yield* Effect.gen(function* () {
122+ const existingReceipt = yield* commandReceiptRepository.getByCommandId({
123+ commandId: envelope.command.commandId,
124+ });
125+ if (Option.isSome(existingReceipt)) {
126+ if (existingReceipt.value.status === "accepted") {
127+ yield* Deferred.succeed(envelope.result, {
128+ sequence: existingReceipt.value.resultSequence,
129+ });
130+ return;
131+ }
132+ yield* Deferred.fail(
133+ envelope.result,
134+ new OrchestrationCommandPreviouslyRejectedError({
135+ commandId: envelope.command.commandId,
136+ detail: existingReceipt.value.error ?? "Previously rejected.",
137+ }),
138+ );
129139 return;
130140 }
131− yield* Deferred.fail(
132− envelope.result,
133− new OrchestrationCommandPreviouslyRejectedError({
134− commandId: envelope.command.commandId,
135− detail: existingReceipt.value.error ?? "Previously rejected.",
136− }),
137− );
138− return;
139− }
140141
141− const eventBase = yield* decideOrchestrationCommand({
142− command: envelope.command,
143− readModel,
144− });
145− const eventBases = Array.isArray(eventBase) ? eventBase : [eventBase];
146− const committedCommand = yield* sql
147− .withTransaction(
148− Effect.gen(function* () {
149− const committedEvents: OrchestrationEvent[] = [];
150− let nextReadModel = readModel;
151−
152− for (const nextEvent of eventBases) {
153− const savedEvent = yield* eventStore.append(nextEvent);
154− nextReadModel = yield* projectEvent(nextReadModel, savedEvent);
155− yield* projectionPipeline.projectEvent(savedEvent);
156− committedEvents.push(savedEvent);
157− }
158−
159− const lastSavedEvent = committedEvents.at(-1) ?? null;
160− if (lastSavedEvent === null) {
161− return yield* new OrchestrationCommandInvariantError({
162− commandType: envelope.command.type,
163− detail: "Command produced no events.",
142+ const eventBase = yield* decideOrchestrationCommand({
143+ command: envelope.command,
144+ readModel,
145+ });
146+ const eventBases = Array.isArray(eventBase) ? eventBase : [eventBase];
147+ const committedCommand = yield* sql
148+ .withTransaction(
149+ Effect.gen(function* () {
150+ const committedEvents: OrchestrationEvent[] = [];
151+ let nextReadModel = readModel;
152+
153+ for (const nextEvent of eventBases) {
154+ const savedEvent = yield* eventStore.append(nextEvent);
155+ nextReadModel = yield* projectEvent(nextReadModel, savedEvent);
156+ yield* projectionPipeline.projectEvent(savedEvent);
157+ committedEvents.push(savedEvent);
158+ }
159+
160+ const lastSavedEvent = committedEvents.at(-1) ?? null;
161+ if (lastSavedEvent === null) {
162+ return yield* new OrchestrationCommandInvariantError({
163+ commandType: envelope.command.type,
164+ detail: "Command produced no events.",
165+ });
166+ }
167+
168+ yield* commandReceiptRepository.upsert({
169+ commandId: envelope.command.commandId,
170+ aggregateKind: lastSavedEvent.aggregateKind,
171+ aggregateId: lastSavedEvent.aggregateId,
172+ acceptedAt: lastSavedEvent.occurredAt,
173+ resultSequence: lastSavedEvent.sequence,
174+ status: "accepted",
175+ error: null,
164176 });
165− }
166177
167− yield* commandReceiptRepository.upsert({
168− commandId: envelope.command.commandId,
169− aggregateKind: lastSavedEvent.aggregateKind,
170− aggregateId: lastSavedEvent.aggregateId,
171− acceptedAt: lastSavedEvent.occurredAt,
172− resultSequence: lastSavedEvent.sequence,
173− status: "accepted",
174− error: null,
175− });
176−
177− return {
178− committedEvents,
179− lastSequence: lastSavedEvent.sequence,
180− nextReadModel,
181− } as const;
182− }),
183− )
184− .pipe(
185− Effect.catchTag("SqlError", (sqlError) =>
186− Effect.fail(
187− toPersistenceSqlError("OrchestrationEngine.processEnvelope:transaction")(sqlError),
178+ return {
179+ committedEvents,
180+ lastSequence: lastSavedEvent.sequence,
181+ nextReadModel,
182+ } as const;
183+ }),
184+ )
185+ .pipe(
186+ Effect.catchTag("SqlError", (sqlError) =>
187+ Effect.fail(
188+ toPersistenceSqlError("OrchestrationEngine.processEnvelope:transaction")(sqlError),
189+ ),
188190 ),
189− ),
190− Effect.catch(handleError),
191− );
192− if (!committedCommand) {
193− return;
194− }
195−
196− readModel = committedCommand.nextReadModel;
197− for (const event of committedCommand.committedEvents) {
198− yield* PubSub.publish(eventPubSub, event);
199− }
200− yield* Deferred.succeed(envelope.result, { sequence: committedCommand.lastSequence });
191+ );
192+ if (!committedCommand) {
193+ return;
194+ }
195+
196+ readModel = committedCommand.nextReadModel;
197+ for (const event of committedCommand.committedEvents) {
198+ yield* PubSub.publish(eventPubSub, event);
199+ }
200+ yield* Deferred.succeed(envelope.result, { sequence: committedCommand.lastSequence });
201+ }).pipe(Effect.catch(handleError), Effect.asVoid);
201202 });
202203
203204 yield* projectionPipeline.bootstrap;