diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 9a0dff477368a..688a1e5b79aac 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -69,6 +69,7 @@ import Migration0063 from "./Migrations/063_DropUnusedV1Projections.ts"; import Migration0064 from "./Migrations/064_CanonicalizeThreadPullRequestLinks.ts"; import Migration0065 from "./Migrations/065_CodexUsageNativeTurnIndex.ts"; import Migration0066 from "./Migrations/066_RepairClientSettlementTimestamps.ts"; +import Migration0067 from "./Migrations/067_OrchestrationSlowQueryIndexes.ts"; export const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -138,6 +139,7 @@ export const migrationEntries = [ [64, "CanonicalizeThreadPullRequestLinks", Migration0064], [65, "CodexUsageNativeTurnIndex", Migration0065], [66, "RepairClientSettlementTimestamps", Migration0066], + [67, "OrchestrationSlowQueryIndexes", Migration0067], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts b/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts index cd926c899d978..c8237cadea8a1 100644 --- a/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts +++ b/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts @@ -15,7 +15,7 @@ layer("055_OrchestrationV2", (it) => { Effect.sync(() => { assert.deepStrictEqual( migrationEntries.map(([id]) => id), - Array.from({ length: 66 }, (_, index) => index + 1), + Array.from({ length: 67 }, (_, index) => index + 1), ); }), ); @@ -40,6 +40,7 @@ layer("055_OrchestrationV2", (it) => { [64, "CanonicalizeThreadPullRequestLinks"], [65, "CodexUsageNativeTurnIndex"], [66, "RepairClientSettlementTimestamps"], + [67, "OrchestrationSlowQueryIndexes"], ]); assert.deepStrictEqual(yield* runMigrations(), []); @@ -72,6 +73,7 @@ layer("055_OrchestrationV2", (it) => { { migration_id: 64, name: "CanonicalizeThreadPullRequestLinks" }, { migration_id: 65, name: "CodexUsageNativeTurnIndex" }, { migration_id: 66, name: "RepairClientSettlementTimestamps" }, + { migration_id: 67, name: "OrchestrationSlowQueryIndexes" }, ]); const tables = yield* sql<{ readonly name: string }>` diff --git a/apps/server/src/persistence/Migrations/067_OrchestrationSlowQueryIndexes.test.ts b/apps/server/src/persistence/Migrations/067_OrchestrationSlowQueryIndexes.test.ts new file mode 100644 index 0000000000000..757273bb70b56 --- /dev/null +++ b/apps/server/src/persistence/Migrations/067_OrchestrationSlowQueryIndexes.test.ts @@ -0,0 +1,109 @@ +import { assert, it } from "@effect/vitest"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +import { runMigrations } from "../Migrations.ts"; + +const layer = it.layer(NodeSqliteClient.layer({ filename: ":memory:" })); + +layer("067_OrchestrationSlowQueryIndexes", (it) => { + it.effect("creates the indexes and the planner uses them for the slow queries", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 66 }); + yield* sql` + INSERT INTO orchestration_v2_projection_turn_items ( + turn_item_id, thread_id, run_id, ordinal, type, status, payload_json, updated_at + ) VALUES ( + 'item-1', 'thread-1', NULL, 1, 'user_message', 'completed', + '{"messageId":"message-1"}', '2026-01-01' + ) + `; + const beforeMessages = yield* sql` + SELECT json_extract(payload_json, '$.messageId') AS message_id, MIN(ordinal) AS ordinal + FROM orchestration_v2_projection_turn_items + WHERE thread_id = 'thread-1' AND type IN ('user_message', 'assistant_message') + GROUP BY json_extract(payload_json, '$.messageId') + `; + const beforeRunning = yield* sql<{ readonly detail: string }>` + EXPLAIN QUERY PLAN + SELECT thread_id FROM orchestration_v2_projection_runs WHERE status = 'running' + `; + assert.isFalse( + beforeRunning.some((row) => row.detail.includes("runs_recovery_idx")), + beforeRunning.map((row) => row.detail).join("\n"), + ); + yield* runMigrations({ toMigrationInclusive: 67 }); + assert.deepEqual(yield* runMigrations(), []); + + const afterRunning = yield* sql<{ readonly detail: string }>` + EXPLAIN QUERY PLAN + SELECT thread_id FROM orchestration_v2_projection_runs WHERE status = 'running' + `; + assert.isTrue( + afterRunning.some((row) => row.detail.includes("runs_running_idx")), + afterRunning.map((row) => row.detail).join("\n"), + ); + const recoveryPlan = yield* sql<{ readonly detail: string }>` + EXPLAIN QUERY PLAN + SELECT thread_id FROM orchestration_v2_projection_runs + WHERE thread_id = 'thread-1' + AND status IN ('queued', 'preparing', 'starting', 'running', 'waiting') + `; + assert.isTrue( + recoveryPlan.some((row) => row.detail.includes("runs_recovery_idx")), + recoveryPlan.map((row) => row.detail).join("\n"), + ); + + const rows = yield* sql<{ readonly name: string }>` + SELECT name FROM sqlite_master WHERE type = 'index' + `; + const indexes = new Set(rows.map((row) => row.name)); + for (const name of [ + "orchestration_v2_projection_turn_items_thread_type_idx", + "orchestration_v2_projection_turn_items_thread_message_idx", + "orchestration_v2_projection_turn_items_run_status_idx", + "orchestration_v2_projection_runs_running_idx", + "orchestration_v2_projection_runs_user_message_idx", + "orchestration_v2_projection_provider_turns_thread_completed_idx", + "idx_orchestration_events_aggregate_version_sequence", + ]) { + assert.isTrue(indexes.has(name), name); + } + + const afterMessages = yield* sql` + SELECT json_extract(payload_json, '$.messageId') AS message_id, MIN(ordinal) AS ordinal + FROM orchestration_v2_projection_turn_items + WHERE thread_id = 'thread-1' AND type IN ('user_message', 'assistant_message') + GROUP BY json_extract(payload_json, '$.messageId') + `; + assert.deepEqual(afterMessages, beforeMessages); + const messagePlan = yield* sql<{ readonly detail: string }>` + EXPLAIN QUERY PLAN + SELECT json_extract(payload_json, '$.messageId'), MIN(ordinal) + FROM orchestration_v2_projection_turn_items + WHERE thread_id = 'thread-1' AND type IN ('user_message', 'assistant_message') + GROUP BY json_extract(payload_json, '$.messageId') + `; + assert.isTrue( + messagePlan.some((row) => + row.detail.includes("orchestration_v2_projection_turn_items_thread_message_idx"), + ), + messagePlan.map((row) => row.detail).join("\n"), + ); + + const runPlan = yield* sql<{ readonly detail: string }>` + EXPLAIN QUERY PLAN + SELECT thread_id FROM orchestration_v2_projection_runs + WHERE json_extract(payload_json, '$.userMessageId') = 'message-1' + `; + assert.isTrue( + runPlan.some((row) => + row.detail.includes("orchestration_v2_projection_runs_user_message_idx"), + ), + runPlan.map((row) => row.detail).join("\n"), + ); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/067_OrchestrationSlowQueryIndexes.ts b/apps/server/src/persistence/Migrations/067_OrchestrationSlowQueryIndexes.ts new file mode 100644 index 0000000000000..dd0a9015e0f2d --- /dev/null +++ b/apps/server/src/persistence/Migrations/067_OrchestrationSlowQueryIndexes.ts @@ -0,0 +1,39 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +// Each expression index repeats the json_extract text of the query it serves. +// SQLite only uses an expression index when the query expression matches it. +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_thread_type_idx + ON orchestration_v2_projection_turn_items(thread_id, type) + `; + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_thread_message_idx + ON orchestration_v2_projection_turn_items(thread_id, json_extract(payload_json, '$.messageId')) + WHERE type IN ('user_message', 'assistant_message') + `; + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_run_status_idx + ON orchestration_v2_projection_turn_items(run_id, status) + `; + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_runs_running_idx + ON orchestration_v2_projection_runs(thread_id) + WHERE status = 'running' + `; + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_runs_user_message_idx + ON orchestration_v2_projection_runs(json_extract(payload_json, '$.userMessageId')) + `; + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_provider_turns_thread_completed_idx + ON orchestration_v2_projection_provider_turns(thread_id, completed_at DESC) + `; + yield* sql` + CREATE INDEX IF NOT EXISTS idx_orchestration_events_aggregate_version_sequence + ON orchestration_events(aggregate_kind, application_event_version, sequence) + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index f1495d1e62a4f..a0943315df785 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -46,6 +46,7 @@ describe("V2 preview upgrade", () => { [64, "CanonicalizeThreadPullRequestLinks"], [65, "CodexUsageNativeTurnIndex"], [66, "RepairClientSettlementTimestamps"], + [67, "OrchestrationSlowQueryIndexes"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -135,6 +136,7 @@ describe("V2 preview upgrade", () => { [64, "CanonicalizeThreadPullRequestLinks"], [65, "CodexUsageNativeTurnIndex"], [66, "RepairClientSettlementTimestamps"], + [67, "OrchestrationSlowQueryIndexes"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), );