From 4e3e1ad18370727e6633ef886ce0bc635a8006be Mon Sep 17 00:00:00 2001 From: credkeydev Date: Tue, 20 Oct 2026 16:29:40 -0700 Subject: [PATCH] perf(server): avoid historical scans during startup recovery --- .../src/orchestration-v2/ProjectionStore.ts | 14 +- apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/055_OrchestrationV2.test.ts | 4 +- .../068_StartupRecoveryIndexes.test.ts | 170 ++++++++++++++++++ .../Migrations/068_StartupRecoveryIndexes.ts | 20 +++ .../reconcileV2PreviewMigration.test.ts | 2 + 6 files changed, 203 insertions(+), 9 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/068_StartupRecoveryIndexes.test.ts create mode 100644 apps/server/src/persistence/Migrations/068_StartupRecoveryIndexes.ts diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 747c7c6f395c2..1ae9a01b312cf 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3930,15 +3930,13 @@ export const layer: Layer.Layer = ELSE 0 END ) SELECT thread_id FROM orchestration_v2_projection_runs - WHERE status IN ('preparing', 'starting', 'running', 'waiting') - UNION - -- A held queue already went through recovery; rereading it on - -- every boot costs a projection read per held thread. - SELECT thread_id FROM orchestration_v2_projection_runs - WHERE status = 'queued' - AND CASE WHEN json_valid(payload_json) + -- Match the recovery index's partial predicate: SQLite does not + -- infer that either a subset IN list or status = 'queued' implies it. + WHERE status IN ('queued', 'preparing', 'starting', 'running', 'waiting') + -- Held queues have already gone through recovery. + AND (status <> 'queued' OR CASE WHEN json_valid(payload_json) THEN json_extract(payload_json, '$.queueHeld') IS NOT 1 - ELSE 1 END + ELSE 1 END) UNION SELECT thread_id FROM orchestration_v2_projection_runtime_requests WHERE status = 'pending' diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 8dfbfe712a7de..0899505adc2bf 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -70,6 +70,7 @@ import Migration0064 from "./Migrations/064_CanonicalizeThreadPullRequestLinks.t import Migration0065 from "./Migrations/065_CodexUsageNativeTurnIndex.ts"; import Migration0066 from "./Migrations/066_RepairClientSettlementTimestamps.ts"; import Migration0067 from "./Migrations/067_WebhookDedupe.ts"; +import Migration0068 from "./Migrations/068_StartupRecoveryIndexes.ts"; export const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -140,6 +141,7 @@ export const migrationEntries = [ [65, "CodexUsageNativeTurnIndex", Migration0065], [66, "RepairClientSettlementTimestamps", Migration0066], [67, "WebhookDedupe", Migration0067], + [68, "StartupRecoveryIndexes", Migration0068], ] 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 5bbf3cd9a7e62..62de33ff31c79 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: 67 }, (_, index) => index + 1), + Array.from({ length: 68 }, (_, index) => index + 1), ); }), ); @@ -41,6 +41,7 @@ layer("055_OrchestrationV2", (it) => { [65, "CodexUsageNativeTurnIndex"], [66, "RepairClientSettlementTimestamps"], [67, "WebhookDedupe"], + [68, "StartupRecoveryIndexes"], ]); assert.deepStrictEqual(yield* runMigrations(), []); @@ -74,6 +75,7 @@ layer("055_OrchestrationV2", (it) => { { migration_id: 65, name: "CodexUsageNativeTurnIndex" }, { migration_id: 66, name: "RepairClientSettlementTimestamps" }, { migration_id: 67, name: "WebhookDedupe" }, + { migration_id: 68, name: "StartupRecoveryIndexes" }, ]); const tables = yield* sql<{ readonly name: string }>` diff --git a/apps/server/src/persistence/Migrations/068_StartupRecoveryIndexes.test.ts b/apps/server/src/persistence/Migrations/068_StartupRecoveryIndexes.test.ts new file mode 100644 index 0000000000000..8914bc3ddf6c0 --- /dev/null +++ b/apps/server/src/persistence/Migrations/068_StartupRecoveryIndexes.test.ts @@ -0,0 +1,170 @@ +import { assert, it } from "@effect/vitest"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +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 ProjectionStore from "../../orchestration-v2/ProjectionStore.ts"; +import { runMigrations } from "../Migrations.ts"; + +it.effect("upgrades startup recovery lookups without changing candidate selection", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const store = yield* ProjectionStore.ProjectionStoreV2; + yield* runMigrations({ toMigrationInclusive: 67 }); + + for (const threadId of [ + "history", + "active", + "queued", + "held", + "malformed", + "native", + "child", + "wrong-kind", + "attached-root", + "deleted", + "archived", + "duplicate", + ]) { + yield* sql`INSERT INTO orchestration_v2_projection_threads ${sql.insert({ + thread_id: threadId, + project_id: "project", + title: threadId, + default_provider: "codex", + runtime_mode: "full-access", + interaction_mode: "default", + created_at: "2026-10-20", + updated_at: "2026-10-20", + deleted_at: threadId === "deleted" ? "2026-10-20" : null, + archived_at: threadId === "archived" ? "2026-10-20" : null, + payload_json: "{}", + })}`; + } + for (const [threadId, status, payload] of [ + ["active", "running", '{"queueHeld":true}'], + ["queued", "queued", "{}"], + ["held", "queued", '{"queueHeld":true}'], + ["malformed", "queued", "{"], + ["deleted", "running", "{}"], + ["archived", "waiting", "{}"], + ["duplicate", "running", "{}"], + ["duplicate", "queued", "{}"], + ] as const) { + yield* sql`INSERT INTO orchestration_v2_projection_runs ${sql.insert({ + run_id: `${threadId}-${status}`, + thread_id: threadId, + ordinal: status === "queued" ? 2 : 1, + provider: "codex", + status, + requested_at: "2026-10-20", + payload_json: payload, + })}`; + } + for (const [id, threadId, origin, status, childThreadId] of [ + ["native-live", "native", "provider_native", "waiting", null], + ["app-live", "wrong-kind", "app_owned", "running", null], + ["child-link", "history", "app_owned", "completed", "child"], + ["wrong-kind-link", "history", "app_owned", "completed", "wrong-kind"], + ["attached-link", "history", "app_owned", "completed", "attached-root"], + ] as const) { + yield* sql`INSERT INTO orchestration_v2_projection_subagents ${sql.insert({ + subagent_id: id, + thread_id: threadId, + parent_node_id: "parent", + provider: "codex", + origin, + status, + child_thread_id: childThreadId, + updated_at: "2026-10-20", + payload_json: "{}", + })}`; + } + for (const [threadId, kind, runId] of [ + ["child", "root_turn", null], + ["wrong-kind", "tool_call", null], + ["attached-root", "root_turn", "finished-run"], + ] as const) { + yield* sql`INSERT INTO orchestration_v2_projection_nodes ${sql.insert({ + node_id: `${threadId}-node`, + thread_id: threadId, + run_id: runId, + root_node_id: `${threadId}-node`, + kind, + status: "running", + payload_json: "{}", + })}`; + } + + // Completed records dominate a real database. Their large payloads should + // not need to be read just to find the few remaining recovery candidates. + yield* sql` + WITH RECURSIVE history(n) AS (SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 2048) + INSERT INTO orchestration_v2_projection_runs + (run_id, thread_id, ordinal, provider, status, requested_at, payload_json) + SELECT 'historical-run-' || n, 'history', n, 'codex', 'completed', '2026-10-20', + json_object('padding', printf('%.*c', 2048, 'x')) FROM history + `; + yield* sql` + WITH RECURSIVE history(n) AS (SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 2048) + INSERT INTO orchestration_v2_projection_nodes + (node_id, thread_id, root_node_id, kind, status, payload_json) + SELECT 'historical-node-' || n, 'child', 'child-node', 'tool_call', 'completed', + json_object('padding', printf('%.*c', 2048, 'x')) FROM history + `; + yield* sql` + WITH RECURSIVE history(n) AS (SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 2048) + INSERT INTO orchestration_v2_projection_subagents + (subagent_id, thread_id, parent_node_id, provider, origin, status, updated_at, payload_json) + SELECT 'historical-subagent-' || n, 'history', 'parent', 'codex', + 'provider_native', 'completed', '2026-10-20', + json_object('padding', printf('%.*c', 2048, 'x')) FROM history + `; + + const expected = ["active", "archived", "child", "duplicate", "malformed", "native", "queued"]; + assert.deepStrictEqual((yield* store.getRecoveryThreadIds("runtime")).toSorted(), expected); + assert.deepStrictEqual(yield* runMigrations(), [[68, "StartupRecoveryIndexes"]]); + assert.deepStrictEqual(yield* runMigrations(), []); + + const statements: Array = []; + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options); + const end = span.end.bind(span); + span.end = (time, exit) => { + end(time, exit); + const query = span.attributes.get("db.query.text"); + if (typeof query === "string") statements.push(query); + }; + return span; + }, + }); + const candidates = yield* store.getRecoveryThreadIds("runtime").pipe(Effect.withTracer(tracer)); + assert.deepStrictEqual(candidates.toSorted(), expected); + + const statement = statements.find((query) => query.includes("pending_provider_threads")); + assert.isDefined(statement); + const plan = (yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${statement}`, + )).map(({ detail }) => detail); + const details = plan.join("\n"); + assert.include(details, "orchestration_v2_projection_runs_recovery_idx"); + assert.include(details, "orchestration_v2_projection_nodes_unattached_recovery_idx"); + assert.include(details, "orchestration_v2_projection_subagents_live_native_idx"); + assert.isFalse( + plan.some((detail) => detail === "SCAN orchestration_v2_projection_runs"), + details, + ); + assert.isFalse( + plan.some((detail) => detail === "SCAN orchestration_v2_projection_subagents"), + details, + ); + }).pipe( + Effect.provide( + ProjectionStore.layer.pipe( + Layer.provideMerge(NodeSqliteClient.layer({ filename: ":memory:" })), + ), + ), + ), +); diff --git a/apps/server/src/persistence/Migrations/068_StartupRecoveryIndexes.ts b/apps/server/src/persistence/Migrations/068_StartupRecoveryIndexes.ts new file mode 100644 index 0000000000000..f608a043990c9 --- /dev/null +++ b/apps/server/src/persistence/Migrations/068_StartupRecoveryIndexes.ts @@ -0,0 +1,20 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Startup recovery only needs thread IDs. Keep historical payloads out of + // these lookups, including root nodes without an attached run. + yield* sql` + CREATE INDEX orchestration_v2_projection_nodes_unattached_recovery_idx + ON orchestration_v2_projection_nodes(thread_id, kind, status) + WHERE run_id IS NULL + `; + yield* sql` + CREATE INDEX orchestration_v2_projection_subagents_live_native_idx + ON orchestration_v2_projection_subagents(thread_id) + WHERE origin = 'provider_native' + AND status IN ('pending', 'running', 'waiting') + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 990d450cde6b5..c9b3e83110c10 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -47,6 +47,7 @@ describe("V2 preview upgrade", () => { [65, "CodexUsageNativeTurnIndex"], [66, "RepairClientSettlementTimestamps"], [67, "WebhookDedupe"], + [68, "StartupRecoveryIndexes"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -137,6 +138,7 @@ describe("V2 preview upgrade", () => { [65, "CodexUsageNativeTurnIndex"], [66, "RepairClientSettlementTimestamps"], [67, "WebhookDedupe"], + [68, "StartupRecoveryIndexes"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), );