From 776f1ea2d80801b64279a403b495cf8cd9e1fa87 Mon Sep 17 00:00:00 2001 From: bompus Date: Mon, 19 Oct 2026 01:19:25 -0600 Subject: [PATCH] fix(server): turn starts in long threads no longer stall the server Each turn start reads the thread's runtime recovery state four times. Several of those reads look up unfinished turn items without a type filter, so no partial index applied and SQLite read every item in the thread. On threads with 14,000-17,500 items that cost 75-93 ms per read, so each turn start blocked the event loop for roughly 300-370 ms, freezing every client request, browser stream input included. Add a partial index over unfinished turn items, ordered like the recovery read, and drop a redundant run_id IS NOT NULL filter that steered one subquery onto the full-history run index. The same reads now take 4-7 ms on those threads. --- .../ProjectionRecovery.test.ts | 39 +++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 1 - apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/055_OrchestrationV2.test.ts | 4 +- ...CanonicalizeThreadPullRequestLinks.test.ts | 6 ++- .../065_TurnItemsActiveStatusIndex.ts | 14 +++++++ .../reconcileV2PreviewMigration.test.ts | 2 + 7 files changed, 64 insertions(+), 4 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/065_TurnItemsActiveStatusIndex.ts diff --git a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts index 6e0dd990ce38..a5987300a897 100644 --- a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts @@ -21,6 +21,7 @@ import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as SqlClient from "effect/sql/SqlClient"; +import * as Tracer from "effect/Tracer"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ServerSettings from "../serverSettings.ts"; @@ -369,6 +370,44 @@ it.effect("selects unfinished recovery work without reading settled thread histo }).pipe(Effect.provide(layerTest)), ); +it.effect("looks up a thread's unfinished turn items without reading its finished history", () => + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const threadId = yield* createThread("unfinished-item-lookups"); + yield* createRun(threadId, "running"); + const statements: Array = []; + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options); + const end = span.end.bind(span); + span.end = (endTime, exit) => { + end(endTime, exit); + const query = span.attributes.get("db.query.text"); + if (typeof query === "string") statements.push(query); + }; + return span; + }, + }); + yield* projections.getRuntimeRecoveryProjection(threadId).pipe(Effect.withTracer(tracer)); + + const itemLookups: Array = []; + for (const statement of statements) { + const plan = yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${statement}`, + ); + itemLookups.push( + ...plan.map((row) => row.detail).filter((detail) => /^(?:SEARCH|SCAN) item\b/.test(detail)), + ); + } + // Partial indexes hold only unfinished items, so a long thread's history stays unread. + assert.isNotEmpty(itemLookups); + for (const lookup of itemLookups) { + assert.match(lookup, /turn_items_(?:active|recovery)_idx\b/); + } + }).pipe(Effect.provide(layerTest)), +); + it.effect("recovers terminal subagent results until their cross-thread transfer exists", () => Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 002ef7bafa95..06724628181a 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -4143,7 +4143,6 @@ export const layer: Layer.Layer = WHERE item.thread_id = ${threadId} AND item.type IN ('command_execution', 'dynamic_tool', 'subagent') AND item.status IN ('pending', 'running', 'waiting') - AND item.run_id IS NOT NULL ) ) ORDER BY run.ordinal ASC diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 2b4981e04d1b..c3fb3b4ef67c 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -67,6 +67,7 @@ import Migration0061 from "./Migrations/061_PullRequestReviewGuides.ts"; import Migration0062 from "./Migrations/062_ThreadFailureLookupIndex.ts"; import Migration0063 from "./Migrations/063_DropUnusedV1Projections.ts"; import Migration0064 from "./Migrations/064_CanonicalizeThreadPullRequestLinks.ts"; +import Migration0065 from "./Migrations/065_TurnItemsActiveStatusIndex.ts"; export const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -134,6 +135,7 @@ export const migrationEntries = [ [62, "ThreadFailureLookupIndex", Migration0062], [63, "DropUnusedV1Projections", Migration0063], [64, "CanonicalizeThreadPullRequestLinks", Migration0064], + [65, "TurnItemsActiveStatusIndex", Migration0065], ] 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 447c365082f2..71edb8a2b6f2 100644 --- a/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts +++ b/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts @@ -13,7 +13,7 @@ layer("055_OrchestrationV2", (it) => { Effect.sync(() => { assert.deepStrictEqual( migrationEntries.map(([id]) => id), - Array.from({ length: 64 }, (_, index) => index + 1), + Array.from({ length: 65 }, (_, index) => index + 1), ); }), ); @@ -36,6 +36,7 @@ layer("055_OrchestrationV2", (it) => { [62, "ThreadFailureLookupIndex"], [63, "DropUnusedV1Projections"], [64, "CanonicalizeThreadPullRequestLinks"], + [65, "TurnItemsActiveStatusIndex"], ]); assert.deepStrictEqual(yield* runMigrations(), []); @@ -66,6 +67,7 @@ layer("055_OrchestrationV2", (it) => { { migration_id: 62, name: "ThreadFailureLookupIndex" }, { migration_id: 63, name: "DropUnusedV1Projections" }, { migration_id: 64, name: "CanonicalizeThreadPullRequestLinks" }, + { migration_id: 65, name: "TurnItemsActiveStatusIndex" }, ]); const tables = yield* sql<{ readonly name: string }>` diff --git a/apps/server/src/persistence/Migrations/064_CanonicalizeThreadPullRequestLinks.test.ts b/apps/server/src/persistence/Migrations/064_CanonicalizeThreadPullRequestLinks.test.ts index 706cc02760ea..69eb8eff5702 100644 --- a/apps/server/src/persistence/Migrations/064_CanonicalizeThreadPullRequestLinks.test.ts +++ b/apps/server/src/persistence/Migrations/064_CanonicalizeThreadPullRequestLinks.test.ts @@ -123,8 +123,10 @@ layer("064_CanonicalizeThreadPullRequestLinks", (it) => { 'user', ${v1Payload}, '{}', 1) `; - assert.deepEqual(yield* runMigrations(), [[64, "CanonicalizeThreadPullRequestLinks"]]); - assert.deepEqual(yield* runMigrations(), []); + assert.deepEqual(yield* runMigrations({ toMigrationInclusive: 64 }), [ + [64, "CanonicalizeThreadPullRequestLinks"], + ]); + assert.deepEqual(yield* runMigrations({ toMigrationInclusive: 64 }), []); const rows = yield* sql<{ readonly thread_id: string; readonly payload_json: string }>` SELECT thread_id, payload_json FROM orchestration_v2_projection_threads `; diff --git a/apps/server/src/persistence/Migrations/065_TurnItemsActiveStatusIndex.ts b/apps/server/src/persistence/Migrations/065_TurnItemsActiveStatusIndex.ts new file mode 100644 index 000000000000..1001100e8d79 --- /dev/null +++ b/apps/server/src/persistence/Migrations/065_TurnItemsActiveStatusIndex.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; + + // ProjectionStore's runtime recovery reads look up a thread's unfinished + // items of any type; without it they read the thread's whole history. + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_active_idx + ON orchestration_v2_projection_turn_items(thread_id, ordinal, turn_item_id) + WHERE status IN ('pending', 'running', 'waiting') + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 802c5110490a..d3dbdb35237d 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -44,6 +44,7 @@ describe("V2 preview upgrade", () => { [62, "ThreadFailureLookupIndex"], [63, "DropUnusedV1Projections"], [64, "CanonicalizeThreadPullRequestLinks"], + [65, "TurnItemsActiveStatusIndex"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -131,6 +132,7 @@ describe("V2 preview upgrade", () => { [62, "ThreadFailureLookupIndex"], [63, "DropUnusedV1Projections"], [64, "CanonicalizeThreadPullRequestLinks"], + [65, "TurnItemsActiveStatusIndex"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), );