From 759566fba60c9b0e159574c90d9d40de3b73cff0 Mon Sep 17 00:00:00 2001 From: bompus Date: Mon, 19 Oct 2026 01:19:25 -0600 Subject: [PATCH 1/2] 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 +- .../067_TurnItemsActiveStatusIndex.ts | 14 +++++++ .../reconcileV2PreviewMigration.test.ts | 2 + 6 files changed, 60 insertions(+), 2 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/067_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 747c7c6f395c..5f8d25ba1b2e 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -4353,7 +4353,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 9a0dff477368..a187fb9c5fd7 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_TurnItemsActiveStatusIndex.ts"; export const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -138,6 +139,7 @@ export const migrationEntries = [ [64, "CanonicalizeThreadPullRequestLinks", Migration0064], [65, "CodexUsageNativeTurnIndex", Migration0065], [66, "RepairClientSettlementTimestamps", Migration0066], + [67, "TurnItemsActiveStatusIndex", 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 cd926c899d97..f0a42df9030b 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, "TurnItemsActiveStatusIndex"], ]); 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: "TurnItemsActiveStatusIndex" }, ]); const tables = yield* sql<{ readonly name: string }>` diff --git a/apps/server/src/persistence/Migrations/067_TurnItemsActiveStatusIndex.ts b/apps/server/src/persistence/Migrations/067_TurnItemsActiveStatusIndex.ts new file mode 100644 index 000000000000..1001100e8d79 --- /dev/null +++ b/apps/server/src/persistence/Migrations/067_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 f1495d1e62a4..560e5709e1d8 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, "TurnItemsActiveStatusIndex"], ]); 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, "TurnItemsActiveStatusIndex"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); From 79adb8eb5100cb36a981f2ca7fed2875999691fa Mon Sep 17 00:00:00 2001 From: bompus Date: Wed, 21 Oct 2026 01:35:02 -0600 Subject: [PATCH 2/2] test(server): require all recovery item lookup plans --- apps/server/src/orchestration-v2/ProjectionRecovery.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts index a5987300a897..929015d3330b 100644 --- a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts @@ -401,7 +401,7 @@ it.effect("looks up a thread's unfinished turn items without reading its finishe ); } // Partial indexes hold only unfinished items, so a long thread's history stays unread. - assert.isNotEmpty(itemLookups); + assert.equal(itemLookups.length, 7); for (const lookup of itemLookups) { assert.match(lookup, /turn_items_(?:active|recovery)_idx\b/); }