authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-04-01 15:07:15-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-04-01 15:07:15-07:00
log823d69f6215d80ce4e75ac4fb9a8f763d5d27b87
tree15329fa554e90daadb2b8452445025ee619277f5
parent4963fccb05c9ae46fc5153bc76008d7c8e21ba0a
signature Signed by PGP key B5690EEEBB952194

perf(server): bootstrap engine from persisted projections (#1648)

Co-authored-by: codex <codex@users.noreply.github.com>

8 files changed, 141 insertions(+), 17 deletions(-)

apps/server/integration/OrchestrationEngineHarness.integration.ts+6-3
......@@ -289,9 +289,10 @@ export const makeOrchestrationIntegrationHarness = (
289289 );
290290
291291 const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(GitCoreLive));
292 const projectionSnapshotQueryLayer = OrchestrationProjectionSnapshotQueryLive;
292293 const runtimeServicesLayer = Layer.mergeAll(
293 orchestrationLayer,
294 OrchestrationProjectionSnapshotQueryLive,
294 projectionSnapshotQueryLayer,
295 orchestrationLayer.pipe(Layer.provide(projectionSnapshotQueryLayer)),
295296 ProjectionCheckpointRepositoryLive,
296297 ProjectionPendingApprovalRepositoryLive,
297298 checkpointStoreLayer,
......@@ -333,7 +334,9 @@ export const makeOrchestrationIntegrationHarness = (
333334 Layer.provideMerge(providerCommandReactorLayer),
334335 Layer.provideMerge(checkpointReactorLayer),
335336 );
336 const layer = orchestrationReactorLayer.pipe(
337 const layer = Layer.empty.pipe(
338 Layer.provideMerge(runtimeServicesLayer),
339 Layer.provideMerge(orchestrationReactorLayer),
337340 Layer.provide(persistenceLayer),
338341 Layer.provideMerge(ServerSettingsService.layerTest()),
339342 Layer.provideMerge(ServerConfig.layerTest(workspaceDir, rootDir)),
apps/server/src/orchestration/Layers/CheckpointReactor.test.ts+2
......@@ -23,6 +23,7 @@ import { GitCoreLive } from "../../git/Layers/GitCore.ts";
2323import { CheckpointReactorLive } from "./CheckpointReactor.ts";
2424import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
2525import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
26import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
2627import { RuntimeReceiptBusLive } from "./RuntimeReceiptBus.ts";
2728import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";
2829import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";
......@@ -249,6 +250,7 @@ describe("CheckpointReactor", () => {
249250 options?.providerName ?? "codex",
250251 );
251252 const orchestrationLayer = OrchestrationEngineLive.pipe(
253 Layer.provide(OrchestrationProjectionSnapshotQueryLive),
252254 Layer.provide(OrchestrationProjectionPipelineLive),
253255 Layer.provide(OrchestrationEventStoreLive),
254256 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+106-1
......@@ -8,7 +8,7 @@ import {
88 TurnId,
99 type OrchestrationEvent,
1010} from "@t3tools/contracts";
11import { Effect, Layer, ManagedRuntime, Queue, Stream } from "effect";
11import { Effect, Layer, ManagedRuntime, Option, Queue, Stream } from "effect";
1212import { describe, expect, it } from "vitest";
1313
1414import { PersistenceSqlError } from "../../persistence/Errors.ts";
......@@ -21,11 +21,13 @@ import {
2121} from "../../persistence/Services/OrchestrationEventStore.ts";
2222import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
2323import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
24import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
2425import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
2526import {
2627 OrchestrationProjectionPipeline,
2728 type OrchestrationProjectionPipelineShape,
2829} from "../Services/ProjectionPipeline.ts";
30import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";
2931import { ServerConfig } from "../../config.ts";
3032import * as NodeServices from "@effect/platform-node/NodeServices";
3133
......@@ -39,6 +41,7 @@ async function createOrchestrationSystem() {
3941 prefix: "t3-orchestration-engine-test-",
4042 });
4143 const orchestrationLayer = OrchestrationEngineLive.pipe(
44 Layer.provide(OrchestrationProjectionSnapshotQueryLive),
4245 Layer.provide(OrchestrationProjectionPipelineLive),
4346 Layer.provide(OrchestrationEventStoreLive),
4447 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
......@@ -60,6 +63,105 @@ function now() {
6063}
6164
6265describe("OrchestrationEngine", () => {
66 it("bootstraps the in-memory read model from persisted projections", async () => {
67 const failOnHistoricalReplayStore: OrchestrationEventStoreShape = {
68 append: () =>
69 Effect.fail(
70 new PersistenceSqlError({
71 operation: "test.append",
72 detail: "append should not be called during bootstrap",
73 }),
74 ),
75 readFromSequence: () => Stream.empty,
76 readAll: () =>
77 Stream.fail(
78 new PersistenceSqlError({
79 operation: "test.readAll",
80 detail: "historical replay should not be used during bootstrap",
81 }),
82 ),
83 };
84
85 const projectionSnapshot = {
86 snapshotSequence: 7,
87 updatedAt: "2026-03-03T00:00:04.000Z",
88 projects: [
89 {
90 id: asProjectId("project-bootstrap"),
91 title: "Bootstrap Project",
92 workspaceRoot: "/tmp/project-bootstrap",
93 defaultModelSelection: {
94 provider: "codex" as const,
95 model: "gpt-5-codex",
96 },
97 scripts: [],
98 createdAt: "2026-03-03T00:00:00.000Z",
99 updatedAt: "2026-03-03T00:00:01.000Z",
100 deletedAt: null,
101 },
102 ],
103 threads: [
104 {
105 id: ThreadId.makeUnsafe("thread-bootstrap"),
106 projectId: asProjectId("project-bootstrap"),
107 title: "Bootstrap Thread",
108 modelSelection: {
109 provider: "codex" as const,
110 model: "gpt-5-codex",
111 },
112 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
113 runtimeMode: "full-access" as const,
114 branch: null,
115 worktreePath: null,
116 latestTurn: null,
117 createdAt: "2026-03-03T00:00:02.000Z",
118 updatedAt: "2026-03-03T00:00:03.000Z",
119 archivedAt: null,
120 deletedAt: null,
121 messages: [],
122 proposedPlans: [],
123 activities: [],
124 checkpoints: [],
125 session: null,
126 },
127 ],
128 };
129
130 const layer = OrchestrationEngineLive.pipe(
131 Layer.provide(
132 Layer.succeed(ProjectionSnapshotQuery, {
133 getSnapshot: () => Effect.succeed(projectionSnapshot),
134 getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }),
135 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
136 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
137 getThreadCheckpointContext: () => Effect.succeed(Option.none()),
138 }),
139 ),
140 Layer.provide(
141 Layer.succeed(OrchestrationProjectionPipeline, {
142 bootstrap: Effect.void,
143 projectEvent: () => Effect.void,
144 } satisfies OrchestrationProjectionPipelineShape),
145 ),
146 Layer.provide(Layer.succeed(OrchestrationEventStore, failOnHistoricalReplayStore)),
147 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
148 Layer.provide(SqlitePersistenceMemory),
149 );
150
151 const runtime = ManagedRuntime.make(layer);
152
153 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));
154 const readModel = await runtime.runPromise(engine.getReadModel());
155
156 expect(readModel.snapshotSequence).toBe(7);
157 expect(readModel.projects).toHaveLength(1);
158 expect(readModel.projects[0]?.title).toBe("Bootstrap Project");
159 expect(readModel.threads).toHaveLength(1);
160 expect(readModel.threads[0]?.title).toBe("Bootstrap Thread");
161
162 await runtime.dispose();
163 });
164
63165 it("returns deterministic read models for repeated reads", async () => {
64166 const createdAt = now();
65167 const system = await createOrchestrationSystem();
......@@ -417,6 +519,7 @@ describe("OrchestrationEngine", () => {
417519
418520 const runtime = ManagedRuntime.make(
419521 OrchestrationEngineLive.pipe(
522 Layer.provide(OrchestrationProjectionSnapshotQueryLive),
420523 Layer.provide(OrchestrationProjectionPipelineLive),
421524 Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)),
422525 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
......@@ -512,6 +615,7 @@ describe("OrchestrationEngine", () => {
512615
513616 const runtime = ManagedRuntime.make(
514617 OrchestrationEngineLive.pipe(
618 Layer.provide(OrchestrationProjectionSnapshotQueryLive),
515619 Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)),
516620 Layer.provide(OrchestrationEventStoreLive),
517621 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
......@@ -653,6 +757,7 @@ describe("OrchestrationEngine", () => {
653757
654758 const runtime = ManagedRuntime.make(
655759 OrchestrationEngineLive.pipe(
760 Layer.provide(OrchestrationProjectionSnapshotQueryLive),
656761 Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)),
657762 Layer.provide(Layer.succeed(OrchestrationEventStore, nonTransactionalStore)),
658763 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
apps/server/src/orchestration/Layers/OrchestrationEngine.ts+3-7
......@@ -19,6 +19,7 @@ import {
1919import { decideOrchestrationCommand } from "../decider.ts";
2020import { createEmptyReadModel, projectEvent } from "../projector.ts";
2121import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts";
22import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";
2223import {
2324 OrchestrationEngineService,
2425 type OrchestrationEngineShape,
......@@ -54,6 +55,7 @@ const makeOrchestrationEngine = Effect.gen(function* () {
5455 const eventStore = yield* OrchestrationEventStore;
5556 const commandReceiptRepository = yield* OrchestrationCommandReceiptRepository;
5657 const projectionPipeline = yield* OrchestrationProjectionPipeline;
58 const projectionSnapshotQuery = yield* ProjectionSnapshotQuery;
5759
5860 let readModel = createEmptyReadModel(new Date().toISOString());
5961
......@@ -195,13 +197,7 @@ const makeOrchestrationEngine = Effect.gen(function* () {
195197 };
196198
197199 yield* projectionPipeline.bootstrap;
198
199 // bootstrap in-memory read model from event store
200 yield* Stream.runForEach(eventStore.readAll(), (event) =>
201 Effect.gen(function* () {
202 readModel = yield* projectEvent(readModel, event);
203 }),
204 );
200 readModel = yield* projectionSnapshotQuery.getSnapshot();
205201
206202 const worker = Effect.forever(Queue.take(commandQueue).pipe(Effect.flatMap(processEnvelope)));
207203 yield* Effect.forkScoped(worker);
apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+2
......@@ -25,6 +25,7 @@ import {
2525 ORCHESTRATION_PROJECTOR_NAMES,
2626 OrchestrationProjectionPipelineLive,
2727} from "./ProjectionPipeline.ts";
28import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
2829import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
2930import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts";
3031import { ServerConfig } from "../../config.ts";
......@@ -1841,6 +1842,7 @@ it.effect("restores pending turn-start metadata across projection pipeline resta
18411842
18421843const engineLayer = it.layer(
18431844 OrchestrationEngineLive.pipe(
1845 Layer.provide(OrchestrationProjectionSnapshotQueryLive),
18441846 Layer.provide(OrchestrationProjectionPipelineLive),
18451847 Layer.provide(OrchestrationEventStoreLive),
18461848 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+2
......@@ -30,6 +30,7 @@ import { GitCore, type GitCoreShape } from "../../git/Services/GitCore.ts";
3030import { TextGeneration, type TextGenerationShape } from "../../git/Services/TextGeneration.ts";
3131import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
3232import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
33import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
3334import { ProviderCommandReactorLive } from "./ProviderCommandReactor.ts";
3435import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
3536import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts";
......@@ -211,6 +212,7 @@ describe("ProviderCommandReactor", () => {
211212 };
212213
213214 const orchestrationLayer = OrchestrationEngineLive.pipe(
215 Layer.provide(OrchestrationProjectionSnapshotQueryLive),
214216 Layer.provide(OrchestrationProjectionPipelineLive),
215217 Layer.provide(OrchestrationEventStoreLive),
216218 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+2
......@@ -31,6 +31,7 @@ import {
3131} from "../../provider/Services/ProviderService.ts";
3232import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
3333import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
34import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
3435import { ProviderRuntimeIngestionLive } from "./ProviderRuntimeIngestion.ts";
3536import {
3637 OrchestrationEngineService,
......@@ -197,6 +198,7 @@ describe("ProviderRuntimeIngestion", () => {
197198 fs.mkdirSync(path.join(workspaceRoot, ".git"));
198199 const provider = createProviderServiceHarness();
199200 const orchestrationLayer = OrchestrationEngineLive.pipe(
201 Layer.provide(OrchestrationProjectionSnapshotQueryLive),
200202 Layer.provide(OrchestrationProjectionPipelineLive),
201203 Layer.provide(OrchestrationEventStoreLive),
202204 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
apps/server/src/server.ts+18-6
......@@ -99,12 +99,24 @@ const ReactorLayerLive = Layer.empty.pipe(
9999 Layer.provideMerge(RuntimeReceiptBusLive),
100100);
101101
102const OrchestrationLayerLive = Layer.empty.pipe(
103 Layer.provideMerge(OrchestrationProjectionSnapshotQueryLive),
104 Layer.provideMerge(OrchestrationEngineLive),
105 Layer.provideMerge(OrchestrationProjectionPipelineLive),
106 Layer.provideMerge(OrchestrationEventStoreLive),
107 Layer.provideMerge(OrchestrationCommandReceiptRepositoryLive),
102const OrchestrationEventInfrastructureLayerLive = Layer.mergeAll(
103 OrchestrationEventStoreLive,
104 OrchestrationCommandReceiptRepositoryLive,
105);
106
107const OrchestrationProjectionPipelineLayerLive = OrchestrationProjectionPipelineLive.pipe(
108 Layer.provide(OrchestrationEventStoreLive),
109);
110
111const OrchestrationInfrastructureLayerLive = Layer.mergeAll(
112 OrchestrationProjectionSnapshotQueryLive,
113 OrchestrationEventInfrastructureLayerLive,
114 OrchestrationProjectionPipelineLayerLive,
115);
116
117const OrchestrationLayerLive = Layer.mergeAll(
118 OrchestrationInfrastructureLayerLive,
119 OrchestrationEngineLive.pipe(Layer.provide(OrchestrationInfrastructureLayerLive)),
108120);
109121
110122const CheckpointingLayerLive = Layer.empty.pipe(