From 5a08b9dd64e5f13a9ad88c910e9f34cb7598199a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 25 Feb 2026 20:40:51 -0800 Subject: [PATCH] run in transaction --- .plans/16c-pr89-remediation-checklist.md | 4 +- .../OrchestrationEngine.test.ts} | 127 ++++++++++++++++-- .../Layers/OrchestrationEngine.ts | 74 ++++++---- 3 files changed, 170 insertions(+), 35 deletions(-) rename apps/server/src/orchestration/{Layer.test.ts => Layers/OrchestrationEngine.test.ts} (75%) diff --git a/.plans/16c-pr89-remediation-checklist.md b/.plans/16c-pr89-remediation-checklist.md index 5ebe2fe235451e5229650ef57640e690222677d2..61b8c2c9c5b99848fb7fff9d563a6b40957d947c 100644 --- a/.plans/16c-pr89-remediation-checklist.md +++ b/.plans/16c-pr89-remediation-checklist.md @@ -154,8 +154,8 @@ Counts: active `53` (`valid=34`, `partially-valid=19`), closed-invalid `5` ### Phase 2 -- [ ] `C001` Non-atomic event appending can corrupt state on retry. If an error occurs mid-loop (lines 96-102) after some events are persisted but before the receipt is written, the command appears to fail. A retry generates new UUIDs via `crypto.randomUUID()` in the decider, appending duplicate events. Consider wrapping the loop in a transaction or using deterministic event IDs derived from `commandId`.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent: - - Status: `TODO` +- [x] `C001` Non-atomic event appending can corrupt state on retry. If an error occurs mid-loop (lines 96-102) after some events are persisted but before the receipt is written, the command appears to fail. A retry generates new UUIDs via `crypto.randomUUID()` in the decider, appending duplicate events. Consider wrapping the loop in a transaction or using deterministic event IDs derived from `commandId`.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent: + - Status: `DONE` - Verdict: `valid` - Severity: `High` - Area: `Event ordering and state consistency` diff --git a/apps/server/src/orchestration/Layer.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts similarity index 75% rename from apps/server/src/orchestration/Layer.test.ts rename to apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index 1ccf75579fa93ef7cae818baeafe2ff3d22ef5a0..e2c8ab35f5789e1a5281cce93367d7a310ada88c 100644 --- a/apps/server/src/orchestration/Layer.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -10,17 +10,21 @@ import { import { Effect, Layer, ManagedRuntime, Queue, Stream } from "effect"; import { describe, expect, it } from "vitest"; -import { PersistenceSqlError } from "../persistence/Errors.ts"; -import { OrchestrationCommandReceiptRepositoryLive } from "../persistence/Layers/OrchestrationCommandReceipts.ts"; -import { OrchestrationEventStoreLive } from "../persistence/Layers/OrchestrationEventStore.ts"; -import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; +import { PersistenceSqlError } from "../../persistence/Errors.ts"; +import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; +import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts"; +import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; import { OrchestrationEventStore, type OrchestrationEventStoreShape, -} from "../persistence/Services/OrchestrationEventStore.ts"; -import { OrchestrationEngineLive } from "./Layers/OrchestrationEngine.ts"; -import { OrchestrationProjectionPipelineLive } from "./Layers/ProjectionPipeline.ts"; -import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts"; +} from "../../persistence/Services/OrchestrationEventStore.ts"; +import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; +import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts"; +import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; +import { + OrchestrationProjectionPipeline, + type OrchestrationProjectionPipelineShape, +} from "../Services/ProjectionPipeline.ts"; const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); @@ -356,6 +360,113 @@ describe("OrchestrationEngine", () => { await runtime.dispose(); }); + it("rolls back all events for a multi-event command when projection fails mid-dispatch", async () => { + let shouldFailRequestedProjection = true; + const flakyProjectionPipeline: OrchestrationProjectionPipelineShape = { + bootstrap: Effect.void, + projectEvent: (event) => { + if ( + shouldFailRequestedProjection && + event.commandId === CommandId.makeUnsafe("cmd-turn-start-atomic") && + event.type === "thread.turn-start-requested" + ) { + shouldFailRequestedProjection = false; + return Effect.fail( + new PersistenceSqlError({ + operation: "test.projection", + detail: "projection failed", + }), + ); + } + return Effect.void; + }, + }; + + const runtime = ManagedRuntime.make( + OrchestrationEngineLive.pipe( + Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)), + Layer.provide(OrchestrationEventStoreLive), + Layer.provide(OrchestrationCommandReceiptRepositoryLive), + Layer.provide(SqlitePersistenceMemory), + ), + ); + const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); + const createdAt = now(); + + await runtime.runPromise( + engine.dispatch({ + type: "project.create", + commandId: CommandId.makeUnsafe("cmd-project-atomic-create"), + projectId: asProjectId("project-atomic"), + title: "Atomic Project", + workspaceRoot: "/tmp/project-atomic", + defaultModel: "gpt-5-codex", + createdAt, + }), + ); + await runtime.runPromise( + engine.dispatch({ + type: "thread.create", + commandId: CommandId.makeUnsafe("cmd-thread-atomic-create"), + threadId: ThreadId.makeUnsafe("thread-atomic"), + projectId: asProjectId("project-atomic"), + title: "atomic", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }), + ); + + const turnStartCommand = { + type: "thread.turn.start" as const, + commandId: CommandId.makeUnsafe("cmd-turn-start-atomic"), + threadId: ThreadId.makeUnsafe("thread-atomic"), + message: { + messageId: asMessageId("msg-atomic-1"), + role: "user" as const, + text: "hello", + attachments: [], + }, + createdAt, + }; + + await expect(runtime.runPromise(engine.dispatch(turnStartCommand))).rejects.toThrow( + "projection failed", + ); + + const eventsAfterFailure = await runtime.runPromise( + Stream.runCollect(engine.readEvents(0)).pipe( + Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)), + ), + ); + expect(eventsAfterFailure.map((event) => event.type)).toEqual([ + "project.created", + "thread.created", + ]); + expect((await runtime.runPromise(engine.getReadModel())).snapshotSequence).toBe(2); + + const retryResult = await runtime.runPromise(engine.dispatch(turnStartCommand)); + expect(retryResult.sequence).toBe(4); + + const eventsAfterRetry = await runtime.runPromise( + Stream.runCollect(engine.readEvents(0)).pipe( + Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)), + ), + ); + expect(eventsAfterRetry.map((event) => event.type)).toEqual([ + "project.created", + "thread.created", + "thread.message-sent", + "thread.turn-start-requested", + ]); + expect( + eventsAfterRetry.filter((event) => event.commandId === turnStartCommand.commandId), + ).toHaveLength(2); + + await runtime.dispose(); + }); + it("fails command dispatch when command invariants are violated", async () => { const system = await createOrchestrationSystem(); const { engine } = system; diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index 52734f063fb41b7d1f5ff312a5222cfd5b1a00c7..ebe5547700892ad3deb86eacfaf83af5990f4a5e 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -6,7 +6,9 @@ import type { } from "@t3tools/contracts"; import { OrchestrationCommand } from "@t3tools/contracts"; import { Deferred, Effect, Layer, Option, PubSub, Queue, Stream } from "effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import { toPersistenceSqlError } from "../../persistence/Errors.ts"; import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts"; import { OrchestrationCommandReceiptRepository } from "../../persistence/Services/OrchestrationCommandReceipts.ts"; import { @@ -55,6 +57,7 @@ function formatDispatchError(error: OrchestrationDispatchError): string { } const makeOrchestrationEngine = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; const eventStore = yield* OrchestrationEventStore; const commandReceiptRepository = yield* OrchestrationCommandReceiptRepository; const projectionPipeline = yield* OrchestrationProjectionPipeline; @@ -91,36 +94,57 @@ const makeOrchestrationEngine = Effect.gen(function* () { readModel, }); const eventBases = Array.isArray(eventBase) ? eventBase : [eventBase]; - let lastSavedEvent = null as OrchestrationEvent | null; + const committedCommand = yield* sql + .withTransaction( + Effect.gen(function* () { + const committedEvents: OrchestrationEvent[] = []; + let nextReadModel = readModel; - for (const nextEvent of eventBases) { - const savedEvent = yield* eventStore.append(nextEvent); - yield* projectionPipeline.projectEvent(savedEvent); - readModel = yield* projectEvent(readModel, savedEvent); - yield* PubSub.publish(eventPubSub, savedEvent); - lastSavedEvent = savedEvent; - } + for (const nextEvent of eventBases) { + const savedEvent = yield* eventStore.append(nextEvent); + yield* projectionPipeline.projectEvent(savedEvent); + nextReadModel = yield* projectEvent(nextReadModel, savedEvent); + committedEvents.push(savedEvent); + } - if (lastSavedEvent === null) { - return yield* Deferred.fail( - envelope.result, - new OrchestrationCommandInvariantError({ - commandType: envelope.command.type, - detail: "Command produced no events.", + const lastSavedEvent = committedEvents.at(-1) ?? null; + if (lastSavedEvent === null) { + return yield* new OrchestrationCommandInvariantError({ + commandType: envelope.command.type, + detail: "Command produced no events.", + }); + } + + yield* commandReceiptRepository.upsert({ + commandId: envelope.command.commandId, + aggregateKind: lastSavedEvent.aggregateKind, + aggregateId: lastSavedEvent.aggregateId, + acceptedAt: lastSavedEvent.occurredAt, + resultSequence: lastSavedEvent.sequence, + status: "accepted", + error: null, + }); + + return { + committedEvents, + lastSequence: lastSavedEvent.sequence, + nextReadModel, + } as const; }), + ) + .pipe( + Effect.catchTag("SqlError", (sqlError) => + Effect.fail( + toPersistenceSqlError("OrchestrationEngine.processEnvelope:transaction")(sqlError), + ), + ), ); - } - yield* commandReceiptRepository.upsert({ - commandId: envelope.command.commandId, - aggregateKind: lastSavedEvent.aggregateKind, - aggregateId: lastSavedEvent.aggregateId, - acceptedAt: lastSavedEvent.occurredAt, - resultSequence: lastSavedEvent.sequence, - status: "accepted", - error: null, - }); - yield* Deferred.succeed(envelope.result, { sequence: lastSavedEvent.sequence }); + readModel = committedCommand.nextReadModel; + for (const event of committedCommand.committedEvents) { + yield* PubSub.publish(eventPubSub, event); + } + yield* Deferred.succeed(envelope.result, { sequence: committedCommand.lastSequence }); }).pipe( Effect.catch((error) => Effect.gen(function* () { -- 2.54.0