From 384e41a0aad7df33b4e074bc0eaa07bfeea871f5 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 02:15:59 -0800 Subject: [PATCH] perf(server): sending to a very long thread reads about 5x less of its history Every message.dispatch reads its thread before it plans. On a thread with 59,000 turn items that read fetched every finished tool call and node only to drop it: 417 MB read to return 22,000 rows. - Typed turn-item reads use the conversation index of migration 082. - Settled leaf nodes are skipped through a new expression index. - Settled subagents are dropped inside their existing index. - Commands load only pending context transfers, through (thread, status). - The message ordering lookup does not run when the loaded items give it. Builds on #29521 and #31851. Refs #29468. Co-Authored-By: Claude Fable 5.2 --- .../Orchestrator.control-reads.test.ts | 20 +++ .../src/orchestration-v2/Orchestrator.ts | 2 + .../orchestration-v2/ProjectionStore.test.ts | 41 ++++++ .../src/orchestration-v2/ProjectionStore.ts | 138 +++++++++++++----- apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/083_CommandReadIndexes.ts | 27 ++++ .../reconcileV2PreviewMigration.test.ts | 2 + 7 files changed, 195 insertions(+), 37 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/083_CommandReadIndexes.ts diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index 6cb8356889e3b..117a7898a601f 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -827,6 +827,23 @@ it.effect( yield* sql`INSERT INTO orchestration_v2_projection_context_handoffs (context_handoff_id, thread_id, target_run_id, to_provider_thread_id, strategy, status, updated_at, payload_json) VALUES ('obsolete-handoff', ${threadId}, 'old-run', 'old-provider-thread', 'full_thread_summary', 'ready', ${DateTime.formatIso(now)}, '{"obsolete":true}')`; + for (const status of [ + "resolved_native", + "resolved_portable", + "failed", + "consumed", + "superseded", + ]) { + yield* sql`INSERT INTO orchestration_v2_projection_context_transfers + (context_transfer_id, source_thread_id, target_thread_id, target_run_id, + type, status, updated_at, payload_json) + VALUES (${`obsolete-transfer:${status}`}, 'old-source', ${threadId}, NULL, + 'merge_back', ${status}, ${DateTime.formatIso(now)}, '{"obsolete":true}')`; + } + assert.equal( + (yield* Effect.exit(projections.getThreadRecords(threadId, ["contextTransfers"])))._tag, + "Failure", + ); yield* orchestrator.dispatch({ type: "message.dispatch", commandId: CommandId.make("dispatch-with-old-history"), @@ -842,6 +859,9 @@ it.effect( turnItemTypes: ["user_message"], }); assert.isAbove(fresh.turnItems.at(-1)!.ordinal, 900); + // Deletion reads transfer history separately to find side chats. + yield* sql`DELETE FROM orchestration_v2_projection_context_transfers + WHERE target_thread_id = ${threadId}`; yield* orchestrator.dispatch({ type: "thread.archive", commandId: CommandId.make("archive-with-old-history"), diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 6a585abc5e5d5..b39e92d71417a 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1489,6 +1489,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // Commands here read only undelivered or active tasks; a dispatcher thread keeps // thousands of delivered ones, each carrying its full prompt. omitSettledSubagents: true, + // Commands act on pending forks and merge-backs only; consumed ones stay unread. + contextTransferStatuses: ["pending"], }, ) .pipe( diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 06fc3ca868ae1..580cdc24cac46 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -741,6 +741,37 @@ it.effect("idle compaction filters keep the latest run, live work and the newest .join("\n"); assert.include(plan, `USING INDEX sqlite_autoindex_orchestration_v2_projection_${table}_1`); } + // The filters every command reads with seek the rows they keep, not every row of the thread. + const command = traceSqlStatements(); + yield* store + .getThreadRecords(threadId, ["nodes", "subagents", "turnItems", "contextTransfers"], { + turnItemTypes: ["user_message", "error"], + omitSettledLeafNodes: true, + omitSettledSubagents: true, + contextTransferStatuses: ["pending"], + }) + .pipe(Effect.withTracer(command.tracer)); + for (const [table, seeks] of [ + ["turn_items", ["USING INDEX orchestration_v2_projection_turn_items_conversation_idx"]], + ["nodes", ["nodes_settled_leaf_idx (thread_id=? AND =?)"]], + ["subagents", ["USING COVERING INDEX orchestration_v2_projection_subagents_settled_idx"]], + [ + "context_transfers", + [ + "source_thread_status_idx (source_thread_id=? AND status=?)", + "target_thread_idx (target_thread_id=? AND status=?)", + ], + ], + ] as const) { + const query = command.statements.find((statement) => + statement.includes(`FROM orchestration_v2_projection_${table}`), + ); + assert.isDefined(query); + const plan = (yield* sql.unsafe<{ readonly detail: string }>(`EXPLAIN QUERY PLAN ${query}`)) + .map((row) => row.detail) + .join("\n"); + for (const seek of seeks) assert.include(plan, seek); + } assert.deepEqual( records.providerTurns.map((turn) => turn.id), ["turn:idle-newest"], @@ -828,6 +859,16 @@ it.effect("record reads filter notification sources without dropping matching me plan.some((detail) => detail.includes("turn_items_user_message_idx")), plan.join("\n"), ); + // With every user item loaded, the items give the order and the lookup does not run. + const withItems = traceSqlStatements(); + const ordered = yield* store + .getThreadRecords(threadId, ["messages", "turnItems"], { + messageRoles: ["user"], + turnItemTypes: ["user_message"], + }) + .pipe(Effect.withTracer(withItems.tracer)); + assert.deepEqual(ordered.messages, userMessages.messages); + assert.isFalse(withItems.statements.some((statement) => statement.includes("MIN(ordinal)"))); assert.lengthOf((yield* store.getThreadRecords(threadId, ["messages"])).messages, 4); assert.lengthOf( (yield* store.getThreadRecords(threadId, ["messages"], { diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 797427ede42ee..2a0fdc7518d10 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1116,6 +1116,15 @@ const encodeIdList = Schema.encodeSync(Schema.fromJsonString(Schema.Array(Schema const matchesNothing = (...filters: ReadonlyArray | undefined>) => filters.some((values) => values?.length === 0); +// A typed item read that names no tool type fits in migration 082's conversation index. A run or +// node filter is left alone: it already reads through a narrower index. +const readsConversationItemsOnly = (filter: ProjectionRecordFilter | undefined) => + filter?.turnItemTypes !== undefined && + filter.turnItemRunId === undefined && + filter.turnItemRunIds === undefined && + filter.turnItemNodeIds === undefined && + filter.turnItemTypes.every((type) => DEFERRED_WORK_ITEM_TYPES.every((tool) => tool !== type)); + const encodeThreadPayload = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2AppThreadJsonSchema), ); @@ -2072,6 +2081,17 @@ export const layer: Layer.Layer = `(${DEFERRED_WORK_ITEM_TYPES.map((type) => `'${type}'`).join(", ")})`, ); + // An ended task whose completion was acknowledged, delivered or disposed, outside an active run. + const settledSubagent = (row: Statement.Fragment) => sql`( + ${row}.status IN ('completed', 'failed', 'cancelled', 'interrupted') + AND COALESCE(${row}.completion_delivery_state, '') IN ('acknowledged', 'delivered', 'disposed') + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_runs AS run + WHERE run.run_id = ${row}.run_id + AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') + ) + )`; + // A todo list is never closed when its run ends, so a long thread holds an active one per // old run. Counting those as live work drags each run's nodes and subagents into every window. const todoListOfSettledRun = (row: Statement.Fragment) => sql`( @@ -3429,7 +3449,14 @@ export const layer: Layer.Layer = ? yield* sql` SELECT payload_json FROM orchestration_v2_projection_turn_items - WHERE thread_id = ${threadId} + ${ + // A type list of conversation items reads them from the conversation index. + // Any other plan fetches every tool call of the thread to check its type. + readsConversationItemsOnly(filter) + ? sql`INDEXED BY orchestration_v2_projection_turn_items_conversation_idx + WHERE thread_id = ${threadId} AND type NOT IN ${CONVERSATION_INDEX_EXCLUDED_TYPES}` + : sql`WHERE 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?.turnItemStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemStatuses)}))`} @@ -3694,22 +3721,31 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_nodes AS node - WHERE ${threadColumn(filter?.nodeIds, "node.thread_id")} = ${threadId} - ${filter?.nodeIds === undefined ? sql`` : sql`AND node.node_id IN (SELECT value FROM json_each(${encodeIdList(filter.nodeIds)}))`} - ${ - filter?.omitSettledLeafNodes === true - ? sql`AND NOT ( - node.kind IN ('tool_call', 'assistant_message', 'reasoning') - AND node.status IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back') - AND node.run_id IS NOT NULL - AND NOT EXISTS ( - SELECT 1 FROM orchestration_v2_projection_runs AS run - WHERE run.run_id = node.run_id - AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') + WHERE ${ + filter?.omitSettledLeafNodes === true + ? // Seeks the kept nodes. One filter over the thread reads every settled leaf to + // drop it. The first predicate must match migration 083's index expression. A + // node's run is always a run of its own thread. + sql`node.rowid IN ( + SELECT rowid FROM orchestration_v2_projection_nodes + WHERE thread_id = ${threadId} + AND (kind IN ('tool_call', 'assistant_message', 'reasoning') + AND status IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back') + AND run_id IS NOT NULL) = 0 + UNION ALL + SELECT rowid FROM orchestration_v2_projection_nodes + WHERE thread_id = ${threadId} + AND run_id IN ( + SELECT run_id FROM orchestration_v2_projection_runs + WHERE thread_id = ${threadId} + AND status IN ('preparing', 'queued', 'starting', 'running', 'waiting') ) - )` - : sql`` - } + AND kind IN ('tool_call', 'assistant_message', 'reasoning') + AND status IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back') + )` + : sql`${threadColumn(filter?.nodeIds, "node.thread_id")} = ${threadId}` + } + ${filter?.nodeIds === undefined ? sql`` : sql`AND node.node_id IN (SELECT value FROM json_each(${encodeIdList(filter.nodeIds)}))`} ORDER BY COALESCE(node.started_at, ''), node.node_id ASC ` : sql` @@ -3737,22 +3773,22 @@ export const layer: Layer.Layer = ${filter?.subagentIds === undefined ? sql`` : sql`AND subagent_id IN (SELECT value FROM json_each(${encodeIdList(filter.subagentIds)}))`} ${filter?.subagentRunIds === undefined ? sql`` : sql`AND run_id IN (SELECT value FROM json_each(${encodeIdList(filter.subagentRunIds)}))`} ${ - filter?.omitSettledSubagents !== undefined - ? sql`AND (${ - filter.omitSettledSubagents === true - ? sql`0` - : sql`subagent_id IN (SELECT value FROM json_each(${encodeIdList(filter.omitSettledSubagents.except)}))` - } OR NOT ( - subagent.status IN ('completed', 'failed', 'cancelled', 'interrupted') - AND COALESCE(subagent.completion_delivery_state, '') - IN ('acknowledged', 'delivered', 'disposed') - AND NOT EXISTS ( - SELECT 1 FROM orchestration_v2_projection_runs AS run - WHERE run.run_id = subagent.run_id - AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') - ) - ))` - : sql`` + filter?.omitSettledSubagents === true + ? // The settled index holds every column this check reads, so the settled + // tasks are dropped there and their rows, each with a full prompt, stay unread. + sql`AND subagent.rowid IN ( + SELECT candidate.rowid + FROM orchestration_v2_projection_subagents AS candidate + INDEXED BY orchestration_v2_projection_subagents_settled_idx + WHERE candidate.thread_id = ${threadId} + AND NOT ${settledSubagent(sql`candidate`)} + )` + : filter?.omitSettledSubagents !== undefined + ? sql`AND ( + subagent_id IN (SELECT value FROM json_each(${encodeIdList(filter.omitSettledSubagents.except)})) + OR NOT ${settledSubagent(sql`subagent`)} + )` + : sql`` } ORDER BY COALESCE(started_at, ''), subagent_id ASC ` @@ -3984,8 +4020,21 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_context_transfers - WHERE (source_thread_id = ${threadId} OR target_thread_id = ${threadId}) - ${filter?.contextTransferStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.contextTransferStatuses)}))`} + WHERE ${ + filter?.contextTransferStatuses === undefined + ? sql`(source_thread_id = ${threadId} OR target_thread_id = ${threadId})` + : // Each side seeks its (thread, status) index. One OR over both sides checks the + // status on every row the thread ever sent or received. + sql`rowid IN ( + SELECT rowid FROM orchestration_v2_projection_context_transfers + WHERE source_thread_id = ${threadId} + AND status IN (SELECT value FROM json_each(${encodeIdList(filter.contextTransferStatuses)})) + UNION + SELECT rowid FROM orchestration_v2_projection_context_transfers + WHERE target_thread_id = ${threadId} + AND status IN (SELECT value FROM json_each(${encodeIdList(filter.contextTransferStatuses)})) + )` + } ORDER BY rowid ASC ` : sql` @@ -4031,7 +4080,24 @@ export const layer: Layer.Layer = decodeRows(decodeContextTransferPayload, threadId)(contextTransferRows), ]); const messageOrdinals = new Map(); - if (fields !== undefined && fields.includes("messages") && messages.length > 1) { + // The literal `type = 'user_message'` lets SQLite pick the partial user-message index. + const userOnly = messages.every((message) => message.role === "user"); + // The loaded items already order the messages when they hold every item the lookup reads. + const loadsEveryItemOf = (type: OrchestrationV2TurnItem["type"]) => + window === undefined && + (fields === undefined || fields.includes("turnItems")) && + filter?.turnItemRequestIds === undefined && + filter?.turnItemStatuses === undefined && + filter?.turnItemRunId === undefined && + filter?.turnItemRunIds === undefined && + filter?.turnItemNodeIds === undefined && + (filter?.turnItemTypes === undefined || filter.turnItemTypes.includes(type)); + if ( + fields !== undefined && + fields.includes("messages") && + messages.length > 1 && + !(loadsEveryItemOf("user_message") && (userOnly || loadsEveryItemOf("assistant_message"))) + ) { // Broad role reads keep the original query to avoid per-ID lookup overhead. const loadedMessageIds = filter?.messageIds === undefined && @@ -4040,8 +4106,6 @@ export const layer: Layer.Layer = (filter.messageRoles.includes("user") && filter.messageRoles.includes("assistant"))) ? undefined : messages.map((message) => message.id); - // The literal `type = 'user_message'` lets SQLite pick the partial user-message index. - const userOnly = messages.every((message) => message.role === "user"); 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 diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 5e4784e26fc95..422d2122d03bd 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -85,6 +85,7 @@ import Migration0079 from "./Migrations/079_DelegatedRecoveryIndexes.ts"; import Migration0080 from "./Migrations/080_ScheduledTaskThreadMode.ts"; import Migration0081 from "./Migrations/081_ProjectionMessageSearchText.ts"; import Migration0082 from "./Migrations/082_TurnItemConversationIndex.ts"; +import Migration0083 from "./Migrations/083_CommandReadIndexes.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -170,6 +171,7 @@ const migrationEntries = [ [80, "ScheduledTaskThreadMode", Migration0080], [81, "ProjectionMessageSearchText", Migration0081], [82, "TurnItemConversationIndex", Migration0082], + [83, "CommandReadIndexes", Migration0083], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/083_CommandReadIndexes.ts b/apps/server/src/persistence/Migrations/083_CommandReadIndexes.ts new file mode 100644 index 0000000000000..864b16305ffe4 --- /dev/null +++ b/apps/server/src/persistence/Migrations/083_CommandReadIndexes.ts @@ -0,0 +1,27 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Every command reads its thread's nodes without the settled leaves. This index lets that read + // seek the nodes it keeps instead of fetching each finished tool call to drop it. The + // omitSettledLeafNodes read in ProjectionStore must spell this expression the same way. + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_nodes_settled_leaf_idx + ON orchestration_v2_projection_nodes( + thread_id, + (kind IN ('tool_call', 'assistant_message', 'reasoning') + AND status IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back') + AND run_id IS NOT NULL) + ) + `; + + // A dispatcher thread is the source of thousands of consumed transfers. With status in the + // index, a read of the pending ones no longer fetches each consumed row to check it. + yield* sql`DROP INDEX IF EXISTS orchestration_v2_projection_context_transfers_source_thread_idx`; + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_context_transfers_source_thread_status_idx + ON orchestration_v2_projection_context_transfers(source_thread_id, status) + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 88348b742b470..05f09f8e16850 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -62,6 +62,7 @@ describe("V2 preview upgrade", () => { [80, "ScheduledTaskThreadMode"], [81, "ProjectionMessageSearchText"], [82, "TurnItemConversationIndex"], + [83, "CommandReadIndexes"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -167,6 +168,7 @@ describe("V2 preview upgrade", () => { [80, "ScheduledTaskThreadMode"], [81, "ProjectionMessageSearchText"], [82, "TurnItemConversationIndex"], + [83, "CommandReadIndexes"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), );