authorgravatar for 76637177+SunkenInTime@users.noreply.github.comDara Adedeji <76637177+SunkenInTime@users.noreply.github.com> 2026-10-07 23:16:52-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-07 20:16:52-07:00
log04ad17425a80484bf323a7d5076bd477a9499119
tree476435246b7d7c6e4f9147264c50f786550091a8
parent9fba2093e50f208c52d9041c70a793296e311477
signature Signed by PGP key B5690EEEBB952194

fix(server): opening a long thread no longer blocks the server for a second (#17029)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

6 files changed, 90 insertions(+), 9 deletions(-)

apps/server/src/orchestration-v2/ProjectionStore.test.ts+44
...@@ -27,6 +27,7 @@ import * as DateTime from "effect/DateTime";...@@ -27,6 +27,7 @@ import * as DateTime from "effect/DateTime";
27import * as Effect from "effect/Effect";27import * as Effect from "effect/Effect";
28import * as Layer from "effect/Layer";28import * as Layer from "effect/Layer";
29import * as Schema from "effect/Schema";29import * as Schema from "effect/Schema";
30import * as Tracer from "effect/Tracer";
30import * as SqlClient from "effect/sql/SqlClient";31import * as SqlClient from "effect/sql/SqlClient";
31import { projectThreadAwarenessV2 } from "@t3tools/shared/agentAwareness";32import { projectThreadAwarenessV2 } from "@t3tools/shared/agentAwareness";
3233
...@@ -588,12 +589,26 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {...@@ -588,12 +589,26 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
588 });589 });
589 yield* sql`INSERT INTO orchestration_v2_projection_turn_items ${sql.insert(rows)}`;590 yield* sql`INSERT INTO orchestration_v2_projection_turn_items ${sql.insert(rows)}`;
590 }591 }
592 const statements: Array<string> = [];
593 const tracer = Tracer.make({
594 span(options) {
595 const span = new Tracer.NativeSpan(options);
596 const end = span.end.bind(span);
597 span.end = (endTime, exit) => {
598 end(endTime, exit);
599 const query = span.attributes.get("db.query.text");
600 if (typeof query === "string") statements.push(query);
601 };
602 return span;
603 },
604 });
591 const initial = yield* Effect.acquireUseRelease(605 const initial = yield* Effect.acquireUseRelease(
592 Effect.sync(() => vi.spyOn(JSON, "parse")),606 Effect.sync(() => vi.spyOn(JSON, "parse")),
593 (parse) =>607 (parse) =>
594 projectionStore608 projectionStore
595 .getThreadSnapshotWindow(threadId, { rowLimit: 77, userTurnLimit: 10 })609 .getThreadSnapshotWindow(threadId, { rowLimit: 77, userTurnLimit: 10 })
596 .pipe(610 .pipe(
611 Effect.withTracer(tracer),
597 Effect.tap(() =>612 Effect.tap(() =>
598 Effect.sync(() => {613 Effect.sync(() => {
599 // Tool outputs must not be allocated again just to collect cohort IDs.614 // Tool outputs must not be allocated again just to collect cohort IDs.
...@@ -606,6 +621,35 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {...@@ -606,6 +621,35 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
606 ),621 ),
607 (parse) => Effect.sync(() => parse.mockRestore()),622 (parse) => Effect.sync(() => parse.mockRestore()),
608 );623 );
624 // Large threads blocked the server for over a second here. The boundary
625 // must come from the user-message index rather than a scan of every tool
626 // row, and payloads must not go through a sort after they are fetched.
627 const windowStatement = statements.find((statement) => statement.includes("turn_anchors"));
628 assert.isDefined(windowStatement);
629 const windowPlan = yield* sql.unsafe<{ readonly parent: number; readonly detail: string }>(
630 `EXPLAIN QUERY PLAN ${windowStatement}`,
631 );
632 assert.match(
633 windowPlan.map((row) => row.detail).join("\n"),
634 /SEARCH item USING INDEX orchestration_v2_projection_turn_items_user_message_idx \(thread_id=\? AND ordinal<\?\)/,
635 );
636 const topLevel = windowPlan.filter((row) => row.parent === 0).map((row) => row.detail);
637 assert.include(
638 topLevel,
639 "SEARCH item USING INDEX sqlite_autoindex_orchestration_v2_projection_turn_items_1 (turn_item_id=?)",
640 );
641 assert.notInclude(topLevel, "USE TEMP B-TREE FOR ORDER BY");
642 const nodeStatement = statements.find((statement) =>
643 statement.includes("WITH RECURSIVE retained"),
644 );
645 assert.isDefined(nodeStatement);
646 const nodePlan = yield* sql.unsafe<{ readonly detail: string }>(
647 `EXPLAIN QUERY PLAN ${nodeStatement}`,
648 );
649 assert.include(
650 nodePlan.map((row) => row.detail),
651 "SEARCH orchestration_v2_projection_nodes USING INDEX orchestration_v2_projection_nodes_live_idx (thread_id=?)",
652 );
609 // Only the selected turn cohort and two lookahead anchors are decoded.653 // Only the selected turn cohort and two lookahead anchors are decoded.
610 assert.lengthOf(initial.projection.turnItems, 12 * 102);654 assert.lengthOf(initial.projection.turnItems, 12 * 102);
611 const bounded = buildBoundedThreadProjection({655 const bounded = buildBoundedThreadProjection({
apps/server/src/orchestration-v2/ProjectionStore.ts+16-8
...@@ -2762,7 +2762,10 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =...@@ -2762,7 +2762,10 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
2762 END AS ordinal2762 END AS ordinal
2763 FROM user_anchors2763 FROM user_anchors
2764 ), selected AS (2764 ), selected AS (
2765 SELECT payload_json, ordinal, turn_item_id, run_id, type2765 -- Choose and sort rows by ID, then fetch payloads in that
2766 -- order. A turn window can hold megabytes of tool output, and
2767 -- carrying it through the union and sort cost about a second.
2768 SELECT ordinal, turn_item_id, run_id, type
2766 FROM eligible2769 FROM eligible
2767 WHERE ordinal >= (SELECT ordinal FROM boundary)2770 WHERE ordinal >= (SELECT ordinal FROM boundary)
2768 ORDER BY ordinal DESC, turn_item_id DESC2771 ORDER BY ordinal DESC, turn_item_id DESC
...@@ -2771,10 +2774,10 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =...@@ -2771,10 +2774,10 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
2771 WHEN (SELECT COUNT(*) FROM user_anchors) > 0 THEN -12774 WHEN (SELECT COUNT(*) FROM user_anchors) > 0 THEN -1
2772 ELSE ${window.rowLimit}2775 ELSE ${window.rowLimit}
2773 END2776 END
2774 ), retained AS (2777 ), retained AS MATERIALIZED (
2775 SELECT payload_json, ordinal, turn_item_id FROM selected2778 SELECT ordinal, turn_item_id FROM selected
2776 UNION2779 UNION
2777 SELECT request.payload_json, request.ordinal, request.turn_item_id2780 SELECT request.ordinal, request.turn_item_id
2778 FROM orchestration_v2_projection_turn_items AS request2781 FROM orchestration_v2_projection_turn_items AS request
2779 WHERE request.run_id IN (2782 WHERE request.run_id IN (
2780 SELECT run_id FROM selected2783 SELECT run_id FROM selected
...@@ -2782,9 +2785,9 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =...@@ -2782,9 +2785,9 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
2782 )2785 )
2783 AND request.type = 'run_interrupt_request'2786 AND request.type = 'run_interrupt_request'
2784 UNION2787 UNION
2785 SELECT latest.payload_json, latest.ordinal, latest.turn_item_id2788 SELECT latest.ordinal, latest.turn_item_id
2786 FROM (2789 FROM (
2787 SELECT payload_json, ordinal, turn_item_id2790 SELECT ordinal, turn_item_id
2788 FROM orchestration_v2_projection_turn_items2791 FROM orchestration_v2_projection_turn_items
2789 WHERE thread_id = ${threadId}2792 WHERE thread_id = ${threadId}
2790 AND ${window.anchorItemId ?? null} IS NULL2793 AND ${window.anchorItemId ?? null} IS NULL
...@@ -2792,10 +2795,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =...@@ -2792,10 +2795,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
2792 ORDER BY ordinal DESC, turn_item_id DESC2795 ORDER BY ordinal DESC, turn_item_id DESC
2793 LIMIT 12796 LIMIT 1
2794 ) AS latest2797 ) AS latest
2798 ORDER BY ordinal ASC, turn_item_id ASC
2795 )2799 )
2796 SELECT payload_json2800 -- CROSS JOIN keeps the sorted IDs as the outer loop, so SQLite
2801 -- skips sorting again once the payloads are attached.
2802 SELECT item.payload_json
2797 FROM retained2803 FROM retained
2798 ORDER BY ordinal ASC, turn_item_id ASC2804 CROSS JOIN orchestration_v2_projection_turn_items AS item
2805 ON item.turn_item_id = retained.turn_item_id
2806 ORDER BY retained.ordinal ASC, retained.turn_item_id ASC
2799 `;2807 `;
2800 // Reuse the decoded items for cohort IDs and the resulting projection.2808 // Reuse the decoded items for cohort IDs and the resulting projection.
2801 // Parsing these rows separately duplicates every retained tool output.2809 // Parsing these rows separately duplicates every retained tool output.
apps/server/src/persistence/Migrations.ts+2
...@@ -73,6 +73,7 @@ import Migration0056 from "./Migrations/056_RemoveRedundantProjectionIndexes.ts"...@@ -73,6 +73,7 @@ import Migration0056 from "./Migrations/056_RemoveRedundantProjectionIndexes.ts"
73import Migration0057 from "./Migrations/057_ScheduledTaskWebhooks.ts";73import Migration0057 from "./Migrations/057_ScheduledTaskWebhooks.ts";
74import Migration0058 from "./Migrations/058_WebhookRelayDeliveries.ts";74import Migration0058 from "./Migrations/058_WebhookRelayDeliveries.ts";
75import Migration0059 from "./Migrations/059_McpAppModelContext.ts";75import Migration0059 from "./Migrations/059_McpAppModelContext.ts";
76import Migration0060 from "./Migrations/060_ThreadSnapshotWindowIndexes.ts";
7677
77/**78/**
78 * Migration loader with all migrations defined inline.79 * Migration loader with all migrations defined inline.
...@@ -146,6 +147,7 @@ export const migrationEntries = [...@@ -146,6 +147,7 @@ export const migrationEntries = [
146 [57, "ScheduledTaskWebhooks", Migration0057],147 [57, "ScheduledTaskWebhooks", Migration0057],
147 [58, "WebhookRelayDeliveries", Migration0058],148 [58, "WebhookRelayDeliveries", Migration0058],
148 [59, "McpAppModelContext", Migration0059],149 [59, "McpAppModelContext", Migration0059],
150 [60, "ThreadSnapshotWindowIndexes", Migration0060],
149] as const;151] as const;
150152
151export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);153export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);
apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts+3-1
...@@ -13,7 +13,7 @@ layer("055_OrchestrationV2", (it) => {...@@ -13,7 +13,7 @@ layer("055_OrchestrationV2", (it) => {
13 Effect.sync(() => {13 Effect.sync(() => {
14 assert.deepStrictEqual(14 assert.deepStrictEqual(
15 migrationEntries.map(([id]) => id),15 migrationEntries.map(([id]) => id),
16 Array.from({ length: 59 }, (_, index) => index + 1),16 Array.from({ length: 60 }, (_, index) => index + 1),
17 );17 );
18 }),18 }),
19 );19 );
...@@ -31,6 +31,7 @@ layer("055_OrchestrationV2", (it) => {...@@ -31,6 +31,7 @@ layer("055_OrchestrationV2", (it) => {
31 [57, "ScheduledTaskWebhooks"],31 [57, "ScheduledTaskWebhooks"],
32 [58, "WebhookRelayDeliveries"],32 [58, "WebhookRelayDeliveries"],
33 [59, "McpAppModelContext"],33 [59, "McpAppModelContext"],
34 [60, "ThreadSnapshotWindowIndexes"],
34 ]);35 ]);
35 assert.deepStrictEqual(yield* runMigrations(), []);36 assert.deepStrictEqual(yield* runMigrations(), []);
3637
...@@ -56,6 +57,7 @@ layer("055_OrchestrationV2", (it) => {...@@ -56,6 +57,7 @@ layer("055_OrchestrationV2", (it) => {
56 { migration_id: 57, name: "ScheduledTaskWebhooks" },57 { migration_id: 57, name: "ScheduledTaskWebhooks" },
57 { migration_id: 58, name: "WebhookRelayDeliveries" },58 { migration_id: 58, name: "WebhookRelayDeliveries" },
58 { migration_id: 59, name: "McpAppModelContext" },59 { migration_id: 59, name: "McpAppModelContext" },
60 { migration_id: 60, name: "ThreadSnapshotWindowIndexes" },
59 ]);61 ]);
6062
61 const tables = yield* sql<{ readonly name: string }>`63 const tables = yield* sql<{ readonly name: string }>`
apps/server/src/persistence/Migrations/060_ThreadSnapshotWindowIndexes.ts created+23
...@@ -0,0 +1,23 @@
1import * as Effect from "effect/Effect";
2import * as SqlClient from "effect/sql/SqlClient";
3
4export default Effect.gen(function* () {
5 const sql = yield* SqlClient.SqlClient;
6
7 // A bounded thread snapshot finds its turn boundary from the newest user
8 // messages. Without this index it reads every turn item in the thread to
9 // find a few dozen of them. ProjectionStore's turn_anchors CTE relies on it.
10 yield* sql`
11 CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_user_message_idx
12 ON orchestration_v2_projection_turn_items(thread_id, ordinal, turn_item_id)
13 WHERE type = 'user_message'
14 `;
15 // The same snapshot keeps every unfinished node. A long thread has thousands
16 // of finished ones, and reading them all to find a few live ones was more
17 // than half of the node query's time.
18 yield* sql`
19 CREATE INDEX IF NOT EXISTS orchestration_v2_projection_nodes_live_idx
20 ON orchestration_v2_projection_nodes(thread_id)
21 WHERE status IN ('pending', 'starting', 'running', 'waiting')
22 `;
23});
apps/server/src/persistence/reconcileV2PreviewMigration.test.ts+2
...@@ -40,6 +40,7 @@ describe("V2 preview upgrade", () => {...@@ -40,6 +40,7 @@ describe("V2 preview upgrade", () => {
40 [57, "ScheduledTaskWebhooks"],40 [57, "ScheduledTaskWebhooks"],
41 [58, "WebhookRelayDeliveries"],41 [58, "WebhookRelayDeliveries"],
42 [59, "McpAppModelContext"],42 [59, "McpAppModelContext"],
43 [60, "ThreadSnapshotWindowIndexes"],
43 ]);44 ]);
44 assert.deepStrictEqual(yield* runMigrations(), []);45 assert.deepStrictEqual(yield* runMigrations(), []);
45 assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports);46 assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports);
...@@ -122,6 +123,7 @@ describe("V2 preview upgrade", () => {...@@ -122,6 +123,7 @@ describe("V2 preview upgrade", () => {
122 [57, "ScheduledTaskWebhooks"],123 [57, "ScheduledTaskWebhooks"],
123 [58, "WebhookRelayDeliveries"],124 [58, "WebhookRelayDeliveries"],
124 [59, "McpAppModelContext"],125 [59, "McpAppModelContext"],
126 [60, "ThreadSnapshotWindowIndexes"],
125 ]);127 ]);
126 }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))),128 }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))),
127 );129 );