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 { `; const indexNames = indexes.map((row) => row.name); assert.notInclude(indexNames, "idx_orch_events_correlation_id"); + assert.notInclude(indexNames, "idx_orch_events_stream_sequence"); 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", ]) { diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index f1b8f6c9d806e..5882a43998462 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -99,6 +99,7 @@ import Migration0088 from "./Migrations/088_TurnItemTypeStatusIndex.ts"; import Migration0089 from "./Migrations/089_ShellThreadReadIndexes.ts"; import Migration0090 from "./Migrations/090_AuthConnectDevices.ts"; import Migration0091 from "./Migrations/091_AuthConnectDeniedRequests.ts"; +import Migration0092 from "./Migrations/092_DropOrchestrationEventStreamSequenceIndex.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -193,6 +194,7 @@ const migrationEntries = [ [89, "ShellThreadReadIndexes", Migration0089], [90, "AuthConnectDevices", Migration0090], [91, "AuthConnectDeniedRequests", Migration0091], + [92, "DropOrchestrationEventStreamSequenceIndex", Migration0092], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/092_DropOrchestrationEventStreamSequenceIndex.ts b/apps/server/src/persistence/Migrations/092_DropOrchestrationEventStreamSequenceIndex.ts new file mode 100644 index 0000000000000..22c5bef345987 --- /dev/null +++ b/apps/server/src/persistence/Migrations/092_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 89c06944d2bf4..2a1e761eb2505 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -71,6 +71,7 @@ describe("V2 preview upgrade", () => { [89, "ShellThreadReadIndexes"], [90, "AuthConnectDevices"], [91, "AuthConnectDeniedRequests"], + [92, "DropOrchestrationEventStreamSequenceIndex"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -185,6 +186,7 @@ describe("V2 preview upgrade", () => { [89, "ShellThreadReadIndexes"], [90, "AuthConnectDevices"], [91, "AuthConnectDeniedRequests"], + [92, "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