From 04ad17425a80484bf323a7d5076bd477a9499119 Mon Sep 17 00:00:00 2001 From: Dara Adedeji <76637177+SunkenInTime@users.noreply.github.com> Date: Wed, 7 Oct 2026 23:16:52 -0400 Subject: [PATCH] fix(server): opening a long thread no longer blocks the server for a second (#17029) Co-authored-by: Claude Opus 5.5 (1M context) --- .../orchestration-v2/ProjectionStore.test.ts | 44 +++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 24 ++++++---- apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/055_OrchestrationV2.test.ts | 4 +- .../060_ThreadSnapshotWindowIndexes.ts | 23 ++++++++++ .../reconcileV2PreviewMigration.test.ts | 2 + 6 files changed, 90 insertions(+), 9 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/060_ThreadSnapshotWindowIndexes.ts diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index f47b4d6829e86eb6f2ba980fa5819d477d9add07..194d7507f2e6eaa4b1dd9b4ce26deb76ec75d885 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -27,6 +27,7 @@ import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; +import * as Tracer from "effect/Tracer"; import * as SqlClient from "effect/sql/SqlClient"; import { projectThreadAwarenessV2 } from "@t3tools/shared/agentAwareness"; @@ -588,12 +589,26 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }); yield* sql`INSERT INTO orchestration_v2_projection_turn_items ${sql.insert(rows)}`; } + 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; + }, + }); const initial = yield* Effect.acquireUseRelease( Effect.sync(() => vi.spyOn(JSON, "parse")), (parse) => projectionStore .getThreadSnapshotWindow(threadId, { rowLimit: 77, userTurnLimit: 10 }) .pipe( + Effect.withTracer(tracer), Effect.tap(() => Effect.sync(() => { // Tool outputs must not be allocated again just to collect cohort IDs. @@ -606,6 +621,35 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { ), (parse) => Effect.sync(() => parse.mockRestore()), ); + // Large threads blocked the server for over a second here. The boundary + // must come from the user-message index rather than a scan of every tool + // row, and payloads must not go through a sort after they are fetched. + const windowStatement = statements.find((statement) => statement.includes("turn_anchors")); + assert.isDefined(windowStatement); + const windowPlan = yield* sql.unsafe<{ readonly parent: number; readonly detail: string }>( + `EXPLAIN QUERY PLAN ${windowStatement}`, + ); + assert.match( + windowPlan.map((row) => row.detail).join("\n"), + /SEARCH item USING INDEX orchestration_v2_projection_turn_items_user_message_idx \(thread_id=\? AND ordinal<\?\)/, + ); + const topLevel = windowPlan.filter((row) => row.parent === 0).map((row) => row.detail); + assert.include( + topLevel, + "SEARCH item USING INDEX sqlite_autoindex_orchestration_v2_projection_turn_items_1 (turn_item_id=?)", + ); + assert.notInclude(topLevel, "USE TEMP B-TREE FOR ORDER BY"); + const nodeStatement = statements.find((statement) => + statement.includes("WITH RECURSIVE retained"), + ); + assert.isDefined(nodeStatement); + const nodePlan = yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${nodeStatement}`, + ); + assert.include( + nodePlan.map((row) => row.detail), + "SEARCH orchestration_v2_projection_nodes USING INDEX orchestration_v2_projection_nodes_live_idx (thread_id=?)", + ); // Only the selected turn cohort and two lookahead anchors are decoded. assert.lengthOf(initial.projection.turnItems, 12 * 102); const bounded = buildBoundedThreadProjection({ diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 07d847e2b91d76a1fff64130cc204f67c0b39688..4da19774075b7ff90fcd361dfb4aed03d37b5e65 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -2762,7 +2762,10 @@ export const layer: Layer.Layer = END AS ordinal FROM user_anchors ), selected AS ( - SELECT payload_json, ordinal, turn_item_id, run_id, type + -- Choose and sort rows by ID, then fetch payloads in that + -- order. A turn window can hold megabytes of tool output, and + -- carrying it through the union and sort cost about a second. + SELECT ordinal, turn_item_id, run_id, type FROM eligible WHERE ordinal >= (SELECT ordinal FROM boundary) ORDER BY ordinal DESC, turn_item_id DESC @@ -2771,10 +2774,10 @@ export const layer: Layer.Layer = WHEN (SELECT COUNT(*) FROM user_anchors) > 0 THEN -1 ELSE ${window.rowLimit} END - ), retained AS ( - SELECT payload_json, ordinal, turn_item_id FROM selected + ), retained AS MATERIALIZED ( + SELECT ordinal, turn_item_id FROM selected UNION - SELECT request.payload_json, request.ordinal, request.turn_item_id + SELECT request.ordinal, request.turn_item_id FROM orchestration_v2_projection_turn_items AS request WHERE request.run_id IN ( SELECT run_id FROM selected @@ -2782,9 +2785,9 @@ export const layer: Layer.Layer = ) AND request.type = 'run_interrupt_request' UNION - SELECT latest.payload_json, latest.ordinal, latest.turn_item_id + SELECT latest.ordinal, latest.turn_item_id FROM ( - SELECT payload_json, ordinal, turn_item_id + SELECT ordinal, turn_item_id FROM orchestration_v2_projection_turn_items WHERE thread_id = ${threadId} AND ${window.anchorItemId ?? null} IS NULL @@ -2792,10 +2795,15 @@ export const layer: Layer.Layer = ORDER BY ordinal DESC, turn_item_id DESC LIMIT 1 ) AS latest + ORDER BY ordinal ASC, turn_item_id ASC ) - SELECT payload_json + -- CROSS JOIN keeps the sorted IDs as the outer loop, so SQLite + -- skips sorting again once the payloads are attached. + SELECT item.payload_json FROM retained - ORDER BY ordinal ASC, turn_item_id ASC + CROSS JOIN orchestration_v2_projection_turn_items AS item + ON item.turn_item_id = retained.turn_item_id + ORDER BY retained.ordinal ASC, retained.turn_item_id ASC `; // Reuse the decoded items for cohort IDs and the resulting projection. // Parsing these rows separately duplicates every retained tool output. diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 1e1f207a7dac8e0bfc95ed373fe178908082d1d7..35c1a3fdce4bb58ee773a6eac4f4fef1a589a04a 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -73,6 +73,7 @@ import Migration0056 from "./Migrations/056_RemoveRedundantProjectionIndexes.ts" import Migration0057 from "./Migrations/057_ScheduledTaskWebhooks.ts"; import Migration0058 from "./Migrations/058_WebhookRelayDeliveries.ts"; import Migration0059 from "./Migrations/059_McpAppModelContext.ts"; +import Migration0060 from "./Migrations/060_ThreadSnapshotWindowIndexes.ts"; /** * Migration loader with all migrations defined inline. @@ -146,6 +147,7 @@ export const migrationEntries = [ [57, "ScheduledTaskWebhooks", Migration0057], [58, "WebhookRelayDeliveries", Migration0058], [59, "McpAppModelContext", Migration0059], + [60, "ThreadSnapshotWindowIndexes", Migration0060], ] 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 5f2e05f2809f31c9f0d98c0297629b53cfa4a029..5e45ed05d30241e9d53a244a32610351b731875d 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: 59 }, (_, index) => index + 1), + Array.from({ length: 60 }, (_, index) => index + 1), ); }), ); @@ -31,6 +31,7 @@ layer("055_OrchestrationV2", (it) => { [57, "ScheduledTaskWebhooks"], [58, "WebhookRelayDeliveries"], [59, "McpAppModelContext"], + [60, "ThreadSnapshotWindowIndexes"], ]); assert.deepStrictEqual(yield* runMigrations(), []); @@ -56,6 +57,7 @@ layer("055_OrchestrationV2", (it) => { { migration_id: 57, name: "ScheduledTaskWebhooks" }, { migration_id: 58, name: "WebhookRelayDeliveries" }, { migration_id: 59, name: "McpAppModelContext" }, + { migration_id: 60, name: "ThreadSnapshotWindowIndexes" }, ]); const tables = yield* sql<{ readonly name: string }>` diff --git a/apps/server/src/persistence/Migrations/060_ThreadSnapshotWindowIndexes.ts b/apps/server/src/persistence/Migrations/060_ThreadSnapshotWindowIndexes.ts new file mode 100644 index 0000000000000000000000000000000000000000..6977a165a30362bb38a02768a811dc9117f87ec9 --- /dev/null +++ b/apps/server/src/persistence/Migrations/060_ThreadSnapshotWindowIndexes.ts @@ -0,0 +1,23 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // A bounded thread snapshot finds its turn boundary from the newest user + // messages. Without this index it reads every turn item in the thread to + // find a few dozen of them. ProjectionStore's turn_anchors CTE relies on it. + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_user_message_idx + ON orchestration_v2_projection_turn_items(thread_id, ordinal, turn_item_id) + WHERE type = 'user_message' + `; + // The same snapshot keeps every unfinished node. A long thread has thousands + // of finished ones, and reading them all to find a few live ones was more + // than half of the node query's time. + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_nodes_live_idx + ON orchestration_v2_projection_nodes(thread_id) + WHERE status IN ('pending', 'starting', 'running', 'waiting') + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 826dc4831d8db5ac5b4a47480aa077405f72c5f6..675d5f5db0215842459e835ecf0a736a6274e877 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -40,6 +40,7 @@ describe("V2 preview upgrade", () => { [57, "ScheduledTaskWebhooks"], [58, "WebhookRelayDeliveries"], [59, "McpAppModelContext"], + [60, "ThreadSnapshotWindowIndexes"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -122,6 +123,7 @@ describe("V2 preview upgrade", () => { [57, "ScheduledTaskWebhooks"], [58, "WebhookRelayDeliveries"], [59, "McpAppModelContext"], + [60, "ThreadSnapshotWindowIndexes"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); -- 2.54.0