perf(orchestration): open no database for profiles without one, and read open-time sweeps through indexes (#41283)

This commit is contained in:
Jinwoo Hong
2026-11-11 15:59:56 -08:00
committed by GitHub
parent c2e716dbbb
commit 11f4abef80
11 changed files with 169 additions and 19 deletions
@@ -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) =>
@@ -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()
})
})
@@ -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)
@@ -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;
@@ -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<string, Set<string>> | 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`
@@ -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')
])
)
})
})
@@ -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`
@@ -65,7 +65,11 @@ async function runReconciliationPasses(
async function reconcileRequestedWorkerTerminalReleasesOnce(
runtime: WorkerTerminalReleaseRuntime
): Promise<WorkerTerminalReleaseReconciliationResult> {
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) {
@@ -9,6 +9,7 @@ export type WorkerTerminalReleaseRuntime = StructuredSessionSurfaceRuntime &
| 'callOrchestrationWorkerServer'
| 'getOrchestrationFleetAgentStatusSnapshot'
| 'getOrchestrationDb'
| 'getExistingOrchestrationDb'
| 'showTerminal'
| 'resolveTerminalHandleByProcessIncarnation'
| 'getTerminalPaneKey'
@@ -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)
@@ -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')