From e67f00e670a0aecc3007d32efdd1138354472761 Mon Sep 17 00:00:00 2001 From: Bogdan Date: Fri, 6 Nov 2026 05:28:12 +0000 Subject: [PATCH 1/4] perf(server): shell reads no longer walk every run of an idle thread The thread shell query finds a thread's active run four times. SQLite served each lookup by walking the ordinal index newest-first and reading every run's status, so a thread with no active run had its whole run history read on every shell read. Read those lookups through the existing (thread_id, status) index instead. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 18 +++++++++++++----- .../src/orchestration-v2/ProjectionStore.ts | 6 ++++++ 2 files changed, 19 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 580cdc24cac46..a7cad92f35269 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -5221,7 +5221,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); - it.effect("looks up shell stop requests through the interrupt request index", () => + it.effect("looks up shell stop requests and active runs through their indexes", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; const sql = yield* SqlClient.SqlClient; @@ -5235,15 +5235,23 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { statement.includes("AS latest_run_stop_requested"), ); assert.isDefined(threadQuery); - const stopRequestLookups = (yield* sql.unsafe<{ readonly detail: string }>( + const plan = (yield* sql.unsafe<{ readonly detail: string }>( `EXPLAIN QUERY PLAN ${threadQuery}`, - )) - .map((row) => row.detail) - .filter((detail) => detail.startsWith("SEARCH stop_request ")); + )).map((row) => row.detail); + const stopRequestLookups = plan.filter((detail) => detail.startsWith("SEARCH stop_request ")); assert.lengthOf(stopRequestLookups, 2); for (const lookup of stopRequestLookups) { assert.include(lookup, "orchestration_v2_projection_turn_items_stop_request_idx"); } + // The active, activity, and pending-secret lookups; walking the ordinal index instead + // reads every run of an idle thread. + const activeRunLookups = plan.filter( + (detail) => detail.startsWith("SEARCH r ") && detail.includes("(thread_id=?"), + ); + assert.lengthOf(activeRunLookups, 4); + for (const lookup of activeRunLookups) { + assert.include(lookup, "orchestration_v2_projection_runs_thread_status_idx"); + } }), ); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 2a0fdc7518d10..adedeaf9b5956 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6684,7 +6684,10 @@ export const layer: Layer.Layer = active.status AS active_run_status, ( SELECT r.status + -- Read active runs by status. Walking the ordinal index to skip this sort + -- reads every run of a thread with none active. FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running', 'waiting') AND NOT ${isSilentShellRun("r")} @@ -6699,6 +6702,7 @@ export const layer: Layer.Layer = r.requested_at ) FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running', 'waiting') AND NOT ${isSilentShellRun("r")} @@ -6761,6 +6765,7 @@ export const layer: Layer.Layer = SELECT secret.payload_json -- Keep runs outermost so completed history is never scanned for a pending secret. FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx CROSS JOIN orchestration_v2_projection_turn_items secret INDEXED BY orchestration_v2_projection_turn_items_thread_run_idx WHERE r.thread_id = t.thread_id @@ -6809,6 +6814,7 @@ export const layer: Layer.Layer = LEFT JOIN orchestration_v2_projection_runs active ON active.run_id = ( SELECT r.run_id FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running') AND NOT ${isSilentShellRun("r")} From 4a3f35fd5ca5bbbbddc4a1bad8511a297b09376e Mon Sep 17 00:00:00 2001 From: Michael Zhou Date: Thu, 12 Nov 2026 12:32:06 +0800 Subject: [PATCH 2/4] perf(server): the shell read no longer sorts every run of every thread The thread-shell list finds each thread's last executed run with a correlated subquery ordered by completion time. No index carried that order, so SQLite sorted the thread's whole run history for every thread in the list. On a 200-thread fixture with one million runs the shell query took 214 ms; with the index it takes 2 ms and returns the same rows. Migration 83 adds a partial index on the runs table in the order that lookup reads, with the lookup's own predicate, so the shell query walks the index instead of sorting. The lookup keeps its semantics: same runs, same order. Refs pingdotgg/t3code#23000 Refs pingdotgg/t3code#9661 Co-Authored-By: Claude Code --- .../orchestration-v2/ProjectionStore.test.ts | 26 +++++++++++++++++++ apps/server/src/persistence/Migrations.ts | 2 ++ .../Migrations/084_LastExecutedRunIndex.ts | 18 +++++++++++++ .../reconcileV2PreviewMigration.test.ts | 2 ++ 4 files changed, 48 insertions(+) create mode 100644 apps/server/src/persistence/Migrations/084_LastExecutedRunIndex.ts diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index a7cad92f35269..15ebc75e9c99a 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -5255,6 +5255,32 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); + it.effect("reads a thread's last executed run through the last executed run index", () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const { projectId } = yield* assertProjectScopedShells("sql-shell-last-executed-index"); + const { statements, tracer } = traceSqlStatements(); + yield* projectionStore + .getShellSnapshot({ projectId, location: "active" }) + .pipe(Effect.withTracer(tracer)); + + const threadQuery = statements.find((statement) => + statement.includes("AS latest_run_stop_requested"), + ); + assert.isDefined(threadQuery); + const plan = (yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${threadQuery}`, + )).map((row) => row.detail); + const lastExecutedIndex = plan.findIndex((detail) => + detail.includes("orchestration_v2_projection_runs_last_executed_idx"), + ); + assert.notStrictEqual(lastExecutedIndex, -1); + // The lookup that follows the search is the sort it no longer needs. + assert.notInclude(plan[lastExecutedIndex + 1] ?? "", "USE TEMP B-TREE"); + }), + ); + it.effect("reads a run's turn-start items by thread and run, not by walking thread history", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 422d2122d03bd..4de9edc9a1a30 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -86,6 +86,7 @@ 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"; +import Migration0084 from "./Migrations/084_LastExecutedRunIndex.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -172,6 +173,7 @@ const migrationEntries = [ [81, "ProjectionMessageSearchText", Migration0081], [82, "TurnItemConversationIndex", Migration0082], [83, "CommandReadIndexes", Migration0083], + [84, "LastExecutedRunIndex", Migration0084], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/084_LastExecutedRunIndex.ts b/apps/server/src/persistence/Migrations/084_LastExecutedRunIndex.ts new file mode 100644 index 0000000000000..2c36d62634873 --- /dev/null +++ b/apps/server/src/persistence/Migrations/084_LastExecutedRunIndex.ts @@ -0,0 +1,18 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // The shell's last executed run lookup sorts every run of the thread. This index carries + // that order and the same predicate the lookup filters on. A database that already has + // the index keeps it, so a hand-applied one does not fail the migration. + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_runs_last_executed_idx + ON orchestration_v2_projection_runs( + thread_id, (completed_at IS NULL) DESC, completed_at DESC, ordinal DESC, run_id DESC + ) + WHERE status <> 'queued' + AND NOT (status = 'cancelled' AND json_extract(payload_json, '$.startedAt') IS NULL) + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 05f09f8e16850..dcb876e18b9bf 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -63,6 +63,7 @@ describe("V2 preview upgrade", () => { [81, "ProjectionMessageSearchText"], [82, "TurnItemConversationIndex"], [83, "CommandReadIndexes"], + [84, "LastExecutedRunIndex"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -169,6 +170,7 @@ describe("V2 preview upgrade", () => { [81, "ProjectionMessageSearchText"], [82, "TurnItemConversationIndex"], [83, "CommandReadIndexes"], + [84, "LastExecutedRunIndex"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); From f00e0fe968e7c3ef8665195d2898ce7540e4b839 Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Sat, 31 Oct 2026 04:16:24 +0100 Subject: [PATCH 3/4] perf(server): shell reads stop counting every turn item Shell snapshots and getThreadShell counted each thread's items by walking the (thread_id, run_id) index, one entry per item. Migration 085 keeps one count per (thread, run) with triggers on turn items, and the shell count queries read it. (cherry picked from commit f8b758dc495 of #27975; the migration moved from 076 to 085) --- .../src/orchestration-v2/ProjectionStore.ts | 22 +++---- apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/085_TurnItemCounts.test.ts | 60 +++++++++++++++++++ .../Migrations/085_TurnItemCounts.ts | 60 +++++++++++++++++++ .../reconcileV2PreviewMigration.test.ts | 2 + docs/internals/performance-regressions.md | 6 +- 6 files changed, 135 insertions(+), 17 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/085_TurnItemCounts.test.ts create mode 100644 apps/server/src/persistence/Migrations/085_TurnItemCounts.ts diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index adedeaf9b5956..5f65eb1846139 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6886,27 +6886,21 @@ export const layer: Layer.Layer = const selectShellRunItemCounts = (threadIds: ReadonlyArray) => sql` - SELECT thread_id, run_id, COUNT(*) AS item_count - FROM orchestration_v2_projection_turn_items - WHERE run_id IS NOT NULL AND thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) - GROUP BY thread_id, run_id + SELECT thread_id, run_key AS run_id, item_count + FROM orchestration_v2_projection_turn_item_counts + WHERE run_key <> '' AND thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) `; - // Count all selected threads in one index walk; fork sources still need per-run counts. const selectShellItemCounts = (threadIds: ReadonlyArray) => sql` SELECT counts.thread_id, - SUM(CASE WHEN counts.run_id IS NULL OR r.status <> 'rolled_back' + SUM(CASE WHEN counts.run_key = '' OR r.status <> 'rolled_back' THEN counts.item_count ELSE 0 END) AS item_count, - SUM(CASE WHEN counts.run_id IS NULL THEN counts.item_count ELSE 0 END) + SUM(CASE WHEN counts.run_key = '' THEN counts.item_count ELSE 0 END) AS runless_item_count - FROM ( - SELECT thread_id, run_id, COUNT(*) AS item_count - FROM orchestration_v2_projection_turn_items - WHERE thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) - GROUP BY thread_id, run_id - ) counts - LEFT JOIN orchestration_v2_projection_runs r ON r.run_id = counts.run_id + FROM orchestration_v2_projection_turn_item_counts counts + LEFT JOIN orchestration_v2_projection_runs r ON r.run_id = counts.run_key + WHERE counts.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) GROUP BY counts.thread_id `; diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 4de9edc9a1a30..37b74d5c6db01 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -87,6 +87,7 @@ import Migration0081 from "./Migrations/081_ProjectionMessageSearchText.ts"; import Migration0082 from "./Migrations/082_TurnItemConversationIndex.ts"; import Migration0083 from "./Migrations/083_CommandReadIndexes.ts"; import Migration0084 from "./Migrations/084_LastExecutedRunIndex.ts"; +import Migration0085 from "./Migrations/085_TurnItemCounts.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -174,6 +175,7 @@ const migrationEntries = [ [82, "TurnItemConversationIndex", Migration0082], [83, "CommandReadIndexes", Migration0083], [84, "LastExecutedRunIndex", Migration0084], + [85, "TurnItemCounts", Migration0085], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/085_TurnItemCounts.test.ts b/apps/server/src/persistence/Migrations/085_TurnItemCounts.test.ts new file mode 100644 index 0000000000000..cb73f5a8fd8d0 --- /dev/null +++ b/apps/server/src/persistence/Migrations/085_TurnItemCounts.test.ts @@ -0,0 +1,60 @@ +import { assert, it } from "@effect/vitest"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +import { runMigrations } from "../Migrations.ts"; + +const layer = it.layer(NodeSqliteClient.layer({ filename: ":memory:" })); + +layer("085_TurnItemCounts", (it) => { + it.effect("backfills item counts, then follows upserts, moves, and deletes", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 84 }); + const upsert = (id: string, threadId: string, runId: string | null) => sql` + INSERT INTO orchestration_v2_projection_turn_items ( + turn_item_id, thread_id, run_id, ordinal, type, status, updated_at, payload_json + ) + VALUES ( + ${id}, ${threadId}, ${runId}, 1, 'assistant_message', 'completed', + '2026-10-29T00:00:00.000Z', '{}' + ) + ON CONFLICT(turn_item_id) DO UPDATE SET + thread_id = excluded.thread_id, + run_id = excluded.run_id, + status = excluded.status + `; + yield* upsert("runless-1", "thread-1", null); + yield* upsert("runless-2", "thread-1", null); + yield* upsert("run-1", "thread-1", "run-1"); + + yield* runMigrations(); + + const counts = () => + sql<{ readonly thread_id: string; readonly run_key: string; readonly item_count: number }>` + SELECT thread_id, run_key, item_count + FROM orchestration_v2_projection_turn_item_counts + ORDER BY thread_id, run_key + `; + assert.deepEqual(yield* counts(), [ + { thread_id: "thread-1", run_key: "", item_count: 2 }, + { thread_id: "thread-1", run_key: "run-1", item_count: 1 }, + ]); + + yield* upsert("run-1", "thread-1", "run-1"); + yield* upsert("run-2", "thread-2", "run-2"); + yield* upsert("runless-2", "thread-1", "run-1"); + assert.deepEqual(yield* counts(), [ + { thread_id: "thread-1", run_key: "", item_count: 1 }, + { thread_id: "thread-1", run_key: "run-1", item_count: 2 }, + { thread_id: "thread-2", run_key: "run-2", item_count: 1 }, + ]); + + yield* sql`DELETE FROM orchestration_v2_projection_turn_items WHERE thread_id = 'thread-1'`; + assert.deepEqual(yield* counts(), [ + { thread_id: "thread-2", run_key: "run-2", item_count: 1 }, + ]); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/085_TurnItemCounts.ts b/apps/server/src/persistence/Migrations/085_TurnItemCounts.ts new file mode 100644 index 0000000000000..a432046e6361a --- /dev/null +++ b/apps/server/src/persistence/Migrations/085_TurnItemCounts.ts @@ -0,0 +1,60 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +/** + * Shell reads count items per (thread, run) here instead of walking every turn item. Triggers + * keep it exact for every writer, including projection rebuilds and dev-database pruning. + */ +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // run_key is '' for items without a run; NULLs never conflict in a primary key. + yield* sql` + CREATE TABLE orchestration_v2_projection_turn_item_counts ( + thread_id TEXT NOT NULL, + run_key TEXT NOT NULL, + item_count INTEGER NOT NULL, + PRIMARY KEY (thread_id, run_key) + ) WITHOUT ROWID + `; + yield* sql` + INSERT INTO orchestration_v2_projection_turn_item_counts (thread_id, run_key, item_count) + SELECT thread_id, COALESCE(run_id, ''), COUNT(*) + FROM orchestration_v2_projection_turn_items + GROUP BY thread_id, run_id + `; + // An upsert that hits an existing item fires only the update trigger. + yield* sql` + CREATE TRIGGER orchestration_v2_turn_item_counts_insert + AFTER INSERT ON orchestration_v2_projection_turn_items + BEGIN + INSERT INTO orchestration_v2_projection_turn_item_counts (thread_id, run_key, item_count) + VALUES (new.thread_id, COALESCE(new.run_id, ''), 1) + ON CONFLICT (thread_id, run_key) DO UPDATE SET item_count = item_count + 1; + END + `; + yield* sql` + CREATE TRIGGER orchestration_v2_turn_item_counts_update + AFTER UPDATE OF thread_id, run_id ON orchestration_v2_projection_turn_items + WHEN old.thread_id IS NOT new.thread_id OR old.run_id IS NOT new.run_id + BEGIN + UPDATE orchestration_v2_projection_turn_item_counts SET item_count = item_count - 1 + WHERE thread_id = old.thread_id AND run_key = COALESCE(old.run_id, ''); + DELETE FROM orchestration_v2_projection_turn_item_counts + WHERE thread_id = old.thread_id AND run_key = COALESCE(old.run_id, '') AND item_count <= 0; + INSERT INTO orchestration_v2_projection_turn_item_counts (thread_id, run_key, item_count) + VALUES (new.thread_id, COALESCE(new.run_id, ''), 1) + ON CONFLICT (thread_id, run_key) DO UPDATE SET item_count = item_count + 1; + END + `; + yield* sql` + CREATE TRIGGER orchestration_v2_turn_item_counts_delete + AFTER DELETE ON orchestration_v2_projection_turn_items + BEGIN + UPDATE orchestration_v2_projection_turn_item_counts SET item_count = item_count - 1 + WHERE thread_id = old.thread_id AND run_key = COALESCE(old.run_id, ''); + DELETE FROM orchestration_v2_projection_turn_item_counts + WHERE thread_id = old.thread_id AND run_key = COALESCE(old.run_id, '') AND item_count <= 0; + END + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index dcb876e18b9bf..3c2cb075cfab4 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -64,6 +64,7 @@ describe("V2 preview upgrade", () => { [82, "TurnItemConversationIndex"], [83, "CommandReadIndexes"], [84, "LastExecutedRunIndex"], + [85, "TurnItemCounts"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -171,6 +172,7 @@ describe("V2 preview upgrade", () => { [82, "TurnItemConversationIndex"], [83, "CommandReadIndexes"], [84, "LastExecutedRunIndex"], + [85, "TurnItemCounts"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); diff --git a/docs/internals/performance-regressions.md b/docs/internals/performance-regressions.md index e641933b313fc..72046a02bdfd5 100644 --- a/docs/internals/performance-regressions.md +++ b/docs/internals/performance-regressions.md @@ -104,9 +104,9 @@ sessions. The orchestrator still validates the current thread before applying se Read receipts use `ProjectionStore.getThread` to load only canonical thread metadata. A visit advances `lastVisitedAt` monotonically without changing `updatedAt` or decoding history. Shell -queries use a covering thread/run index for item counts and a partial index for nonterminal -background items, including idle items. Latest user-message reads use an index ordered by update -time. +queries read item counts from a per-thread, per-run table that turn-item triggers keep exact, and +use a partial index for nonterminal background items, including idle items. Latest user-message +reads use an index ordered by update time. Command palette results keep live VCS and linked pull-request queries leased to the visible viewport, including a small overscan region. Cached badges remain available while leases are From e3b6e99d7fbad673acbb41c4f559f677f639be97 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 02:29:43 -0800 Subject: [PATCH 4/4] test(server): shell read plans and stored item counts hold through rewrites and rebuilds One plan case now covers every shell lookup that used to read thread history: the active run lookups, the last executed run, and the item counts, which must not read turn items. The item count case writes an item again, moves it between a live run, a rolled back run and no run, rolls a run back and forward, and writes an item before its run exists. These cases come from #34666. The rebuild case checks the shell count after an item rewrite and after a projection rebuild. Co-authored-by: Braulio Oliveira Co-Authored-By: Claude Fable 5.2 --- .../FoundationPersistence.test.ts | 2 + .../orchestration-v2/ProjectionStore.test.ts | 75 ++++++++++++------- 2 files changed, 48 insertions(+), 29 deletions(-) diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index 5dc42df07674f..6a4c51a7a4278 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -3956,6 +3956,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { }); const afterUpdate = yield* projectionStore.getThreadProjection(threadId); assert.equal(afterUpdate.turnItems[0]?.ordinal, ordinals[0]); + assert.equal((yield* projectionStore.getThreadShell(threadId))?.itemCount, 151); assert.equal((yield* maintenance.verify).valid, true); yield* sql` @@ -3970,6 +3971,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { const rebuilt = yield* maintenance.rebuild; assert.isTrue(rebuilt.valid); assert.lengthOf((yield* projectionStore.getThreadProjection(threadId)).turnItems, 151); + assert.equal((yield* projectionStore.getThreadShell(threadId))?.itemCount, 151); }), ); }); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 15ebc75e9c99a..da7592bd65a96 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -5221,7 +5221,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); - it.effect("looks up shell stop requests and active runs through their indexes", () => + it.effect("reads shell runs and item counts without walking a thread's history", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; const sql = yield* SqlClient.SqlClient; @@ -5230,14 +5230,16 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { yield* projectionStore .getShellSnapshot({ projectId, location: "active" }) .pipe(Effect.withTracer(tracer)); + const planOf = (statement: string) => + sql + .unsafe<{ readonly detail: string }>(`EXPLAIN QUERY PLAN ${statement}`) + .pipe(Effect.map((rows) => rows.map((row) => row.detail))); const threadQuery = statements.find((statement) => statement.includes("AS latest_run_stop_requested"), ); assert.isDefined(threadQuery); - const plan = (yield* sql.unsafe<{ readonly detail: string }>( - `EXPLAIN QUERY PLAN ${threadQuery}`, - )).map((row) => row.detail); + const plan = yield* planOf(threadQuery); const stopRequestLookups = plan.filter((detail) => detail.startsWith("SEARCH stop_request ")); assert.lengthOf(stopRequestLookups, 2); for (const lookup of stopRequestLookups) { @@ -5252,32 +5254,20 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { for (const lookup of activeRunLookups) { assert.include(lookup, "orchestration_v2_projection_runs_thread_status_idx"); } - }), - ); - - it.effect("reads a thread's last executed run through the last executed run index", () => - Effect.gen(function* () { - const projectionStore = yield* ProjectionStore.ProjectionStoreV2; - const sql = yield* SqlClient.SqlClient; - const { projectId } = yield* assertProjectScopedShells("sql-shell-last-executed-index"); - const { statements, tracer } = traceSqlStatements(); - yield* projectionStore - .getShellSnapshot({ projectId, location: "active" }) - .pipe(Effect.withTracer(tracer)); - - const threadQuery = statements.find((statement) => - statement.includes("AS latest_run_stop_requested"), - ); - assert.isDefined(threadQuery); - const plan = (yield* sql.unsafe<{ readonly detail: string }>( - `EXPLAIN QUERY PLAN ${threadQuery}`, - )).map((row) => row.detail); const lastExecutedIndex = plan.findIndex((detail) => detail.includes("orchestration_v2_projection_runs_last_executed_idx"), ); assert.notStrictEqual(lastExecutedIndex, -1); // The lookup that follows the search is the sort it no longer needs. assert.notInclude(plan[lastExecutedIndex + 1] ?? "", "USE TEMP B-TREE"); + + const countQueries = statements.filter((statement) => statement.includes("item_count")); + assert.isNotEmpty(countQueries); + for (const countQuery of countQueries) { + for (const detail of yield* planOf(countQuery)) { + assert.notInclude(detail, "orchestration_v2_projection_turn_items"); + } + } }), ); @@ -5532,6 +5522,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }, }); } + let runWrites = 0; const applyRun = ( threadId: ThreadId, suffix: string, @@ -5540,9 +5531,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { ) => { const runId = RunId.make(`run:shell-item-counts:${suffix}`); const rootNodeId = NodeId.make(`node:shell-item-counts:${suffix}`); + runWrites += 1; return projectionStore .apply({ - id: EventId.make(`event:shell-item-counts:${suffix}:run`), + id: EventId.make(`event:shell-item-counts:${suffix}:run:${runWrites}`), type: "run.created", threadId, runId, @@ -5569,11 +5561,14 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }) .pipe(Effect.as(runId)); }; - let ordinal = 0; - const applyItem = (threadId: ThreadId, runId: RunId | null) => { - ordinal += 1; + let items = 0; + let writes = 0; + // Passing `rewrite` writes that earlier item again, possibly under another run. + const applyItem = (threadId: ThreadId, runId: RunId | null, rewrite?: number) => { + const ordinal = rewrite ?? (items += 1); + writes += 1; return projectionStore.apply({ - id: EventId.make(`event:shell-item-counts:item:${ordinal}`), + id: EventId.make(`event:shell-item-counts:write:${writes}`), type: "turn-item.updated", threadId, ...(runId === null ? {} : { runId }), @@ -5626,6 +5621,28 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { ); assert.equal(shell?.itemCount, itemCount); } + + // Writing an item again keeps its count; moving it to another run moves the count. + const itemCountOf = (threadId: ThreadId) => + projectionStore.getThreadShell(threadId).pipe(Effect.map((shell) => shell?.itemCount)); + yield* applyItem(busyThreadId, completedRunId, 3); + assert.equal(yield* itemCountOf(busyThreadId), 4); + yield* applyItem(busyThreadId, rolledBackRunId, 3); + assert.equal(yield* itemCountOf(busyThreadId), 3); + yield* applyItem(busyThreadId, null, 3); + assert.equal(yield* itemCountOf(busyThreadId), 4); + + // A run's rollback uncounts its items, and writing the run again counts them back. + yield* applyRun(busyThreadId, "busy-completed", 1, true); + assert.equal(yield* itemCountOf(busyThreadId), 3); + yield* applyRun(busyThreadId, "busy-completed", 1, false); + assert.equal(yield* itemCountOf(busyThreadId), 4); + // Items written before their run exist count once it does. + const lateRunId = RunId.make("run:shell-item-counts:late"); + yield* applyItem(otherThreadId, lateRunId); + assert.equal(yield* itemCountOf(otherThreadId), 1); + yield* applyRun(otherThreadId, "late", 2, false); + assert.equal(yield* itemCountOf(otherThreadId), 2); }), );