Skip to content

perf(server): prune events the projections already cover #29806

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
juliusmarminge merged 5 commits into from
Nov 3, 2026
+1,144 −1,204
Merged

perf(server): prune events the projections already cover #29806

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
05efb69
perf(server): prune events the projections already cover
juliusmarminge Nov 3, 2026
0955847
fix(server): idle threads resume without a snapshot; keep anchors of …
juliusmarminge Nov 3, 2026
e188f19
fix(server): recheck the pruned boundary after reading a resume replay
juliusmarminge Nov 3, 2026
0b5b885
test(server): require idle-thread resume to succeed, not merely skip …
juliusmarminge Nov 3, 2026
e2ecefd
docs(server): qualify which command receipts survive pruning
juliusmarminge Nov 3, 2026
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
144 changes: 31 additions & 113 deletions apps/server/src/orchestration-v2/FoundationPersistence.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@ import * as EventSink from "./EventSink.ts";
import * as EventStore from "./EventStore.ts";
import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
import * as ProjectionMaintenance from "./ProjectionMaintenance.ts";
import * as ProjectionReplay from "./testkit/ProjectionReplay.ts";
import * as ProjectionStore from "./ProjectionStore.ts";
import * as ProjectStore from "./ProjectStore.ts";
import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts";
Expand All @@ -85,6 +86,9 @@ const layerCommandReceiptStoreProvided = CommandReceiptStore.layer.pipe(
);
const layerIsolatedOutbox = Layer.fresh(EffectOutbox.layer.pipe(Layer.provideMerge(layerDatabase)));
const layerProjectionMaintenanceProvided = ProjectionMaintenance.layer.pipe(
Layer.provide(layerDatabase),
);
const layerProjectionReplayProvided = ProjectionReplay.layer.pipe(
Layer.provide(layerStoresProvided),
);
const layerTest = Layer.mergeAll(
Expand All @@ -94,6 +98,7 @@ const layerTest = Layer.mergeAll(
layerCommandReceiptStoreProvided,
IdAllocator.layer,
layerProjectionMaintenanceProvided,
layerProjectionReplayProvided,
);

const providerInstanceId = ProviderInstanceId.make("codex");
Expand Down Expand Up @@ -148,85 +153,6 @@ function threadCreatedEvent(input: {
};
}

it.effect("rebuilds event history one bounded page at a time", () =>
Effect.gen(function* () {
const eventStore = yield* EventStore.EventStoreV2;
const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
const now = yield* DateTime.now;
const threadId = ThreadId.make("thread:foundation-paged-rebuild");
const thread = makeThread(threadId, now);
const eventCount = 1_005;
yield* eventStore.append({
events: [
threadCreatedEvent({ id: "event:paged-rebuild:0", thread, now }),
...Array.from({ length: eventCount - 1 }, (_, index) => ({
id: EventId.make(`event:paged-rebuild:${index + 1}`),
type: "thread.metadata-updated" as const,
threadId,
occurredAt: now,
payload: { ...thread, title: `Rebuilt update ${index + 1}` },
})),
],
});
let applied = 0;
let failAfterFirstPage = false;
const appliedAtRead: Array<number> = [];
const layerObservedStores = Layer.mergeAll(
Layer.succeed(EventStore.EventStoreV2, {
...eventStore,
read: (input) =>
Stream.suspend(() => {
appliedAtRead.push(applied);
if (failAfterFirstPage && applied >= 500) {
return Stream.fail(
new EventStore.EventStoreReadEventsError({ afterSequence: input?.afterSequence }),
);
}
return eventStore.read(input);
}),
}),
Layer.succeed(ProjectionStore.ProjectionStoreV2, {
...projectionStore,
apply: (event) =>
projectionStore.apply(event).pipe(
Effect.tap(() =>
Effect.sync(() => {
applied += 1;
}),
),
),
}),
);
const rebuild = Effect.gen(function* () {
const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
return yield* maintenance.rebuild;
}).pipe(
Effect.provide(
Layer.fresh(ProjectionMaintenance.layer.pipe(Layer.provide(layerObservedStores))),
),
);
const rebuilt = yield* rebuild;
assert.isTrue(rebuilt.valid);
assert.equal(applied, eventCount);
assert.deepEqual(appliedAtRead, [0, 500, 1_000]);
assert.equal(
(yield* projectionStore.getThreadProjection(threadId)).thread.title,
"Rebuilt update 1004",
);
applied = 0;
failAfterFirstPage = true;
const failed = yield* Effect.exit(rebuild);
assert.equal(failed._tag, "Failure");
assert.equal(applied, 500);
assert.equal(
(yield* projectionStore.getThreadProjection(threadId)).thread.title,
"Rebuilt update 1004",
);
const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
assert.isTrue((yield* maintenance.verify).valid);
}).pipe(Effect.provide(layerTest)),
);

const seedCompaction = Effect.fnUntraced(function* () {
const sql = yield* SqlClient.SqlClient;
yield* sql`INSERT INTO orchestration_v2_legacy_imports (
Expand All @@ -240,6 +166,13 @@ const seedCompaction = Effect.fnUntraced(function* () {
) SELECT 'event-' || n, 'thread', 'thread', n,
CASE WHEN n IN (1, 51, 101) THEN 'message.updated' ELSE 'thread.visited' END,
'2026-10-17', 'server', '{"id":"message"}', '{}', 2 FROM history`;
// A pass a day ago marked the whole log; projections cover it.
yield* sql`INSERT INTO orchestration_v2_projection_metadata (
projection_name, schema_version, last_sequence, updated_at
) VALUES
('event-log-prune-mark', 2, 151, '1969-12-30T00:00:00.000Z'),
('thread-projections', 2, 151, '1969-12-30T00:00:00.000Z')
ON CONFLICT(projection_name) DO UPDATE SET last_sequence = excluded.last_sequence`;
yield* sql`WITH RECURSIVE history(n) AS (
SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 100
) INSERT INTO orchestration_command_receipts (
Expand Down Expand Up @@ -284,19 +217,19 @@ it.effect("discovers on the reader worker and deletes on the writer", () =>
}),
);
const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
const result = yield* maintenance.compactEventStore();
assert.equal(result.deletedEventCount, 148);
const result = yield* maintenance.pruneEventStore();
assert.equal(result.deletedEventCount, 150);
assert.equal(result.deletedReceiptCount, 100);
// Only the in-transaction check for rows appended during the pass reads on the writer.
assert.deepEqual(
reads.filter((query) => /\bSELECT\b/.test(query)),
reads.filter((query) => /\bSELECT\b/.test(query) && !query.includes("AS prompt_sequence")),
[],
);
assert(writes.some((query) => query.includes("DELETE FROM orchestration_events")));
assert(writes.some((query) => query.includes("DELETE FROM orchestration_command_receipts")));
}).pipe(
Effect.provide(
ProjectionMaintenance.layer.pipe(
Layer.provide(Layer.merge(EventStore.layer, ProjectionStore.layer)),
Layer.provideMerge(SqlitePersistence.layerFromPath(filename)),
),
),
Expand All @@ -318,7 +251,7 @@ it.effect("paces scheduled compaction while preserving retention state and fixed
const tracer = Tracer.make({
span(options) {
const span = new Tracer.NativeSpan(options);
if (span.name === "ProjectionMaintenance.compactEventStore") compactionSpans.push(span);
if (span.name === "ProjectionMaintenance.pruneEventStore") compactionSpans.push(span);
const end = span.end.bind(span);
span.end = (time, exit) => {
end(time, exit);
Expand Down Expand Up @@ -377,15 +310,13 @@ it.effect("paces scheduled compaction while preserving retention state and fixed
},
}),
);
assert.deepEqual(pauses, Array(5).fill(270));
assert.deepEqual(pauses, Array(8).fill(270));
assert.deepEqual(
compactionSpans.map((span) => span.status._tag),
["Ended"],
);
// 151 stopped being the stream head when 152 committed mid-pass; 152 is past the bound.
assert.deepEqual(yield* sql`SELECT sequence FROM orchestration_events ORDER BY sequence`, [
{ sequence: 1 },
{ sequence: 101 },
{ sequence: 151 },
{ sequence: 152 },
]);
assert.deepEqual(yield* sql`SELECT command_id FROM orchestration_command_receipts`, [
Expand Down Expand Up @@ -768,7 +699,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
Effect.gen(function* () {
const eventSink = yield* EventSink.EventSinkV2;
const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
const maintenance = yield* ProjectionReplay.ProjectionReplay;
const now = yield* DateTime.now;
const firstThreadId = ThreadId.make("thread:foundation-shared-session:first");
const secondThreadId = ThreadId.make("thread:foundation-shared-session:second");
Expand Down Expand Up @@ -829,7 +760,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
}),
);

it.effect("compacts superseded state events, imported v1 events, and legacy receipts", () =>
it.effect("prunes covered history, imported v1 events, and legacy receipts", () =>
Effect.gen(function* () {
const eventSink = yield* EventSink.EventSinkV2;
const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
Expand Down Expand Up @@ -983,9 +914,12 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
ON CONFLICT(command_id) DO NOTHING
`;

const summary = yield* maintenance.compactEventStore();
// The first pass only marks the head and drops imported legacy rows.
const marked = yield* maintenance.pruneEventStore();
yield* TestClock.adjust("1 day");
const summary = yield* maintenance.pruneEventStore();
assert.isAtLeast(summary.deletedEventCount, 504);
assert.isAtLeast(summary.deletedReceiptCount, 1);
assert.equal(marked.deletedReceiptCount + summary.deletedReceiptCount, 1);

const remaining = yield* sql<{ readonly event_id: string }>`
SELECT event_id
Expand All @@ -996,18 +930,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
`;
assert.deepEqual(
remaining.map((row) => row.event_id),
[
"event:foundation-compact:create",
"event:foundation-compact:meta",
"event:foundation-compact:message-1",
"event:foundation-compact:node-1",
"event:foundation-compact:item-1",
"event:foundation-compact:visit-2",
"event:foundation-compact:message-2",
"event:foundation-compact:node-2",
"event:foundation-compact:item-2",
"event:foundation-compact:tags-2",
],
["event:foundation-compact:create", "event:foundation-compact:tags-2"],
);

const remainingV1 = yield* sql<{ readonly stream_id: string }>`
Expand All @@ -1031,8 +954,6 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
["command:foundation-compact:pending"],
);

assert.isTrue((yield* maintenance.verify).valid);
assert.isTrue((yield* maintenance.rebuild).valid);
assert.deepEqual(yield* projections.getThreadProjection(threadId), beforeCompaction);
}),
);
Expand All @@ -1041,7 +962,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
Effect.gen(function* () {
const eventSink = yield* EventSink.EventSinkV2;
const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
const maintenance = yield* ProjectionReplay.ProjectionReplay;
const sql = yield* SqlClient.SqlClient;
const now = yield* DateTime.now;
const parentThreadId = ThreadId.make("thread:foundation-cross-thread:parent");
Expand Down Expand Up @@ -3152,7 +3073,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
Effect.gen(function* () {
const eventSink = yield* EventSink.EventSinkV2;
const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
const maintenance = yield* ProjectionReplay.ProjectionReplay;
const sql = yield* SqlClient.SqlClient;
const now = yield* DateTime.now;
const threadId = ThreadId.make("thread:foundation-many-items");
Expand Down Expand Up @@ -3384,10 +3305,7 @@ it.effect.each(["events", "receipts"] as const)(
const path = yield* Path.Path;
const directory = yield* fs.makeTempDirectoryScoped({ prefix: "compaction-reader-" });
const database = SqlitePersistence.layerFromPath(path.join(directory, "statev2.sqlite"));
const maintenanceLayer = ProjectionMaintenance.layer.pipe(
Layer.provide(Layer.merge(EventStore.layer, ProjectionStore.layer)),
Layer.provideMerge(database),
);
const maintenanceLayer = ProjectionMaintenance.layer.pipe(Layer.provideMerge(database));
yield* Effect.gen(function* () {
const sql = yield* SqlClient.SqlClient;
const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
Expand Down Expand Up @@ -3434,7 +3352,7 @@ it.effect.each(["events", "receipts"] as const)(
: Effect.succeed(statement);
};
const compacted = yield* Deferred.make<void>();
const compaction = yield* maintenance.compactEventStore().pipe(
const compaction = yield* maintenance.pruneEventStore().pipe(
Effect.provideService(Statement.CurrentTransformer, observeDiscovery),
Effect.provideService(Statement.SpanPropagationEnabled, true),
Effect.tap(() => Deferred.succeed(compacted, undefined)),
Expand Down
Loading
Loading