more effect: engine momre effect native

Julius Marminge committed 8 months ago

Commit 35758ec1 parent 4bddf3a2 files · +137 −44Browse files

Unified · Split

2 files

apps/server/src/orchestration/engine.test.ts +65 −0
apps/server/src/orchestration/engine.ts +72 −44

apps/server/src/orchestration/engine.test.ts+65−0

@@ -2,9 +2,12 @@ import fs from "node:fs"; · 1 unmodified line
2	2	import os from "node:os";
3	3	import path from "node:path";
4	4
	5	+import type { OrchestrationEvent } from "@t3tools/contracts";
	6	+import { Effect } from "effect";
5	7	import { afterEach, describe, expect, it } from "vitest";
6	8
7	9	import { OrchestrationEngine } from "./engine";
	10	+import type { OrchestrationEventStore } from "./eventStore";
8	11
9	12	const tempDirs: string[] = [];
10	13
@@ -310,4 +313,66 @@ describe("OrchestrationEngine", () => { · 299 unmodified lines
310	313	await engine.stop();
311	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 { · 5 unmodified lines
6	6	OrchestrationReadModel,
7	7	} from "@t3tools/contracts";
8	8	import { OrchestrationCommandSchema } from "@t3tools/contracts";
9		−import { PubSub, Queue, Schema, Stream, Effect, Fiber, Runtime, Either } from "effect";
	9	+import {
	10	+ PubSub,
	11	+ Queue,
	12	+ Schema,
	13	+ Stream,
	14	+ Effect,
	15	+ Fiber,
	16	+ Runtime,
	17	+ Either,
	18	+ Deferred,
	19	+ Cause,
	20	+} from "effect";
10	21
11	22	import { createLogger } from "../logger";
12	23	import type { OrchestrationEventStore } from "./eventStore";
@@ -16,10 +27,13 @@ import { UI_ENTITY_CONTRACTS } from "./uiContractInventory"; · 3 unmodified lines
16	27
17	28	type CommandEnvelope = {
18	29	command: OrchestrationCommand;
19		− resolve: (result: { sequence: number }) => void;
20		− reject: (error: Error) => void;
	30	+ result: Deferred.Deferred<{ sequence: number }, Error>;
21	31	};
22	32
	33	+function asError(error: unknown, fallbackMessage: string): Error {
	34	+ return error instanceof Error ? error : new Error(fallbackMessage);
	35	+}
	36	+
23	37	function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEvent, "sequence"> {
24	38	const eventId = crypto.randomUUID();
25	39	switch (command.type) {
@@ -218,43 +232,57 @@ export class OrchestrationEngine { · 192 unmodified lines
218	232	this.eventPubSub = Runtime.runSync(this.runtime)(PubSub.unbounded<OrchestrationEvent>());
219	233	}
220	234
	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	281	async start(): Promise<void> {
222		− const existingEvents = await Runtime.runPromise(this.runtime)(this.eventStore.readAll());
223		− for (const event of existingEvents) {
224		− this.readModel = reduceEvent(this.readModel, event);
225		− }
	282	+ await Runtime.runPromise(this.runtime)(this.bootstrapReadModel());
226	283
227	284	const worker = Stream.fromQueue(this.commandQueue).pipe(
228		− Stream.runForEach((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		− ),
	285	+ Stream.runForEach((envelope) => this.processEnvelope(envelope)),
258	286	);
259	287
260	288	this.workerFiber = Runtime.runFork(this.runtime)(worker);
@@ -301,13 +329,13 @@ export class OrchestrationEngine { · 40 unmodified lines
301	329	}
302	330
303	331	async dispatch(command: OrchestrationCommand): Promise<{ sequence: number }> {
304		− return new Promise<{ sequence: number }>((resolve, reject) => {
305		− Runtime.runPromise(this.runtime)(
306		− Queue.offer(this.commandQueue, { command, resolve, reject }),
307		− ).catch((error) => {
308		− const message = error instanceof Error ? error.message : "Queue offer failed";
309		− reject(new Error(message));
310		− });
	332	+ const program = Effect.gen(this, function* () {
	333	+ const result = yield* Deferred.make<{ sequence: number }, Error>();
	334	+ yield* Queue.offer(this.commandQueue, { command, result });
	335	+ return yield* Deferred.await(result);
	336	+ });
	337	+ return Runtime.runPromise(this.runtime)(program).catch((error) => {
	338	+ throw asError(error, "Queue offer failed");
311	339	});
312	340	}
313	341

more effect: engine momre effect native

Keyboard shortcuts

General

Issues and pull requests

Issues, pull requests and Inbox

Repository code