From 11f4abef802b3d280e7ab8a82c16ef0ef8dfb659 Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Wed, 11 Nov 2026 18:59:56 -0500 Subject: [PATCH] perf(orchestration): open no database for profiles without one, and read open-time sweeps through indexes (#41283) --- .../construction/runtime-host-services.ts | 4 +- ...ery-without-orchestration-database.test.ts | 52 +++++++++++++++ .../orchestration/db/orchestration-db.ts | 2 + ...ing-dispatch-statement-compilation.test.ts | 2 + ...structured-worker-orca-session-backfill.ts | 29 ++++++--- .../db/startup-sweep-query-plans.test.ts | 65 +++++++++++++++++++ .../worker-terminal-recovery.ts | 5 +- .../worker-terminal-release-reconciliation.ts | 6 +- .../worker-terminal-release-runtime.ts | 1 + ...cy-worker-terminal-recovery-persistence.ts | 15 +++-- .../approvals/worker-approval-report.ts | 7 +- 11 files changed, 169 insertions(+), 19 deletions(-) create mode 100644 src/main/runtime/orca-runtime-tests/worker-recovery-without-orchestration-database.test.ts create mode 100644 src/main/runtime/orchestration/db/startup-sweep-query-plans.test.ts diff --git a/src/main/runtime/construction/runtime-host-services.ts b/src/main/runtime/construction/runtime-host-services.ts index 04d2a75acf4..3b972285a01 100644 --- a/src/main/runtime/construction/runtime-host-services.ts +++ b/src/main/runtime/construction/runtime-host-services.ts @@ -207,7 +207,7 @@ export abstract class OrcaRuntimeWithPreservedBranchCleanup extends OrcaRuntimeW protected readonly legacyWorkerRecoveryPersistence = new RuntimeLegacyWorkerTerminalRecoveryPersistence( () => this.store, - () => this.getOrchestrationDb(), + () => this.getExistingOrchestrationDb(), (worktreeId) => this.tryGetWorkspaceSessionHostIdForWorktree(worktreeId) ) @@ -248,7 +248,7 @@ export abstract class OrcaRuntimeWithPreservedBranchCleanup extends OrcaRuntimeW canRecoverPersistentLocalPtys: () => this.canRecoverPersistentLocalPtysFn(), isTerminalProvenAbsent: (candidate) => this.isLeafPtyProvenAbsent(candidate.ptyId), hasRequestedReleases: () => - this.getOrchestrationDb().listWorkerTerminalReleaseBacklog(1).length > 0, + (this.getExistingOrchestrationDb()?.listWorkerTerminalReleaseBacklog(1).length ?? 0) > 0, reconcileRequestedReleases: () => reconcileRequestedWorkerTerminalReleases(this), reconcile: (options) => this.reconcileLegacyWorkerTerminals(options), updateRetry: (plan, deferredDispatchIds, options) => diff --git a/src/main/runtime/orca-runtime-tests/worker-recovery-without-orchestration-database.test.ts b/src/main/runtime/orca-runtime-tests/worker-recovery-without-orchestration-database.test.ts new file mode 100644 index 00000000000..a3e49471887 --- /dev/null +++ b/src/main/runtime/orca-runtime-tests/worker-recovery-without-orchestration-database.test.ts @@ -0,0 +1,52 @@ +import { existsSync, mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it, vi } from 'vitest' +import './runtime-test-module-registration.spec' +import './orca-runtime-test-lifecycle.spec' +import { OrcaRuntimeService } from '../orca-runtime' +import { + makeRuntimeStoreWithWorkspaceSession, + makeWorkspaceSessionWithHeadlessTerminal +} from './orca-runtime-test-fixtures.spec' +import { withDurableRuntimeStore } from './runtime-durable-store-fixture' + +describe('startup worker recovery in a profile that never used orchestration', () => { + const dirs: string[] = [] + afterEach(() => { + for (const dir of dirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }) + } + }) + + it('does not create the orchestration database', async () => { + const dir = mkdtempSync(join(tmpdir(), 'orca-no-orchestration-')) + dirs.push(dir) + const dbPath = join(dir, 'orchestration.db') + const { runtimeStore } = makeRuntimeStoreWithWorkspaceSession( + makeWorkspaceSessionWithHeadlessTerminal() + ) + // A profile whose userData holds no orchestration database. + class FreshProfileRuntime extends OrcaRuntimeService { + protected override orchestrationDbPath(): string { + return dbPath + } + } + const runtime = new FreshProfileRuntime( + withDurableRuntimeStore({ ...runtimeStore, flushOrThrow: vi.fn() }), + undefined, + { canRecoverPersistentLocalPtys: () => true } + ) + + await expect(runtime.reconcileLegacyWorkerTerminals()).resolves.toEqual({ + adoptedDispatchIds: [], + exitedDispatchIds: [], + deferredDispatchIds: [] + }) + // Release reconciliation runs after the pass returns; shutdown drains it. + await runtime.stopLegacyWorkerTerminalRecovery() + + expect(existsSync(dbPath)).toBe(false) + expect(runtime.getExistingOrchestrationDb()).toBeNull() + }) +}) diff --git a/src/main/runtime/orchestration/db/orchestration-db.ts b/src/main/runtime/orchestration/db/orchestration-db.ts index 3fcd88f18b1..dd78a109f35 100644 --- a/src/main/runtime/orchestration/db/orchestration-db.ts +++ b/src/main/runtime/orchestration/db/orchestration-db.ts @@ -80,6 +80,8 @@ class OrchestrationDbCore { CREATE INDEX IF NOT EXISTS idx_messages_native_direct_work ON messages(to_handle, type) WHERE read = 0 AND delivered_at IS NULL AND delivery_contract <> 'audit_only'; + CREATE INDEX IF NOT EXISTS idx_messages_stop_reports + ON messages(run_id) WHERE id LIKE 'msg\\_stop\\_%' ESCAPE '\\'; `) reconcileSettledWorkerDispatches(this.db) createCoordinatorMailRoutingTrigger.call(this as unknown as OrchestrationDb) diff --git a/src/main/runtime/orchestration/db/remaining-dispatch-statement-compilation.test.ts b/src/main/runtime/orchestration/db/remaining-dispatch-statement-compilation.test.ts index f375a451352..b8932847e2a 100644 --- a/src/main/runtime/orchestration/db/remaining-dispatch-statement-compilation.test.ts +++ b/src/main/runtime/orchestration/db/remaining-dispatch-statement-compilation.test.ts @@ -61,6 +61,8 @@ function fixture(count = 1, version: 'fresh' | 'v41' = 'fresh') { ${DROP_REMOVED_WORK_TRIGGERS_SQL} DROP INDEX idx_runs_coordinator_orca_session_id; DROP INDEX idx_dispatch_assignee_orca_session_id; + DROP INDEX idx_dispatch_assignee_orca_session_backfill; + DROP INDEX idx_dispatch_creator_orca_session_backfill; DROP INDEX idx_dispatch_session_presentation; DROP TRIGGER trg_runs_remember_coordinator_insert; DROP TRIGGER trg_runs_remember_coordinator_update; diff --git a/src/main/runtime/orchestration/db/schema/structured-worker-orca-session-backfill.ts b/src/main/runtime/orchestration/db/schema/structured-worker-orca-session-backfill.ts index e706cd05ad0..fd4eee3efa9 100644 --- a/src/main/runtime/orchestration/db/schema/structured-worker-orca-session-backfill.ts +++ b/src/main/runtime/orchestration/db/schema/structured-worker-orca-session-backfill.ts @@ -23,9 +23,21 @@ const RECORDED_WORKER_SESSIONS_SQL = ` SELECT terminal_handle, process_incarnation FROM worker_terminal_resources WHERE terminal_handle GLOB ? AND process_incarnation GLOB ?` +// Partial indexes hold exactly the rows still to backfill, so each open reads those, not every Dispatch. +const ASSIGNEE_PENDING_SQL = `assignee_orca_session_id IS NULL + AND (process_incarnation GLOB '${INCARNATION_GLOB}' OR assignee_handle GLOB '${HANDLE_GLOB}')` +const CREATOR_PENDING_SQL = `creator_orca_session_id IS NULL AND creator_handle GLOB '${HANDLE_GLOB}'` + /** Backfill only unambiguous local non-WSL structured incarnation evidence, never pane identity or a current recorded id. - * Repeat on every open because rollback writers can omit these columns after the v42 stamp. */ + * Repeat on every open because rollback writers can omit these columns after the v42 stamp; SQLite keeps + * the partial indexes current for every writer, old builds included. */ export function backfillStructuredWorkerOrcaSessionIds(db: Database.Database): void { + db.exec(` + CREATE INDEX IF NOT EXISTS idx_dispatch_assignee_orca_session_backfill + ON dispatch_contexts(id) WHERE ${ASSIGNEE_PENDING_SQL}; + CREATE INDEX IF NOT EXISTS idx_dispatch_creator_orca_session_backfill + ON dispatch_contexts(id) WHERE ${CREATOR_PENDING_SQL}; + `) let recordedSessions: Map> | undefined const orcaSessionIdFor = (handle: unknown, incarnation: unknown): string | null => { if (incarnation != null && typeof incarnation !== 'string') { @@ -52,11 +64,11 @@ export function backfillStructuredWorkerOrcaSessionIds(db: Database.Database): v const assignees = db .prepare( - `SELECT id, assignee_handle, process_incarnation FROM dispatch_contexts - WHERE assignee_orca_session_id IS NULL - AND (process_incarnation GLOB ? OR assignee_handle GLOB ?)` + `SELECT id, assignee_handle, process_incarnation + FROM dispatch_contexts INDEXED BY idx_dispatch_assignee_orca_session_backfill + WHERE ${ASSIGNEE_PENDING_SQL}` ) - .all(INCARNATION_GLOB, HANDLE_GLOB) + .all() const setAssignee = db.prepare( `UPDATE dispatch_contexts SET assignee_orca_session_id = ? WHERE id = ? AND assignee_orca_session_id IS NULL` @@ -70,10 +82,11 @@ export function backfillStructuredWorkerOrcaSessionIds(db: Database.Database): v const creators = db .prepare( - `SELECT id, creator_handle FROM dispatch_contexts - WHERE creator_orca_session_id IS NULL AND creator_handle GLOB ?` + `SELECT id, creator_handle + FROM dispatch_contexts INDEXED BY idx_dispatch_creator_orca_session_backfill + WHERE ${CREATOR_PENDING_SQL}` ) - .all(HANDLE_GLOB) + .all() const setCreator = db.prepare( `UPDATE dispatch_contexts SET creator_orca_session_id = ? WHERE id = ? AND creator_orca_session_id IS NULL` diff --git a/src/main/runtime/orchestration/db/startup-sweep-query-plans.test.ts b/src/main/runtime/orchestration/db/startup-sweep-query-plans.test.ts new file mode 100644 index 00000000000..01c71fef1e5 --- /dev/null +++ b/src/main/runtime/orchestration/db/startup-sweep-query-plans.test.ts @@ -0,0 +1,65 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { OrchestrationDb } from './orchestration-db' +import { recentOpenApprovalReports } from '../workers/approvals/worker-approval-report' +import { backfillStructuredWorkerOrcaSessionIds } from './schema/structured-worker-orca-session-backfill' + +/** The query plan of every statement `act` prepares that reads one of `tables`. */ +function plansOf(db: OrchestrationDb, tables: string[], act: () => void): string[] { + const prepare = vi.spyOn(db.db, 'prepare') + act() + const statements = prepare.mock.calls + .map(([sql]) => sql) + .filter((sql) => /^\s*SELECT/i.test(sql) && tables.some((table) => sql.includes(table))) + prepare.mockRestore() + expect(statements.length).toBeGreaterThan(0) + return statements.flatMap((sql) => + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: EXPLAIN QUERY PLAN rows carry a detail column. + (db.db.prepare(`EXPLAIN QUERY PLAN ${sql}`).all() as { detail: string }[]).map( + ({ detail }) => detail + ) + ) +} + +const fullScans = (plans: string[]): string[] => + plans.filter((detail) => /^SCAN \w+$/.test(detail.trim())) + +describe('open-time sweeps read through their indexes, not all history', () => { + let db: OrchestrationDb + + beforeEach(() => { + db = new OrchestrationDb(':memory:') + }) + + afterEach(() => { + db.close() + }) + + it('the restart approval sweep reads only stop reports', () => { + const plans = plansOf(db, ['messages'], () => recentOpenApprovalReports(db)) + expect(fullScans(plans)).toEqual([]) + expect(plans.filter((detail) => detail.includes('idx_messages_stop_reports'))).toHaveLength(2) + }) + + it('legacy worker recovery reads only live workers', () => { + const plans = plansOf(db, ['dispatch_contexts'], () => + db.listLegacyWorkerTerminalRecoveryRows() + ) + expect(fullScans(plans)).toEqual([]) + expect(plans).toEqual( + expect.arrayContaining([expect.stringContaining('idx_worker_dispatches_recoverable')]) + ) + }) + + it('the Orca session backfill reads only rows it may still fill', () => { + const plans = plansOf(db, ['dispatch_contexts'], () => + backfillStructuredWorkerOrcaSessionIds(db.db) + ) + expect(fullScans(plans)).toEqual([]) + expect(plans).toEqual( + expect.arrayContaining([ + expect.stringContaining('idx_dispatch_assignee_orca_session_backfill'), + expect.stringContaining('idx_dispatch_creator_orca_session_backfill') + ]) + ) + }) +}) diff --git a/src/main/runtime/orchestration/db/worker-dispatch/worker-terminal-recovery.ts b/src/main/runtime/orchestration/db/worker-dispatch/worker-terminal-recovery.ts index 64a58fd6fe6..525e5c4ca62 100644 --- a/src/main/runtime/orchestration/db/worker-dispatch/worker-terminal-recovery.ts +++ b/src/main/runtime/orchestration/db/worker-dispatch/worker-terminal-recovery.ts @@ -19,14 +19,15 @@ export function listLegacyWorkerTerminalRecoveryRows( if (dispatchIds?.length === 0) { return [] } + // Why: drive from the live workers' partial index; ORDER BY dc.rowid alone made the planner scan every Dispatch. const rows = this.db .prepare( `SELECT dc.id AS dispatch_id, dc.task_id, dc.status AS dispatch_status, dc.contract_version, dc.assignee_handle, dc.assignee_pane_key, dc.process_incarnation, wd.state AS worker_state, wd.worktree_id, wd.agent_terminal_handle - FROM dispatch_contexts dc - INNER JOIN worker_dispatches wd ON wd.dispatch_id = dc.id + FROM worker_dispatches wd INDEXED BY idx_worker_dispatches_recoverable + CROSS JOIN dispatch_contexts dc ON dc.id = wd.dispatch_id WHERE wd.state IN ('starting', 'ready', 'start_unknown', 'stopping', 'stop_unknown') ${dispatchIds ? 'AND wd.dispatch_id IN (SELECT value FROM json_each(?))' : ''} ORDER BY dc.rowid` diff --git a/src/main/runtime/orchestration/terminal-authority/release/worker-terminal-release-reconciliation.ts b/src/main/runtime/orchestration/terminal-authority/release/worker-terminal-release-reconciliation.ts index 0f7852549dd..9fa5749cf9c 100644 --- a/src/main/runtime/orchestration/terminal-authority/release/worker-terminal-release-reconciliation.ts +++ b/src/main/runtime/orchestration/terminal-authority/release/worker-terminal-release-reconciliation.ts @@ -65,7 +65,11 @@ async function runReconciliationPasses( async function reconcileRequestedWorkerTerminalReleasesOnce( runtime: WorkerTerminalReleaseRuntime ): Promise { - const db = runtime.getOrchestrationDb() + // Startup reconciles for everyone; a profile without a database has nothing to release. + const db = runtime.getExistingOrchestrationDb() + if (!db) { + return emptyResult() + } const backlog = db.listWorkerTerminalReleaseBacklog() const result = { ...emptyResult(), attempted: backlog.length } for (const resource of backlog) { diff --git a/src/main/runtime/orchestration/terminal-authority/release/worker-terminal-release-runtime.ts b/src/main/runtime/orchestration/terminal-authority/release/worker-terminal-release-runtime.ts index 713af5a4c2c..1ae95fe6a16 100644 --- a/src/main/runtime/orchestration/terminal-authority/release/worker-terminal-release-runtime.ts +++ b/src/main/runtime/orchestration/terminal-authority/release/worker-terminal-release-runtime.ts @@ -9,6 +9,7 @@ export type WorkerTerminalReleaseRuntime = StructuredSessionSurfaceRuntime & | 'callOrchestrationWorkerServer' | 'getOrchestrationFleetAgentStatusSnapshot' | 'getOrchestrationDb' + | 'getExistingOrchestrationDb' | 'showTerminal' | 'resolveTerminalHandleByProcessIncarnation' | 'getTerminalPaneKey' diff --git a/src/main/runtime/orchestration/terminal-authority/runtime-legacy-worker-terminal-recovery-persistence.ts b/src/main/runtime/orchestration/terminal-authority/runtime-legacy-worker-terminal-recovery-persistence.ts index 0172e6a9753..d19ad29f918 100644 --- a/src/main/runtime/orchestration/terminal-authority/runtime-legacy-worker-terminal-recovery-persistence.ts +++ b/src/main/runtime/orchestration/terminal-authority/runtime-legacy-worker-terminal-recovery-persistence.ts @@ -28,7 +28,8 @@ export type LostWorkerReport = { export class RuntimeLegacyWorkerTerminalRecoveryPersistence { constructor( private readonly getStore: () => RuntimeStore | null, - private readonly getDb: () => OrchestrationDb, + // A profile without a database has no workers to recover. + private readonly getDb: () => OrchestrationDb | null, private readonly getHostId: (worktreeId: string) => ExecutionHostId | null ) {} @@ -124,6 +125,9 @@ export class RuntimeLegacyWorkerTerminalRecoveryPersistence { ): boolean { try { const db = this.getDb() + if (!db) { + return false + } const before = db.getDispatchContextById(candidate.dispatchId) const worker = db.reconcileMissingWorkerTerminal( candidate.dispatchId, @@ -169,10 +173,13 @@ export class RuntimeLegacyWorkerTerminalRecoveryPersistence { private getPlan(dispatchIds?: readonly string[]): LegacyWorkerTerminalRecoveryPlan | null { try { + const db = this.getDb() return planLegacyWorkerTerminalRecovery( - dispatchIds - ? this.getDb().listLegacyWorkerTerminalRecoveryRows(dispatchIds) - : this.getDb().listLegacyWorkerTerminalRecoveryRows() + !db + ? [] + : dispatchIds + ? db.listLegacyWorkerTerminalRecoveryRows(dispatchIds) + : db.listLegacyWorkerTerminalRecoveryRows() ) } catch (error) { console.warn('[orchestration] failed to plan legacy worker terminal recovery', error) diff --git a/src/main/runtime/orchestration/workers/approvals/worker-approval-report.ts b/src/main/runtime/orchestration/workers/approvals/worker-approval-report.ts index 840b6a4c5a8..ba920641bcb 100644 --- a/src/main/runtime/orchestration/workers/approvals/worker-approval-report.ts +++ b/src/main/runtime/orchestration/workers/approvals/worker-approval-report.ts @@ -58,11 +58,14 @@ export function recentOpenApprovalReports( .array() .parse( db.db - .prepare(`SELECT o.id, o.payload, o.from_handle FROM messages o + // Both sides read only Orca's stop reports through their partial index, never all mail. + .prepare(`SELECT o.id, o.payload, o.from_handle + FROM messages o INDEXED BY idx_messages_stop_reports WHERE o.type = 'escalation' AND o.id LIKE 'msg\\_stop\\_%' ESCAPE '\\' AND json_valid(o.payload) AND json_extract(o.payload, '$.approval.state') = 'open' -- A closed approval keeps its open report; leave it out before the bound. - AND NOT EXISTS (SELECT 1 FROM messages c WHERE c.run_id = o.run_id + AND NOT EXISTS (SELECT 1 FROM messages c INDEXED BY idx_messages_stop_reports + WHERE c.run_id = o.run_id AND c.id LIKE 'msg\\_stop\\_%' ESCAPE '\\' AND json_valid(c.payload) AND json_extract(c.payload, '$.approval.approvalId') = json_extract(o.payload, '$.approval.approvalId') AND json_extract(c.payload, '$.approval.state') <> 'open')