From 11682ffef8cc319d6f041fd394ded6f66280e887 Mon Sep 17 00:00:00 2001 From: Ivan Kosilov Date: Thu, 12 Nov 2026 13:24:47 +0500 Subject: [PATCH] perf(server): events stop maintaining the stream sequence index Every event insert updated idx_orch_events_stream_sequence (aggregate_kind, stream_id, sequence), but each read it served has an index that serves it as well: V2 thread reads use idx_orchestration_events_agent_stream_sequence, which also skips the table row to check the version, and whole-stream reads use idx_orch_events_stream_version, whose entries carry the same rowid. Migration 083 drops it. Without planner statistics, which the server deletes at startup, SQLite would otherwise move these reads to idx_orchestration_events_application_sequence and range over every V2 event, so they now name the thread index: - readAgentEvents for one thread, with or without an event type - getAgentReplayStats - Codex usage history's per-turn selection lookup - ThreadFind's revision, which pinned the dropped index The prune pass finds each stream's newest row as its highest stream_version, which idx_orch_events_stream_version seeks; MAX(sequence) through that index would read the whole stream for every candidate, including the V1 history pruning keeps. Every append takes the stream's highest stream_version plus one, so both pick the same row. Co-Authored-By: Claude Opus 5.5 --- .../ProjectionMaintenance.test.ts | 119 ++++++++++++++++++ .../orchestration-v2/ProjectionMaintenance.ts | 6 +- .../server/src/orchestration-v2/ThreadFind.ts | 2 +- .../server/src/persistence/Migrations.test.ts | 45 ++++++- apps/server/src/persistence/Migrations.ts | 2 + ...opOrchestrationEventStreamSequenceIndex.ts | 10 ++ .../OrchestrationEventStore.sequence.test.ts | 47 +++++++ .../persistence/OrchestrationEventStore.ts | 13 +- .../reconcileV2PreviewMigration.test.ts | 2 + apps/server/src/usage/CodexUsageHistory.ts | 2 + 10 files changed, 239 insertions(+), 9 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/083_DropOrchestrationEventStreamSequenceIndex.ts diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts index 1f42953cc4ec3..09c685e5d4391 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts @@ -1,3 +1,5 @@ +import * as NodeSqlite from "node:sqlite"; + import { assert, it } from "@effect/vitest"; import { EventId, @@ -29,6 +31,13 @@ import * as ProjectionMaintenance from "./ProjectionMaintenance.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts"; +declare module "node:sqlite" { + interface StatementSync { + // Node 26 has this; @types/node 24 does not declare it yet. + stat(counter: "vmStep"): number; + } +} + const database = SqlitePersistence.layerMemory; const eventStoreProvided = EventStore.layer.pipe(Layer.provideMerge(database)); const projectionStoreProvided = ProjectionStore.layer.pipe(Layer.provideMerge(database)); @@ -383,3 +392,113 @@ it.effect("keeps an anchor that a row appended during the pass borrows from", () }), ).pipe(Effect.provide(layerTest)), ); + +it.effect( + "finds each stream's newest event in steps that do not grow with retained V1 history", + () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const sink = yield* EventSink.EventSinkV2; + const events = yield* OrchestrationEventStore.OrchestrationEventStore; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const now = yield* DateTime.now; + const occurredAt = DateTime.formatIso(now); + const projectId = ProjectId.make("project:prune-legacy-history"); + const threadId = ThreadId.make("thread:prune-legacy-history"); + // Pruning keeps V1 rows, so a stream written before V2 keeps its whole V1 history. + const v1History = 1000; + for (const [kind, streamId] of [ + ["project", projectId], + ["thread", threadId], + ] as const) { + yield* sql` + WITH RECURSIVE version(n) AS (SELECT 0 UNION ALL SELECT n + 1 FROM version WHERE n + 1 < ${v1History}) + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, actor_kind, payload_json, metadata_json + ) + SELECT 'event:v1:' || ${kind} || ':' || n, ${kind}, ${streamId}, n, 'legacy.updated', + ${occurredAt}, 'server', '{}', '{}' + FROM version + `; + } + const [v1Head] = yield* sql<{ + readonly sequence: number; + }>`SELECT MAX(sequence) AS sequence FROM orchestration_events`; + const v1Through = v1Head!.sequence; + const updates = 25; + for (let index = 0; index < updates; index++) { + yield* events.appendProjectEvent({ + type: "project.meta-updated", + eventId: EventId.make(`event:prune-legacy-history:project:${index}`), + aggregateKind: "project", + aggregateId: projectId, + occurredAt, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + payload: { projectId, title: `Project ${index}`, updatedAt: occurredAt }, + }); + } + yield* sink.write({ events: [threadCreated(makeThread(threadId, now), now)] }); + for (let revision = 0; revision < updates - 1; revision++) { + yield* sink.write({ events: [itemEvent(threadId, now, itemSpecs[0], revision)] }); + } + yield* maintenance.pruneEventStore(); + yield* TestClock.adjust("1 day"); + + // One copy per planner statistics mode; the server deletes sqlite_stat1 at startup. + using none = new NodeSqlite.DatabaseSync("file:/prune-head-steps-none?vfs=memdb"); + using analyzed = new NodeSqlite.DatabaseSync("file:/prune-head-steps-analyze?vfs=memdb"); + using optimized = new NodeSqlite.DatabaseSync("file:/prune-head-steps-optimize?vfs=memdb"); + const copies = [ + ["none", none, ""], + ["analyze", analyzed, "ANALYZE"], + ["optimize", optimized, "PRAGMA optimize = 0x10002"], + ] as const; + for (const [mode] of copies) + yield* sql`VACUUM INTO ${`file:/prune-head-steps-${mode}?vfs=memdb`}`; + const pages: Array["compile"]>> = []; + const observe: Statement.Transformer = (statement) => { + const compiled = statement.compile(); + if (compiled[0].includes("END AS prunable")) pages.push(compiled); + return Effect.succeed(statement); + }; + // Each stream keeps its newest row, and the thread also keeps thread.created. + assert.equal( + (yield* maintenance + .pruneEventStore() + .pipe(Effect.provideService(Statement.CurrentTransformer, observe))).deletedEventCount, + 2 * updates - 3, + ); + const [query, [, throughSequence, pageSize]] = pages[0]!; + + // VM steps are deterministic; looking up the head by MAX(sequence) through the stream_version + // index read the whole stream for each V2 row: 7.5 times the steps of 100 V1 rows here. + for (const [mode, copy, analyze] of copies) { + if (analyze !== "") copy.exec(analyze); + const page = () => { + const statement = copy.prepare(query); + const rows = statement.all( + v1Through, + throughSequence as NodeSqlite.SQLInputValue, + pageSize as NodeSqlite.SQLInputValue, + ); + return { rows, steps: statement.stat("vmStep") }; + }; + const full = page(); + assert.lengthOf(full.rows, 2 * updates); + assert.deepEqual( + full.rows.flatMap((row, position) => (row.prunable === 0 ? [position] : [])), + [updates - 1, updates, 2 * updates - 1], + ); + copy.exec( + `DELETE FROM orchestration_events WHERE stream_version BETWEEN 100 AND ${v1History - 1}`, + ); + const trimmed = page(); + assert.deepEqual(trimmed.rows, full.rows); + assert.isAtMost(full.steps, trimmed.steps * 1.1, mode); + } + }).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index 210322aed686f..0f3e843f8467d 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts @@ -129,8 +129,10 @@ export const layer: Layer.Layer { assert.notInclude(indexNames, "idx_orch_events_correlation_id"); for (const kept of [ "idx_orch_events_stream_version", - "idx_orch_events_stream_sequence", "idx_orch_events_command_id", "idx_orchestration_events_agent_stream_sequence", ]) { @@ -116,6 +115,50 @@ describe("runMigrations", () => { }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); + it.effect("drops the stream sequence index and keeps the indexes event reads use", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 82 }); + yield* sql` + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at, + command_id, causation_event_id, correlation_id, actor_kind, payload_json, metadata_json, + application_event_version + ) + VALUES ( + 'event-1', 'thread', 'thread-1', 0, 'thread.created', '2026-01-01T00:00:00.000Z', + 'command-1', NULL, 'command-1', 'server', '{}', '{}', 2 + ) + `; + + assert.deepStrictEqual(yield* runMigrations(), [ + [83, "DropOrchestrationEventStreamSequenceIndex"], + ]); + + const indexes = yield* sql<{ readonly name: string }>` + SELECT name FROM sqlite_master + WHERE type = 'index' AND tbl_name = 'orchestration_events' + ORDER BY name + `; + assert.deepStrictEqual( + indexes.map((row) => row.name), + [ + "idx_orch_events_command_id", + "idx_orch_events_stream_version", + "idx_orchestration_events_agent_stream_sequence", + "idx_orchestration_events_application_high_water", + "idx_orchestration_events_application_sequence", + "orchestration_events_v2_created_threads_idx", + "sqlite_autoindex_orchestration_events_1", + ], + ); + const events = yield* sql<{ readonly event_id: string; readonly stream_id: string }>` + SELECT event_id, stream_id FROM orchestration_events + `; + assert.deepStrictEqual(events, [{ event_id: "event-1", stream_id: "thread-1" }]); + }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), + ); + it.effect.each([30, 54])("still starts with a site-local migration at id %s", (id) => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 5e4784e26fc95..af79fd089fa66 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -85,6 +85,7 @@ import Migration0079 from "./Migrations/079_DelegatedRecoveryIndexes.ts"; import Migration0080 from "./Migrations/080_ScheduledTaskThreadMode.ts"; import Migration0081 from "./Migrations/081_ProjectionMessageSearchText.ts"; import Migration0082 from "./Migrations/082_TurnItemConversationIndex.ts"; +import Migration0083 from "./Migrations/083_DropOrchestrationEventStreamSequenceIndex.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -170,6 +171,7 @@ const migrationEntries = [ [80, "ScheduledTaskThreadMode", Migration0080], [81, "ProjectionMessageSearchText", Migration0081], [82, "TurnItemConversationIndex", Migration0082], + [83, "DropOrchestrationEventStreamSequenceIndex", Migration0083], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/083_DropOrchestrationEventStreamSequenceIndex.ts b/apps/server/src/persistence/Migrations/083_DropOrchestrationEventStreamSequenceIndex.ts new file mode 100644 index 0000000000000..22c5bef345987 --- /dev/null +++ b/apps/server/src/persistence/Migrations/083_DropOrchestrationEventStreamSequenceIndex.ts @@ -0,0 +1,10 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // idx_orchestration_events_agent_stream_sequence serves its V2 thread reads and + // idx_orch_events_stream_version its whole-stream reads as well, so it only adds write cost. + yield* sql`DROP INDEX IF EXISTS idx_orch_events_stream_sequence`; +}); diff --git a/apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts b/apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts index c317ab845738b..ed103a8b45d09 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts @@ -7,6 +7,7 @@ import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SqlClient from "effect/sql/SqlClient"; +import * as Statement from "effect/sql/Statement"; import * as OrchestrationEventStore from "./OrchestrationEventStore.ts"; import * as SqlitePersistence from "./Sqlite.ts"; @@ -205,3 +206,49 @@ it.effect( assert.isAbove(liveSequence, highWater); }).pipe(Effect.provide(Layer.fresh(layerEventStore))), ); + +it.effect("reads one thread's events through the thread index, not the V2 sequence range", () => + Effect.gen(function* () { + const store = yield* OrchestrationEventStore.OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const rows = yield* seedEvents( + Arr.makeBy(40, (index) => eventRow(index, "thread", 2, index % 4 === 0 ? "target" : "other")), + ); + const target = rows.filter((row) => row.stream_id === "target").map((row) => row.sequence); + const threadId = ThreadId.make("target"); + const statements: Array["compile"]>> = []; + const observe: Statement.Transformer = (statement) => { + statements.push(statement.compile()); + return Effect.succeed(statement); + }; + const byThread = yield* store + .readAgentEvents({ threadId }) + .pipe(Stream.runCollect, Effect.provideService(Statement.CurrentTransformer, observe)); + const byThreadAndType = yield* store + .readAgentEvents({ threadId, eventType: "provider-session.detached" }) + .pipe(Stream.runCollect, Effect.provideService(Statement.CurrentTransformer, observe)); + const stats = yield* store + .getAgentReplayStats({ threadId, afterSequence: 0, throughSequence: 1_000, maxEvents: 128 }) + .pipe(Effect.provideService(Statement.CurrentTransformer, observe)); + assert.deepEqual( + byThread.map((event) => event.sequence), + target, + ); + assert.deepEqual( + byThreadAndType.map((event) => event.sequence), + target, + ); + assert.equal(stats.eventCount, target.length); + assert.equal(statements.length, 3); + for (const [query, params] of statements) { + const plan = yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${query}`, + params, + ); + assert.include( + plan.map((row) => row.detail), + "SEARCH orchestration_events USING INDEX idx_orchestration_events_agent_stream_sequence (stream_id=? AND sequence>? AND sequence ${input.afterSequence} AND sequence <= ${input.throughSequence ?? Number.MAX_SAFE_INTEGER} @@ -604,7 +606,8 @@ const makeEventStore = Effect.gen(function* () { COALESCE(MAX(event_type = 'thread.created'), 0) AS "hasCreateEvent" FROM ( SELECT ${restoredPayloadJson} AS payload_json, event_type - FROM orchestration_events + -- The planner otherwise ranges over every V2 event after the cursor. + FROM orchestration_events INDEXED BY idx_orchestration_events_agent_stream_sequence WHERE aggregate_kind = 'thread' AND stream_id = ${input.threadId} AND application_event_version = 2 diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 88348b742b470..61a9f2be412f8 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -62,6 +62,7 @@ describe("V2 preview upgrade", () => { [80, "ScheduledTaskThreadMode"], [81, "ProjectionMessageSearchText"], [82, "TurnItemConversationIndex"], + [83, "DropOrchestrationEventStreamSequenceIndex"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -167,6 +168,7 @@ describe("V2 preview upgrade", () => { [80, "ScheduledTaskThreadMode"], [81, "ProjectionMessageSearchText"], [82, "TurnItemConversationIndex"], + [83, "DropOrchestrationEventStreamSequenceIndex"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); diff --git a/apps/server/src/usage/CodexUsageHistory.ts b/apps/server/src/usage/CodexUsageHistory.ts index b032c9b1a0be0..c19d7aad5c2ae 100644 --- a/apps/server/src/usage/CodexUsageHistory.ts +++ b/apps/server/src/usage/CodexUsageHistory.ts @@ -62,7 +62,9 @@ const make = Effect.gen(function* () { json_extract(turn.payload_json, '$.nativeTurnRef.nativeId') AS native_turn_id, ( SELECT json_extract(event.payload_json, '$.modelSelection') + -- The planner otherwise ranges over every V2 event for each turn. FROM orchestration_events AS event + INDEXED BY idx_orchestration_events_agent_stream_sequence WHERE event.aggregate_kind = 'thread' AND event.application_event_version = 2 AND event.stream_id = attempt.thread_id