authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-04 07:19:37-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-04 07:19:37-07:00
log2263e13fda8c9a4f1b6f4dee32e3c9020195e2aa
treefa10fe35976c1b5f9b4384cb209e5b79dc0736b9
parent27e6cc27fe0f3cff53a44905e40615b7db99c80c
signature Signed by PGP key B5690EEEBB952194

perf(server): batch projector cursor writes (#9671)

4 files changed, 161 insertions(+), 37 deletions(-)

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+115-32
......@@ -20,6 +20,7 @@ import * as Layer from "effect/Layer";
2020import * as Path from "effect/Path";
2121import * as SqlClient from "effect/unstable/sql/SqlClient";
2222
23import { makeSqlStatementCounter } from "../../../integration/SqlStatementCounter.integration.ts";
2324import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";
2425import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";
2526import {
......@@ -27,6 +28,7 @@ import {
2728 SqlitePersistenceMemory,
2829} from "../../persistence/Layers/Sqlite.ts";
2930import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";
31import { ProjectionStateRepository } from "../../persistence/Services/ProjectionState.ts";
3032import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts";
3133import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
3234import {
......@@ -58,6 +60,54 @@ const exists = (filePath: string) =>
5860
5961const BaseTestLayer = makeProjectionPipelinePrefixedTestLayer("t3-projection-pipeline-test-");
6062
63it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-cursor-batch-")))(
64 "OrchestrationProjectionPipeline cursor batches",
65 (it) => {
66 it.effect("writes a project and all projector cursors in two statements", () =>
67 Effect.gen(function* () {
68 const projectionPipeline = yield* OrchestrationProjectionPipeline;
69 const eventStore = yield* OrchestrationEventStore;
70 const projectionState = yield* ProjectionStateRepository;
71 const counter = makeSqlStatementCounter();
72 const createdAt = "2026-01-01T00:00:00.000Z";
73 const event = yield* eventStore.append({
74 type: "project.created",
75 eventId: EventId.make("evt-cursor-batch-project"),
76 aggregateKind: "project",
77 aggregateId: ProjectId.make("project-cursor-batch"),
78 occurredAt: createdAt,
79 commandId: CommandId.make("cmd-cursor-batch-project"),
80 causationEventId: null,
81 correlationId: null,
82 metadata: {},
83 payload: {
84 projectId: ProjectId.make("project-cursor-batch"),
85 title: "Cursor batch project",
86 workspaceRoot: "/tmp/project-cursor-batch",
87 defaultModelSelection: null,
88 scripts: [],
89 createdAt,
90 updatedAt: createdAt,
91 },
92 });
93
94 yield* projectionPipeline.projectEvent(event).pipe(Effect.withTracer(counter.tracer));
95 assert.strictEqual(counter.count(), 2);
96 assert.deepEqual(
97 yield* projectionState.listAll(),
98 Object.values(ORCHESTRATION_PROJECTOR_NAMES)
99 .sort()
100 .map((projector) => ({
101 projector,
102 lastAppliedSequence: event.sequence,
103 updatedAt: createdAt,
104 })),
105 );
106 }),
107 );
108 },
109);
110
61111it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
62112 it.effect("bootstraps all projection states and writes projection rows", () =>
63113 Effect.gen(function* () {
......@@ -813,6 +863,8 @@ it.layer(
813863 },
814864 });
815865
866 const projectionState = yield* ProjectionStateRepository;
867 const cursorsBeforeFailure = yield* projectionState.listAll();
816868 yield* sql`
817869 CREATE TRIGGER fail_thread_messages_projection_state_update
818870 BEFORE UPDATE ON projection_state
......@@ -822,39 +874,39 @@ it.layer(
822874 END;
823875 `;
824876
825 const result = yield* Effect.result(
826 appendAndProject({
827 type: "thread.message-sent",
828 eventId: EventId.make("evt-rollback-3"),
829 aggregateKind: "thread",
830 aggregateId: ThreadId.make("thread-rollback"),
831 occurredAt: now,
832 commandId: CommandId.make("cmd-rollback-3"),
833 causationEventId: null,
834 correlationId: CorrelationId.make("cmd-rollback-3"),
835 metadata: {},
836 payload: {
837 threadId: ThreadId.make("thread-rollback"),
838 messageId: MessageId.make("message-rollback"),
839 role: "user",
840 text: "Rollback me",
841 attachments: [
842 {
843 type: "image",
844 id: "thread-rollback-att-1",
845 name: "rollback.png",
846 mimeType: "image/png",
847 sizeBytes: 5,
848 },
849 ],
850 turnId: null,
851 streaming: false,
852 createdAt: now,
853 updatedAt: now,
854 },
855 }),
856 );
877 const pendingEvent = yield* eventStore.append({
878 type: "thread.message-sent",
879 eventId: EventId.make("evt-rollback-3"),
880 aggregateKind: "thread",
881 aggregateId: ThreadId.make("thread-rollback"),
882 occurredAt: now,
883 commandId: CommandId.make("cmd-rollback-3"),
884 causationEventId: null,
885 correlationId: CorrelationId.make("cmd-rollback-3"),
886 metadata: {},
887 payload: {
888 threadId: ThreadId.make("thread-rollback"),
889 messageId: MessageId.make("message-rollback"),
890 role: "user",
891 text: "Rollback me",
892 attachments: [
893 {
894 type: "image",
895 id: "thread-rollback-att-1",
896 name: "rollback.png",
897 mimeType: "image/png",
898 sizeBytes: 5,
899 },
900 ],
901 turnId: null,
902 streaming: false,
903 createdAt: now,
904 updatedAt: now,
905 },
906 });
907 const result = yield* Effect.result(projectionPipeline.projectEvent(pendingEvent));
857908 assert.equal(result._tag, "Failure");
909 assert.deepEqual(yield* projectionState.listAll(), cursorsBeforeFailure);
858910
859911 const rows = yield* sql<{
860912 readonly count: number;
......@@ -869,6 +921,21 @@ it.layer(
869921 const attachmentPath = path.join(attachmentsDir, "thread-rollback-att-1.png");
870922 assert.isFalse(yield* exists(attachmentPath));
871923 yield* sql`DROP TRIGGER IF EXISTS fail_thread_messages_projection_state_update`;
924
925 yield* projectionPipeline.bootstrap;
926 yield* projectionPipeline.bootstrap;
927 assert.deepEqual(
928 yield* projectionState.listAll(),
929 cursorsBeforeFailure.map((cursor) => ({
930 ...cursor,
931 lastAppliedSequence: pendingEvent.sequence,
932 updatedAt: pendingEvent.occurredAt,
933 })),
934 );
935 const replayedMessages = yield* sql<{ readonly text: string }>`
936 SELECT text FROM projection_thread_messages WHERE message_id = 'message-rollback'
937 `;
938 assert.deepEqual(replayedMessages, [{ text: "Rollback me" }]);
872939 }),
873940 );
874941});
......@@ -3888,6 +3955,14 @@ engineLayer("OrchestrationProjectionPipeline via engine dispatch", (it) => {
38883955
38893956 yield* fileSystem.makeDirectory(attachmentsDir, { recursive: true });
38903957 yield* fileSystem.writeFileString(attachmentPath, "keep this attachment");
3958 const readCursors = sql<{
3959 readonly projector: string;
3960 readonly lastAppliedSequence: number;
3961 }>`
3962 SELECT projector, last_applied_sequence AS "lastAppliedSequence"
3963 FROM projection_state ORDER BY projector
3964 `;
3965 const cursorsBeforeFailure = yield* readCursors;
38913966 yield* sql`
38923967 CREATE TRIGGER fail_attachment_command_receipt
38933968 BEFORE INSERT ON orchestration_command_receipts
......@@ -3899,6 +3974,7 @@ engineLayer("OrchestrationProjectionPipeline via engine dispatch", (it) => {
38993974 const deleteCommand = { type: "thread.delete", commandId, threadId } as const;
39003975 const dispatchError = yield* engine.dispatch(deleteCommand).pipe(Effect.flip);
39013976 assert.equal(dispatchError._tag, "PersistenceSqlError");
3977 assert.deepEqual(yield* readCursors, cursorsBeforeFailure);
39023978 assert.equal(yield* fileSystem.readFileString(attachmentPath), "keep this attachment");
39033979 const rolledBackThreads = yield* sql<{ readonly deletedAt: string | null }>`
39043980 SELECT deleted_at AS "deletedAt" FROM projection_threads WHERE thread_id = ${threadId}
......@@ -3915,6 +3991,13 @@ engineLayer("OrchestrationProjectionPipeline via engine dispatch", (it) => {
39153991 yield* sql`DROP TRIGGER fail_attachment_command_receipt`;
39163992
39173993 const result = yield* engine.dispatch(deleteCommand);
3994 assert.deepEqual(
3995 yield* readCursors,
3996 cursorsBeforeFailure.map((cursor) => ({
3997 ...cursor,
3998 lastAppliedSequence: result.sequence,
3999 })),
4000 );
39184001 assert.isFalse(yield* exists(attachmentPath));
39194002 const committedReceipts = yield* sql<{
39204003 readonly status: string;
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+15-5
......@@ -1947,11 +1947,21 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
19471947 prunedThreadRelativePaths: new Map<string, Set<string>>(),
19481948 };
19491949 yield* sql.withTransaction(
1950 Effect.forEach(
1951 projectors,
1952 (projector) => applyProjectorForEvent(projector, event, attachmentSideEffects),
1953 { concurrency: 1, discard: true },
1954 ),
1950 Effect.gen(function* () {
1951 yield* Effect.forEach(
1952 projectors,
1953 (projector) => projector.apply(event, attachmentSideEffects),
1954 { concurrency: 1, discard: true },
1955 );
1956 // Runtime projectors commit together. Bootstrap still advances each cursor separately.
1957 yield* projectionStateRepository.upsertMany(
1958 projectors.map((projector) => ({
1959 projector: projector.name,
1960 lastAppliedSequence: event.sequence,
1961 updatedAt: event.occurredAt,
1962 })),
1963 );
1964 }),
19551965 );
19561966 // Return the cleanup effect so the caller runs it after the outer transaction commits.
19571967 // @effect-diagnostics-next-line returnEffectInGen:off
apps/server/src/persistence/Layers/ProjectionState.ts+26
......@@ -42,6 +42,26 @@ const makeProjectionStateRepository = Effect.gen(function* () {
4242 `,
4343 });
4444
45 const upsertProjectionStateRows = SqlSchema.void({
46 Request: Schema.Array(ProjectionState),
47 execute: (rows) =>
48 rows.length === 0
49 ? Effect.void
50 : sql`
51 INSERT INTO projection_state ${sql.insert(
52 rows.map((row) => ({
53 projector: row.projector,
54 last_applied_sequence: row.lastAppliedSequence,
55 updated_at: row.updatedAt,
56 })),
57 )}
58 ON CONFLICT (projector)
59 DO UPDATE SET
60 last_applied_sequence = excluded.last_applied_sequence,
61 updated_at = excluded.updated_at
62 `,
63 });
64
4565 const getProjectionStateRow = SqlSchema.findOneOption({
4666 Request: GetProjectionStateInput,
4767 Result: ProjectionState,
......@@ -86,6 +106,11 @@ const makeProjectionStateRepository = Effect.gen(function* () {
86106 Effect.mapError(toPersistenceSqlError("ProjectionStateRepository.upsert:query")),
87107 );
88108
109 const upsertMany: ProjectionStateRepositoryShape["upsertMany"] = (rows) =>
110 upsertProjectionStateRows(rows).pipe(
111 Effect.mapError(toPersistenceSqlError("ProjectionStateRepository.upsertMany:query")),
112 );
113
89114 const getByProjector: ProjectionStateRepositoryShape["getByProjector"] = (input) =>
90115 getProjectionStateRow(input).pipe(
91116 Effect.mapError(toPersistenceSqlError("ProjectionStateRepository.getByProjector:query")),
......@@ -106,6 +131,7 @@ const makeProjectionStateRepository = Effect.gen(function* () {
106131
107132 return {
108133 upsert,
134 upsertMany,
109135 getByProjector,
110136 listAll,
111137 minLastAppliedSequence,
apps/server/src/persistence/Services/ProjectionState.ts+5
......@@ -37,6 +37,11 @@ export interface ProjectionStateRepositoryShape {
3737 */
3838 readonly upsert: (row: ProjectionState) => Effect.Effect<void, ProjectionRepositoryError>;
3939
40 /** Insert or replace projector cursors in one statement. Empty batches do nothing. */
41 readonly upsertMany: (
42 rows: ReadonlyArray<ProjectionState>,
43 ) => Effect.Effect<void, ProjectionRepositoryError>;
44
4045 /**
4146 * Read projection cursor state for a projector key.
4247 */