diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index de54a97ebebf8..75780e5c64b8b 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6834,6 +6834,294 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); + it.effect( + "reads the shell of a thread with a long finished run history without scanning it", + () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const suffix = "shell-long-history"; + const threadId = ThreadId.make(`thread:${suffix}`); + const now = yield* DateTime.now; + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:thread-created`), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make(`project:${suffix}`), + title: "Long finished history", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + for (let ordinal = 1; ordinal <= 3; ordinal++) { + const runId = RunId.make(`run:${suffix}:${ordinal}`); + const rootNodeId = NodeId.make(`node:${suffix}:${ordinal}`); + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:run-${ordinal}`), + type: "run.created", + threadId, + runId, + nodeId: rootNodeId, + driver, + providerInstanceId, + occurredAt: now, + payload: { + id: runId, + threadId, + ordinal, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`message:${suffix}:${ordinal}`), + rootNodeId, + activeAttemptId: null, + status: "completed" as const, + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }); + } + + const { statements, tracer } = traceSqlStatements(); + const shell = yield* projectionStore + .getThreadShell(threadId) + .pipe(Effect.withTracer(tracer)); + assert.isNotNull(shell); + const statement = statements.find((query) => query.includes("AS activity_run_status")); + assert.isDefined(statement); + const plan = yield* sql.unsafe<{ + readonly id: number; + readonly parent: number; + readonly detail: string; + }>(`EXPLAIN QUERY PLAN ${statement}`); + // With no live run, a lookup that walks the ordinal index reads every run the thread has had. + const liveRunLookups = plan.filter( + (row) => + row.detail.startsWith("SEARCH r USING INDEX orchestration_v2_projection_runs_") && + row.detail.includes("(thread_id=?"), + ); + assert.lengthOf(liveRunLookups, 4); + assert.isFalse(plan.some((row) => /^SCAN (?:r|candidate)\b/.test(row.detail))); + for (const lookup of liveRunLookups) { + assert.include( + lookup.detail, + "orchestration_v2_projection_runs_thread_status_idx (thread_id=? AND status=?)", + ); + } + // The newest run needs no sort, but the run that ended last must come from an ordered index. + const candidateLookups = plan.filter((row) => + row.detail.startsWith("SEARCH candidate USING INDEX orchestration_v2_projection_runs_"), + ); + assert.isAbove(candidateLookups.length, 0); + const sortedCandidates = plan.filter( + (row) => + row.detail === "USE TEMP B-TREE FOR ORDER BY" && + candidateLookups.some((lookup) => lookup.parent === row.parent), + ); + assert.deepEqual(sortedCandidates, []); + + // A turn-start guard needs one run, not the thread's whole recovery projection. + const runRead = traceSqlStatements(); + const run = yield* projectionStore + .getRun(threadId, RunId.make(`run:${suffix}:2`)) + .pipe(Effect.withTracer(runRead.tracer)); + assert.equal(run?.ordinal, 2); + assert.isFalse( + runRead.statements.some((query) => /turn_items|projection_nodes|messages/.test(query)), + ); + assert.isUndefined( + yield* projectionStore.getRun(threadId, RunId.make(`run:${suffix}:absent`)), + ); + }), + ); + + it.effect("reads a long thread's records for a command without touching every row", () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const suffix = "command-long-history"; + const threadId = ThreadId.make(`thread:${suffix}`); + const now = yield* DateTime.now; + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:thread-created`), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make(`project:${suffix}`), + title: "Long command history", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + for (const ordinal of [1, 2]) { + const messageId = MessageId.make(`message:${suffix}:${ordinal}`); + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:message-${ordinal}`), + type: "message.updated", + threadId, + occurredAt: now, + payload: { + id: messageId, + threadId, + runId: null, + nodeId: null, + role: "user", + text: `prompt ${ordinal}`, + attachments: [], + streaming: false, + createdBy: "user", + creationSource: "web", + createdAt: now, + updatedAt: now, + }, + }); + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:item-${ordinal}`), + type: "turn-item.updated", + threadId, + occurredAt: now, + payload: { + id: TurnItemId.make(`${suffix}:item:${ordinal}`), + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed", + title: "prompt", + startedAt: now, + completedAt: now, + updatedAt: now, + type: "user_message", + createdBy: "user", + creationSource: "web", + messageId, + inputIntent: "turn_start", + text: `prompt ${ordinal}`, + attachments: [], + }, + }); + } + + const { statements, tracer } = traceSqlStatements(); + yield* projectionStore + .getThreadRecords(threadId, ["runs", "nodes", "providerThreads", "messages", "turnItems"], { + turnItemTypes: ["user_message", "error"], + messageRoles: ["user"], + omitSettledLeafNodes: true, + omitSettledSubagents: true, + }) + .pipe(Effect.withTracer(tracer)); + yield* projectionStore + .getThreadRecords(threadId, ["turnItems"], { + turnItemTypes: ["command_execution"], + turnItemRunId: RunId.make(`run:${suffix}`), + }) + .pipe(Effect.withTracer(tracer)); + yield* projectionStore + .getThreadRecords(threadId, ["turnItems"], { + turnItemTypes: ["run_interrupt_request"], + turnItemRunIds: [RunId.make(`run:${suffix}`)], + }) + .pipe(Effect.withTracer(tracer)); + yield* projectionStore + .getThreadRecords(threadId, ["turnItems"], { + turnItemTypes: ["subagent"], + turnItemNodeIds: [NodeId.make(`node:${suffix}`)], + }) + .pipe(Effect.withTracer(tracer)); + yield* projectionStore.getThreadShell(threadId).pipe(Effect.withTracer(tracer)); + const planOf = Effect.fnUntraced(function* (marker: string) { + const statement = statements.find((query) => query.includes(marker)); + assert.isDefined(statement, marker); + return (yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${statement}`, + )).map((row) => row.detail); + }); + // Each read must reach only the rows it returns; a thread can hold tens of thousands. + const expectations = [ + { + marker: "(type || '') IN (SELECT value FROM json_each", + uses: "turn_items_thread_type_idx (=? AND =?)", + avoids: "turn_items_thread_ordinal_idx", + }, + { + marker: "AND run_id = ?", + uses: "turn_items_thread_run_idx (thread_id=? AND run_id=?)", + avoids: "turn_items_thread_ordinal_idx", + }, + { + marker: "AND run_id IN (SELECT value FROM json_each", + uses: "turn_items_thread_run_idx (thread_id=? AND run_id=?)", + avoids: "turn_items_thread_ordinal_idx", + }, + { + marker: "AND node_id IN (SELECT value FROM json_each", + uses: "turn_items_node_ordinal_idx (node_id=?)", + avoids: "turn_items_thread_ordinal_idx", + }, + { + marker: "MIN(ordinal) AS ordinal", + uses: "turn_items_thread_type_idx (=? AND =?)", + avoids: "turn_items_thread_run_idx", + }, + { + marker: "AS runless_item_count", + uses: "r USING COVERING INDEX orchestration_v2_projection_runs_id_status_idx (=?)", + avoids: "sqlite_autoindex_orchestration_v2_projection_runs_1", + }, + ] as const; + for (const { marker, uses, avoids } of expectations) { + const plan = (yield* planOf(marker)).join("\n"); + assert.include(plan, uses, marker); + assert.notInclude(plan, avoids, marker); + } + }), + ); + it.effect("selects the latest waiting secret only from active runs", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 7c7bbecafe5a0..c17cc01055009 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -618,6 +618,10 @@ export interface ProjectionStoreV2Shape { readonly getRuntimeRecoveryProjection: ( threadId: ThreadId, ) => Effect.Effect; + readonly getRun: ( + threadId: ThreadId, + runId: RunId, + ) => Effect.Effect; readonly getPendingNativeUserInputs: ( threadId: ThreadId, providerTurnId: ProviderTurnId, @@ -3196,6 +3200,15 @@ export const layer: Layer.Layer = readonly work_last_ordinal?: number | null; readonly work_item_count?: number | null; }; + const byRun = + filter?.turnItemRunId !== undefined || + (filter?.turnItemRunIds !== undefined && !filter.turnItemRunIds.includes(null)); + const byNode = !byRun && filter?.turnItemNodeIds !== undefined; + const byTypeOnly = + filter?.turnItemTypes !== undefined && + filter.turnItemRunId === undefined && + filter.turnItemRunIds === undefined && + filter.turnItemNodeIds === undefined; const boundedTurnItemRows: ReadonlyArray = (fields !== undefined && !fields.includes("turnItems")) || (window === undefined && @@ -3211,13 +3224,24 @@ export const layer: Layer.Layer = ? yield* sql` SELECT payload_json FROM orchestration_v2_projection_turn_items - WHERE thread_id = ${threadId} + ${ + // Without statistics SQLite walks the thread's ordinal index and reads every + // item of a long thread; each filter names the index that reaches only its rows. + byRun + ? sql`INDEXED BY orchestration_v2_projection_turn_items_thread_run_idx` + : byNode + ? sql`INDEXED BY orchestration_v2_projection_turn_items_node_ordinal_idx` + : byTypeOnly + ? sql`INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx` + : sql`` + } + WHERE ${byTypeOnly ? sql`(thread_id || '') = ${threadId}` : sql`thread_id = ${threadId}`} ${filter?.turnItemRequestIds === undefined ? sql`` : sql`AND json_extract(payload_json, '$.requestId') IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRequestIds)}))`} - ${filter?.turnItemTypes === undefined ? sql`` : sql`AND type IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))`} + ${filter?.turnItemTypes === undefined ? sql`` : byTypeOnly ? sql`AND (type || '') IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))` : sql`AND type IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))`} ${filter?.turnItemStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemStatuses)}))`} ${filter?.turnItemRunId === undefined ? sql`` : sql`AND run_id = ${filter.turnItemRunId}`} ${filter?.turnItemNodeIds === undefined ? sql`` : sql`AND node_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemNodeIds)}))`} - ${filter?.turnItemRunIds === undefined ? sql`` : sql`AND (run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))})) OR (${filter.turnItemRunIds.includes(null) ? 1 : 0} = 1 AND run_id IS NULL))`} + ${filter?.turnItemRunIds === undefined ? sql`` : filter.turnItemRunIds.includes(null) ? sql`AND (run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))})) OR run_id IS NULL)` : sql`AND run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))}))`} ORDER BY ordinal ASC, turn_item_id ASC ` : yield* sql` @@ -3812,7 +3836,8 @@ export const layer: Layer.Layer = const rows = yield* sql<{ message_id: string; ordinal: number }>` SELECT json_extract(payload_json, '$.messageId') AS message_id, MIN(ordinal) AS ordinal FROM orchestration_v2_projection_turn_items - WHERE thread_id = ${threadId} AND type IN ('user_message', 'assistant_message') + INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx + WHERE (thread_id || '') = ${threadId} AND (type || '') IN ('user_message', 'assistant_message') ${loadedMessageIds === undefined ? sql`` : sql`AND json_extract(payload_json, '$.messageId') IN (SELECT value FROM json_each(${encodeIdList(loadedMessageIds)}))`} GROUP BY json_extract(payload_json, '$.messageId') `; @@ -5560,6 +5585,21 @@ export const layer: Layer.Layer = ) .pipe(Effect.mapError(controlReadError(threadId))); + const getRun: ProjectionStoreV2Shape["getRun"] = (threadId, runId) => + sql + .withTransaction( + Effect.gen(function* () { + yield* requireThread(threadId); + const rows = yield* sql` + SELECT payload_json FROM orchestration_v2_projection_runs + WHERE run_id = ${runId} AND thread_id = ${threadId}`; + return rows[0] === undefined + ? undefined + : yield* decodeRunPayload(rows[0].payload_json); + }), + ) + .pipe(Effect.mapError(controlReadError(threadId))); + const canStartQueuedRun: ProjectionStoreV2Shape["canStartQueuedRun"] = (threadId) => sql .withTransaction( @@ -6304,6 +6344,7 @@ export const layer: Layer.Layer = ( SELECT r.status FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running', 'waiting') AND NOT ${isSilentShellRun("r")} @@ -6318,6 +6359,7 @@ export const layer: Layer.Layer = r.requested_at ) FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running', 'waiting') AND NOT ${isSilentShellRun("r")} @@ -6379,6 +6421,7 @@ export const layer: Layer.Layer = SELECT secret.payload_json -- Keep runs outermost so completed history is never scanned for a pending secret. FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx CROSS JOIN orchestration_v2_projection_turn_items secret INDEXED BY orchestration_v2_projection_turn_items_thread_run_idx WHERE r.thread_id = t.thread_id @@ -6427,6 +6470,7 @@ export const layer: Layer.Layer = LEFT JOIN orchestration_v2_projection_runs active ON active.run_id = ( SELECT r.run_id FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running') AND NOT ${isSilentShellRun("r")} @@ -6516,7 +6560,10 @@ export const layer: Layer.Layer = WHERE thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) GROUP BY thread_id, run_id ) counts - LEFT JOIN orchestration_v2_projection_runs r ON r.run_id = counts.run_id + -- The covering index answers each run from a compact page set instead of the runs table. + LEFT JOIN orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_id_status_idx + ON (r.run_id || '') = (counts.run_id || '') GROUP BY counts.thread_id `; @@ -7586,6 +7633,7 @@ export const layer: Layer.Layer = getCheckpointCaptureContext: readOnReader(getCheckpointCaptureContext), getCheckpointRollbackContext: readOnReader(getCheckpointRollbackContext), getRunMessage: readOnReader(getRunMessage), + getRun: readOnReader(getRun), canStartQueuedRun: readOnReader(canStartQueuedRun), getHeldQueuedRuns: readOnReader(getHeldQueuedRuns), getPendingNativeUserInputs: readOnReader(getPendingNativeUserInputs), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 8b30a961c099d..9ec1954620faf 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -231,6 +231,7 @@ it.effect( getTurnStartContext: () => Effect.die("unused"), getTurnStartHistory: () => Effect.die("unused"), getRuntimeRecoveryProjection: () => Effect.die("unused getRuntimeRecoveryProjection"), + getRun: () => Effect.die("unused getRun"), getPlan: () => Effect.die("unused"), hasUnpairedRunInterruptRequest: () => Effect.die("unused interrupt read"), getThreadAttachmentIds: () => Effect.die("Unused attachment lookup"), diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index d1b5168258874..0859fa15e56cc 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -970,6 +970,11 @@ function makeLocalCommandHarness(input: { (m.text.trim().toLowerCase() !== "/compact" || m.attachments.length > 0), ), }), + getRun: (_threadId, id) => + Effect.as( + failReadIfRunning, + projection.runs.find((candidate) => candidate.id === id), + ), getTurnStartHistory: () => input.sideChat !== undefined ? Effect.succeed([]) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 54b16d7aa2aef..2f5cdd03df2bb 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -238,12 +238,13 @@ export const layer: Layer.Layer< >; }) => { const isCurrentAttemptInStatus = (expectedStatus: OrchestrationV2Run["status"]) => - projectionStore.getRuntimeRecoveryProjection(input.threadId).pipe( - Effect.map((current) => { - const run = current.runs.find((candidate) => candidate.id === input.runId); - return run?.activeAttemptId === input.attemptId && run.status === expectedStatus; - }), - ); + projectionStore + .getRun(input.threadId, input.runId) + .pipe( + Effect.map( + (run) => run?.activeAttemptId === input.attemptId && run.status === expectedStatus, + ), + ); return { isCurrentAttemptInStatus, loadInheritedBackgroundTurnItems: () => @@ -261,15 +262,15 @@ export const layer: Layer.Layer< ), shouldStartProviderTurn: () => isCurrentAttemptInStatus("running"), shouldFinalizeRun: () => - projectionStore.getRuntimeRecoveryProjection(input.threadId).pipe( - Effect.map((current) => { - const run = current.runs.find((candidate) => candidate.id === input.runId); - return ( - run?.activeAttemptId === input.attemptId && - (run.status === "starting" || run.status === "running") - ); - }), - ), + projectionStore + .getRun(input.threadId, input.runId) + .pipe( + Effect.map( + (run) => + run?.activeAttemptId === input.attemptId && + (run.status === "starting" || run.status === "running"), + ), + ), hasUnpairedRunInterruptRequest: () => projectionStore .hasUnpairedRunInterruptRequest( diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 6a9c57e18cc43..0c80c57fae550 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -83,6 +83,8 @@ import Migration0077 from "./Migrations/077_RepairBooleanSettledOverride.ts"; import Migration0078 from "./Migrations/078_DropOrchestrationEventCorrelationIndex.ts"; import Migration0079 from "./Migrations/079_DelegatedRecoveryIndexes.ts"; import Migration0080 from "./Migrations/080_ScheduledTaskThreadMode.ts"; +import Migration0081 from "./Migrations/081_ProjectionRunsExecutedIndex.ts"; +import Migration0082 from "./Migrations/082_CommandReadIndexes.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -166,6 +168,8 @@ const migrationEntries = [ [78, "DropOrchestrationEventCorrelationIndex", Migration0078], [79, "DelegatedRecoveryIndexes", Migration0079], [80, "ScheduledTaskThreadMode", Migration0080], + [81, "ProjectionRunsExecutedIndex", Migration0081], + [82, "CommandReadIndexes", Migration0082], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/081_ProjectionRunsExecutedIndex.ts b/apps/server/src/persistence/Migrations/081_ProjectionRunsExecutedIndex.ts new file mode 100644 index 0000000000000..74838b3f96909 --- /dev/null +++ b/apps/server/src/persistence/Migrations/081_ProjectionRunsExecutedIndex.ts @@ -0,0 +1,14 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Matches the shell's latest-executed-run ordering, so it reads the newest rows instead of sorting every run. + yield* sql` + CREATE INDEX orchestration_v2_projection_runs_executed_idx + ON orchestration_v2_projection_runs( + thread_id, (completed_at IS NULL), completed_at, ordinal, run_id + ) + `; +}); diff --git a/apps/server/src/persistence/Migrations/082_CommandReadIndexes.ts b/apps/server/src/persistence/Migrations/082_CommandReadIndexes.ts new file mode 100644 index 0000000000000..2010069164f8c --- /dev/null +++ b/apps/server/src/persistence/Migrations/082_CommandReadIndexes.ts @@ -0,0 +1,18 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Each lets a read of one long thread skip the rows it does not return. Every column a query + // could match is wrapped in `|| ''`, so only a lookup that writes the same expression can use + // them. Plain indexes were chosen for unrelated statements over their partial indexes. + yield* sql` + CREATE INDEX orchestration_v2_projection_turn_items_thread_type_idx + ON orchestration_v2_projection_turn_items((thread_id || ''), (type || ''), ordinal, turn_item_id) + `; + yield* sql` + CREATE INDEX orchestration_v2_projection_runs_id_status_idx + ON orchestration_v2_projection_runs((run_id || ''), status) + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 84fa4ddb44c77..39cb418baf47a 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -60,6 +60,8 @@ describe("V2 preview upgrade", () => { [78, "DropOrchestrationEventCorrelationIndex"], [79, "DelegatedRecoveryIndexes"], [80, "ScheduledTaskThreadMode"], + [81, "ProjectionRunsExecutedIndex"], + [82, "CommandReadIndexes"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -163,6 +165,8 @@ describe("V2 preview upgrade", () => { [78, "DropOrchestrationEventCorrelationIndex"], [79, "DelegatedRecoveryIndexes"], [80, "ScheduledTaskThreadMode"], + [81, "ProjectionRunsExecutedIndex"], + [82, "CommandReadIndexes"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), );