diff --git a/apps/server/src/orchestration/engine.test.ts b/apps/server/src/orchestration/engine.test.ts index d4a0123169c3665ba1d1ee69482473aed6e88420..05cf57c72f2824c33fc0276f2bacd901e74c8886 100644 --- a/apps/server/src/orchestration/engine.test.ts +++ b/apps/server/src/orchestration/engine.test.ts @@ -2,9 +2,12 @@ import fs from "node:fs"; import os from "node:os"; import path from "node:path"; +import type { OrchestrationEvent } from "@t3tools/contracts"; +import { Effect } from "effect"; import { afterEach, describe, expect, it } from "vitest"; import { OrchestrationEngine } from "./engine"; +import type { OrchestrationEventStore } from "./eventStore"; const tempDirs: string[] = []; @@ -310,4 +313,66 @@ describe("OrchestrationEngine", () => { await engine.stop(); await expect(engine.stop()).resolves.toBeUndefined(); }); + + it("keeps processing queued commands after a storage failure", async () => { + const events: OrchestrationEvent[] = []; + let nextSequence = 1; + let shouldFailFirstAppend = true; + + const flakyStore: OrchestrationEventStore = { + append(event) { + if (shouldFailFirstAppend) { + shouldFailFirstAppend = false; + return Effect.sync(() => { + throw new Error("append failed"); + }); + } + const savedEvent = { + ...event, + sequence: nextSequence, + } satisfies OrchestrationEvent; + nextSequence += 1; + events.push(savedEvent); + return Effect.succeed(savedEvent); + }, + readFromSequence(sequenceExclusive) { + return Effect.succeed(events.filter((event) => event.sequence > sequenceExclusive)); + }, + readAll() { + return Effect.succeed(events); + }, + close() {}, + }; + + const engine = new OrchestrationEngine(flakyStore); + const createdAt = new Date().toISOString(); + await engine.start(); + + await expect( + engine.dispatch({ + type: "project.create", + commandId: "cmd-flaky-1", + projectId: "project-flaky-fail", + name: "flaky-fail", + cwd: "/tmp/flaky-fail", + model: "gpt-5-codex", + createdAt, + }), + ).rejects.toThrow("append failed"); + + const result = await engine.dispatch({ + type: "project.create", + commandId: "cmd-flaky-2", + projectId: "project-flaky-ok", + name: "flaky-ok", + cwd: "/tmp/flaky-ok", + model: "gpt-5-codex", + createdAt, + }); + + expect(result.sequence).toBe(1); + expect(engine.getSnapshot().projects.map((project) => project.id)).toEqual(["project-flaky-ok"]); + + await engine.stop(); + }); }); diff --git a/apps/server/src/orchestration/engine.ts b/apps/server/src/orchestration/engine.ts index 9021aa45030dad21a7d2bb9046e586eb18949dfe..5b8590eea5fc8b4b954756dbcedbb6c319c7e72b 100644 --- a/apps/server/src/orchestration/engine.ts +++ b/apps/server/src/orchestration/engine.ts @@ -6,7 +6,18 @@ import type { OrchestrationReadModel, } from "@t3tools/contracts"; import { OrchestrationCommandSchema } from "@t3tools/contracts"; -import { PubSub, Queue, Schema, Stream, Effect, Fiber, Runtime, Either } from "effect"; +import { + PubSub, + Queue, + Schema, + Stream, + Effect, + Fiber, + Runtime, + Either, + Deferred, + Cause, +} from "effect"; import { createLogger } from "../logger"; import type { OrchestrationEventStore } from "./eventStore"; @@ -16,10 +27,13 @@ import { UI_ENTITY_CONTRACTS } from "./uiContractInventory"; type CommandEnvelope = { command: OrchestrationCommand; - resolve: (result: { sequence: number }) => void; - reject: (error: Error) => void; + result: Deferred.Deferred<{ sequence: number }, Error>; }; +function asError(error: unknown, fallbackMessage: string): Error { + return error instanceof Error ? error : new Error(fallbackMessage); +} + function mapCommandToEvent(command: OrchestrationCommand): Omit { const eventId = crypto.randomUUID(); switch (command.type) { @@ -218,43 +232,57 @@ export class OrchestrationEngine { this.eventPubSub = Runtime.runSync(this.runtime)(PubSub.unbounded()); } + private processEnvelope(envelope: CommandEnvelope): Effect.Effect { + return Effect.gen(this, function* () { + const eventBase = mapCommandToEvent(envelope.command); + const savedEvent = yield* this.eventStore.append(eventBase); + + yield* Effect.sync(() => { + this.readModel = reduceEvent(this.readModel, savedEvent); + }); + + const snapshot = this.readModel; + yield* Effect.all([ + PubSub.publish(this.eventPubSub, savedEvent), + PubSub.publish(this.readModelPubSub, snapshot), + ]); + + yield* Effect.sync(() => { + for (const listener of this.domainEventListeners) { + listener(savedEvent); + } + for (const listener of this.readModelListeners) { + listener(snapshot); + } + }); + + yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence }); + }).pipe( + Effect.catchAllCause((cause) => + Deferred.fail( + envelope.result, + asError(Cause.squash(cause), "Unknown command processing error"), + ).pipe(Effect.asVoid), + ), + ); + } + + private bootstrapReadModel(): Effect.Effect { + return Effect.gen(this, function* () { + const existingEvents = yield* this.eventStore.readAll(); + yield* Effect.sync(() => { + for (const event of existingEvents) { + this.readModel = reduceEvent(this.readModel, event); + } + }); + }); + } + async start(): Promise { - const existingEvents = await Runtime.runPromise(this.runtime)(this.eventStore.readAll()); - for (const event of existingEvents) { - this.readModel = reduceEvent(this.readModel, event); - } + await Runtime.runPromise(this.runtime)(this.bootstrapReadModel()); const worker = Stream.fromQueue(this.commandQueue).pipe( - Stream.runForEach((envelope) => - Effect.tryPromise({ - try: async () => { - const eventBase = mapCommandToEvent(envelope.command); - const savedEvent = await Runtime.runPromise(this.runtime)( - this.eventStore.append(eventBase), - ); - this.readModel = reduceEvent(this.readModel, savedEvent); - await Promise.all([ - Runtime.runPromise(this.runtime)(PubSub.publish(this.eventPubSub, savedEvent)), - Runtime.runPromise(this.runtime)( - PubSub.publish(this.readModelPubSub, this.readModel), - ), - ]); - for (const listener of this.domainEventListeners) { - listener(savedEvent); - } - for (const listener of this.readModelListeners) { - listener(this.readModel); - } - envelope.resolve({ sequence: savedEvent.sequence }); - }, - catch: (error) => { - const message = - error instanceof Error ? error.message : "Unknown command processing error"; - envelope.reject(new Error(message)); - return undefined; - }, - }), - ), + Stream.runForEach((envelope) => this.processEnvelope(envelope)), ); this.workerFiber = Runtime.runFork(this.runtime)(worker); @@ -301,13 +329,13 @@ export class OrchestrationEngine { } async dispatch(command: OrchestrationCommand): Promise<{ sequence: number }> { - return new Promise<{ sequence: number }>((resolve, reject) => { - Runtime.runPromise(this.runtime)( - Queue.offer(this.commandQueue, { command, resolve, reject }), - ).catch((error) => { - const message = error instanceof Error ? error.message : "Queue offer failed"; - reject(new Error(message)); - }); + const program = Effect.gen(this, function* () { + const result = yield* Deferred.make<{ sequence: number }, Error>(); + yield* Queue.offer(this.commandQueue, { command, result }); + return yield* Deferred.await(result); + }); + return Runtime.runPromise(this.runtime)(program).catch((error) => { + throw asError(error, "Queue offer failed"); }); }