Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 55 additions & 0 deletions apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,61 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-curs
},
);

it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-transaction-")))(
"OrchestrationProjectionPipeline caller transaction",
(it) => {
it.effect("projects inside the caller's transaction without nesting another", () =>
Effect.gen(function* () {
const projectionPipeline = yield* OrchestrationProjectionPipeline;
const eventStore = yield* OrchestrationEventStore;
const sql = yield* SqlClient.SqlClient;
let transactions = 0;
const tracer = Tracer.make({
span: (options) => {
if (options.name === "sql.transaction") transactions += 1;
return new Tracer.NativeSpan(options);
},
});
const createdAt = "2026-01-01T00:00:00.000Z";
const projectId = ProjectId.make("project-caller-transaction");

const event = yield* eventStore.append({
type: "project.created",
eventId: EventId.make("evt-caller-transaction-project"),
aggregateKind: "project",
aggregateId: projectId,
occurredAt: createdAt,
commandId: CommandId.make("cmd-caller-transaction-project"),
causationEventId: null,
correlationId: null,
metadata: {},
payload: {
projectId,
title: "Caller transaction project",
workspaceRoot: "/tmp/project-caller-transaction",
defaultModelSelection: null,
scripts: [],
createdAt,
updatedAt: createdAt,
},
});

const cleanup = yield* sql
.withTransaction(projectionPipeline.projectEventDeferred(event))
.pipe(Effect.withTracer(tracer));
yield* cleanup;

// A nested transaction is a savepoint SQLite keeps open until the caller commits.
assert.strictEqual(transactions, 1);
assert.deepEqual(
yield* sql`SELECT title FROM projection_projects WHERE project_id = ${projectId}`,
[{ title: "Caller transaction project" }],
);
}),
);
},
);

it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-cleanup-span-")))(
"OrchestrationProjectionPipeline attachment cleanup span",
(it) => {
Expand Down
41 changes: 21 additions & 20 deletions apps/server/src/orchestration/Layers/ProjectionPipeline.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2089,22 +2089,21 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
deletedThreadIds: new Set<string>(),
prunedThreadRelativePaths: new Map<string, Set<string>>(),
};
yield* sql.withTransaction(
Effect.gen(function* () {
yield* Effect.forEach(
projectors,
(projector) => projector.apply(event, attachmentSideEffects),
{ concurrency: 1, discard: true },
);
// Runtime projectors commit together. Bootstrap still advances each cursor separately.
yield* projectionStateRepository.upsertMany(
projectors.map((projector) => ({
projector: projector.name,
lastAppliedSequence: event.sequence,
updatedAt: event.occurredAt,
})),
);
}),
// No transaction here: the caller's makes these writes atomic. A nested one is a
// savepoint that stays open until the caller commits, so SQLite journals every page
// written after it and spills that journal to a temp file on ordinary commands.
yield* Effect.forEach(
projectors,
(projector) => projector.apply(event, attachmentSideEffects),
{ concurrency: 1, discard: true },
);
// Runtime projectors commit together. Bootstrap still advances each cursor separately.
yield* projectionStateRepository.upsertMany(
projectors.map((projector) => ({
projector: projector.name,
lastAppliedSequence: event.sequence,
updatedAt: event.occurredAt,
})),
);
const hasCleanup =
attachmentSideEffects.deletedThreadIds.size > 0 ||
Expand All @@ -2119,15 +2118,17 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
Effect.provideService(FileSystem.FileSystem, fileSystem),
Effect.provideService(Path.Path, path),
Effect.provideService(ServerConfig, serverConfig),
Effect.catchTag("SqlError", (sqlError) =>
Effect.fail(toPersistenceSqlError("ProjectionPipeline.projectEvent:query")(sqlError)),
),
);

const projectEvent: OrchestrationProjectionPipelineShape["projectEvent"] = Effect.fn(
"projectEvent",
)(function* (event) {
const cleanup = yield* projectEventDeferred(event);
const cleanup = yield* sql.withTransaction(projectEventDeferred(event)).pipe(
Effect.catchTags({
SqlError: (sqlError) =>
Effect.fail(toPersistenceSqlError("ProjectionPipeline.projectEvent:query")(sqlError)),
}),
);
yield* cleanup;
});

Expand Down
3 changes: 3 additions & 0 deletions apps/server/src/orchestration/Services/ProjectionPipeline.ts
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,9 @@ export interface OrchestrationProjectionPipelineShape {
/**
* Project an event inside a caller's transaction and return its attachment
* cleanup. Run the returned effect only after the outer transaction commits.
*
* Opens no transaction of its own, so the projectors are atomic only through
* the caller's.
*/
readonly projectEventDeferred: (
event: OrchestrationEvent,
Expand Down
Loading