authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-21 19:29:23-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
logb6d68ee28e03e7fa8f4e69cf325bab6394961122
treeb373e59963e7265435d1adee90967eb981366440
parent2bece01a3be88f5bde92e7ae5633b2d3181955a1

more effect: engine momre effect native

2 files changed, 137 insertions(+), 44 deletions(-)

apps/server/src/orchestration/engine.test.ts+65
...@@ -2,9 +2,12 @@ import fs from "node:fs";...@@ -2,9 +2,12 @@ import fs from "node:fs";
2import os from "node:os";2import os from "node:os";
3import path from "node:path";3import path from "node:path";
44
5import type { OrchestrationEvent } from "@t3tools/contracts";
6import { Effect } from "effect";
5import { afterEach, describe, expect, it } from "vitest";7import { afterEach, describe, expect, it } from "vitest";
68
7import { OrchestrationEngine } from "./engine";9import { OrchestrationEngine } from "./engine";
10import type { OrchestrationEventStore } from "./eventStore";
811
9const tempDirs: string[] = [];12const tempDirs: string[] = [];
1013
...@@ -310,4 +313,66 @@ describe("OrchestrationEngine", () => {...@@ -310,4 +313,66 @@ describe("OrchestrationEngine", () => {
310 await engine.stop();313 await engine.stop();
311 await expect(engine.stop()).resolves.toBeUndefined();314 await expect(engine.stop()).resolves.toBeUndefined();
312 });315 });
316
317 it("keeps processing queued commands after a storage failure", async () => {
318 const events: OrchestrationEvent[] = [];
319 let nextSequence = 1;
320 let shouldFailFirstAppend = true;
321
322 const flakyStore: OrchestrationEventStore = {
323 append(event) {
324 if (shouldFailFirstAppend) {
325 shouldFailFirstAppend = false;
326 return Effect.sync(() => {
327 throw new Error("append failed");
328 });
329 }
330 const savedEvent = {
331 ...event,
332 sequence: nextSequence,
333 } satisfies OrchestrationEvent;
334 nextSequence += 1;
335 events.push(savedEvent);
336 return Effect.succeed(savedEvent);
337 },
338 readFromSequence(sequenceExclusive) {
339 return Effect.succeed(events.filter((event) => event.sequence > sequenceExclusive));
340 },
341 readAll() {
342 return Effect.succeed(events);
343 },
344 close() {},
345 };
346
347 const engine = new OrchestrationEngine(flakyStore);
348 const createdAt = new Date().toISOString();
349 await engine.start();
350
351 await expect(
352 engine.dispatch({
353 type: "project.create",
354 commandId: "cmd-flaky-1",
355 projectId: "project-flaky-fail",
356 name: "flaky-fail",
357 cwd: "/tmp/flaky-fail",
358 model: "gpt-5-codex",
359 createdAt,
360 }),
361 ).rejects.toThrow("append failed");
362
363 const result = await engine.dispatch({
364 type: "project.create",
365 commandId: "cmd-flaky-2",
366 projectId: "project-flaky-ok",
367 name: "flaky-ok",
368 cwd: "/tmp/flaky-ok",
369 model: "gpt-5-codex",
370 createdAt,
371 });
372
373 expect(result.sequence).toBe(1);
374 expect(engine.getSnapshot().projects.map((project) => project.id)).toEqual(["project-flaky-ok"]);
375
376 await engine.stop();
377 });
313});378});
apps/server/src/orchestration/engine.ts+72-44
...@@ -6,7 +6,18 @@ import type {...@@ -6,7 +6,18 @@ import type {
6 OrchestrationReadModel,6 OrchestrationReadModel,
7} from "@t3tools/contracts";7} from "@t3tools/contracts";
8import { OrchestrationCommandSchema } from "@t3tools/contracts";8import { OrchestrationCommandSchema } from "@t3tools/contracts";
9import { PubSub, Queue, Schema, Stream, Effect, Fiber, Runtime, Either } from "effect";9import {
10 PubSub,
11 Queue,
12 Schema,
13 Stream,
14 Effect,
15 Fiber,
16 Runtime,
17 Either,
18 Deferred,
19 Cause,
20} from "effect";
1021
11import { createLogger } from "../logger";22import { createLogger } from "../logger";
12import type { OrchestrationEventStore } from "./eventStore";23import type { OrchestrationEventStore } from "./eventStore";
...@@ -16,10 +27,13 @@ import { UI_ENTITY_CONTRACTS } from "./uiContractInventory";...@@ -16,10 +27,13 @@ import { UI_ENTITY_CONTRACTS } from "./uiContractInventory";
1627
17type CommandEnvelope = {28type CommandEnvelope = {
18 command: OrchestrationCommand;29 command: OrchestrationCommand;
19 resolve: (result: { sequence: number }) => void;30 result: Deferred.Deferred<{ sequence: number }, Error>;
20 reject: (error: Error) => void;
21};31};
2232
33function asError(error: unknown, fallbackMessage: string): Error {
34 return error instanceof Error ? error : new Error(fallbackMessage);
35}
36
23function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEvent, "sequence"> {37function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEvent, "sequence"> {
24 const eventId = crypto.randomUUID();38 const eventId = crypto.randomUUID();
25 switch (command.type) {39 switch (command.type) {
...@@ -218,43 +232,57 @@ export class OrchestrationEngine {...@@ -218,43 +232,57 @@ export class OrchestrationEngine {
218 this.eventPubSub = Runtime.runSync(this.runtime)(PubSub.unbounded<OrchestrationEvent>());232 this.eventPubSub = Runtime.runSync(this.runtime)(PubSub.unbounded<OrchestrationEvent>());
219 }233 }
220234
235 private processEnvelope(envelope: CommandEnvelope): Effect.Effect<void> {
236 return Effect.gen(this, function* () {
237 const eventBase = mapCommandToEvent(envelope.command);
238 const savedEvent = yield* this.eventStore.append(eventBase);
239
240 yield* Effect.sync(() => {
241 this.readModel = reduceEvent(this.readModel, savedEvent);
242 });
243
244 const snapshot = this.readModel;
245 yield* Effect.all([
246 PubSub.publish(this.eventPubSub, savedEvent),
247 PubSub.publish(this.readModelPubSub, snapshot),
248 ]);
249
250 yield* Effect.sync(() => {
251 for (const listener of this.domainEventListeners) {
252 listener(savedEvent);
253 }
254 for (const listener of this.readModelListeners) {
255 listener(snapshot);
256 }
257 });
258
259 yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence });
260 }).pipe(
261 Effect.catchAllCause((cause) =>
262 Deferred.fail(
263 envelope.result,
264 asError(Cause.squash(cause), "Unknown command processing error"),
265 ).pipe(Effect.asVoid),
266 ),
267 );
268 }
269
270 private bootstrapReadModel(): Effect.Effect<void> {
271 return Effect.gen(this, function* () {
272 const existingEvents = yield* this.eventStore.readAll();
273 yield* Effect.sync(() => {
274 for (const event of existingEvents) {
275 this.readModel = reduceEvent(this.readModel, event);
276 }
277 });
278 });
279 }
280
221 async start(): Promise<void> {281 async start(): Promise<void> {
222 const existingEvents = await Runtime.runPromise(this.runtime)(this.eventStore.readAll());282 await Runtime.runPromise(this.runtime)(this.bootstrapReadModel());
223 for (const event of existingEvents) {
224 this.readModel = reduceEvent(this.readModel, event);
225 }
226283
227 const worker = Stream.fromQueue(this.commandQueue).pipe(284 const worker = Stream.fromQueue(this.commandQueue).pipe(
228 Stream.runForEach((envelope) =>285 Stream.runForEach((envelope) => this.processEnvelope(envelope)),
229 Effect.tryPromise({
230 try: async () => {
231 const eventBase = mapCommandToEvent(envelope.command);
232 const savedEvent = await Runtime.runPromise(this.runtime)(
233 this.eventStore.append(eventBase),
234 );
235 this.readModel = reduceEvent(this.readModel, savedEvent);
236 await Promise.all([
237 Runtime.runPromise(this.runtime)(PubSub.publish(this.eventPubSub, savedEvent)),
238 Runtime.runPromise(this.runtime)(
239 PubSub.publish(this.readModelPubSub, this.readModel),
240 ),
241 ]);
242 for (const listener of this.domainEventListeners) {
243 listener(savedEvent);
244 }
245 for (const listener of this.readModelListeners) {
246 listener(this.readModel);
247 }
248 envelope.resolve({ sequence: savedEvent.sequence });
249 },
250 catch: (error) => {
251 const message =
252 error instanceof Error ? error.message : "Unknown command processing error";
253 envelope.reject(new Error(message));
254 return undefined;
255 },
256 }),
257 ),
258 );286 );
259287
260 this.workerFiber = Runtime.runFork(this.runtime)(worker);288 this.workerFiber = Runtime.runFork(this.runtime)(worker);
...@@ -301,13 +329,13 @@ export class OrchestrationEngine {...@@ -301,13 +329,13 @@ export class OrchestrationEngine {
301 }329 }
302330
303 async dispatch(command: OrchestrationCommand): Promise<{ sequence: number }> {331 async dispatch(command: OrchestrationCommand): Promise<{ sequence: number }> {
304 return new Promise<{ sequence: number }>((resolve, reject) => {332 const program = Effect.gen(this, function* () {
305 Runtime.runPromise(this.runtime)(333 const result = yield* Deferred.make<{ sequence: number }, Error>();
306 Queue.offer(this.commandQueue, { command, resolve, reject }),334 yield* Queue.offer(this.commandQueue, { command, result });
307 ).catch((error) => {335 return yield* Deferred.await(result);
308 const message = error instanceof Error ? error.message : "Queue offer failed";336 });
309 reject(new Error(message));337 return Runtime.runPromise(this.runtime)(program).catch((error) => {
310 });338 throw asError(error, "Queue offer failed");
311 });339 });
312 }340 }
313341