Reuse prepared statements for internal worker checks (#24573)

* Reuse prepared statements across remaining Dispatch reads

Use the existing schema-pinned complete Dispatch column list in five remaining read modules, enabling the existing bounded statement cache without caching results.

* Reuse prepared statements for internal worker checks

Use the existing complete Dispatch projection only for named-field internal consumers; preserve original wildcard reads for every returned failure snapshot.
This commit is contained in:
Neil
2026-10-03 15:31:23 -07:00
committed by GitHub
parent 558c10f616
commit caf8078d13
6 changed files with 335 additions and 22 deletions
@@ -3,6 +3,7 @@ import { OrchestrationError } from '../../orchestration-error'
import { generateId } from '../generated-id'
import type { OrchestrationDb } from '../orchestration-db'
import { transitionLifecycleWithDb } from '../lifecycle-transition'
import { DISPATCH_CONTEXT_COLUMN_LIST } from '../row-column-lists'
// ── Decision Gates ──
@@ -27,13 +28,15 @@ export function createGate(
}
const runId = task.run_id
this.requireRun(runId)
const active = this.db
const rawActive = this.db
.prepare(
`SELECT * FROM dispatch_contexts
`SELECT ${DISPATCH_CONTEXT_COLUMN_LIST} FROM dispatch_contexts
WHERE task_id = ? AND status IN ('pending', 'dispatched')
ORDER BY rowid DESC LIMIT 1`
)
.get(gate.taskId) as DispatchContextRow | undefined
.get(gate.taskId)
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: The schema-pinned complete Dispatch projection returns one Dispatch row or undefined; the adapter exposes unknown.
const active = rawActive as DispatchContextRow | undefined
if (
gate.requester &&
(!active ||
@@ -3,6 +3,7 @@ import { OrchestrationError } from '../../orchestration-error'
import { DISPATCH_CIRCUIT_BREAK_FAILURES } from './dispatch-circuit-breaker'
import type { OrchestrationDb } from '../orchestration-db'
import { getActiveDispatchForTask } from './task-dispatch-reconciliation'
import { DISPATCH_CONTEXT_COLUMN_LIST } from '../row-column-lists'
import {
beginLifecycleWriteTransaction,
commitLifecycleWriteTransaction,
@@ -46,11 +47,13 @@ export function settleActiveDispatchesForTask(
status: 'completed' | 'failed',
failure?: string
): void {
const rows = db.db
const rawRows = db.db
.prepare(
"SELECT * FROM dispatch_contexts WHERE task_id = ? AND status IN ('pending', 'dispatched')"
`SELECT ${DISPATCH_CONTEXT_COLUMN_LIST} FROM dispatch_contexts WHERE task_id = ? AND status IN ('pending', 'dispatched')`
)
.all(taskId) as DispatchContextRow[]
.all(taskId)
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: The existing complete Dispatch projection is pinned to this table's schema by row-column-lists.test.ts.
const rows = rawRows as DispatchContextRow[]
for (const row of rows) {
transitionLifecycleWithDb(db.db, {
entity: 'dispatch',
@@ -97,15 +100,17 @@ export function getStaleDispatches(
this: OrchestrationDb,
thresholdIso: string
): DispatchContextRow[] {
return this.db
const rows = this.db
.prepare(
`SELECT * FROM dispatch_contexts
`SELECT ${DISPATCH_CONTEXT_COLUMN_LIST} FROM dispatch_contexts
WHERE status = 'dispatched'
AND dispatched_at IS NOT NULL
AND julianday(dispatched_at) < julianday(?)
AND (last_heartbeat_at IS NULL OR julianday(last_heartbeat_at) < julianday(?))`
)
.all(thresholdIso, thresholdIso) as DispatchContextRow[]
.all(thresholdIso, thresholdIso)
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: The existing complete Dispatch projection is pinned to this table's schema by row-column-lists.test.ts.
return rows as DispatchContextRow[]
}
export function failDispatch(
@@ -164,9 +169,11 @@ export function failDispatch(
capability_revoked_at: before.capability_revoked_at ?? new Date().toISOString()
}
})
const ctx = this.db.prepare('SELECT * FROM dispatch_contexts WHERE id = ?').get(ctxId) as
| DispatchContextRow
| undefined
const rawContext = this.db
.prepare(`SELECT ${DISPATCH_CONTEXT_COLUMN_LIST} FROM dispatch_contexts WHERE id = ?`)
.get(ctxId)
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: The schema-pinned complete Dispatch projection returns one Dispatch row or undefined; the adapter exposes unknown.
const ctx = rawContext as DispatchContextRow | undefined
if (!ctx) {
commitLifecycleWriteTransaction(this.db, transaction)
return undefined
@@ -10,6 +10,7 @@ import type { OrchestrationDb } from './orchestration-db'
import type { OrcaSessionId } from '../../../../shared/orca-session-address'
import type { DispatchContextRow, RemoteDispatchAttachmentRow } from '../types'
import { potentiallyLiveRemoteAttachmentSql } from './federation/remote-attachment-liveness'
import { DISPATCH_CONTEXT_COLUMN_LIST } from './row-column-lists'
/**
* Who is creating a dispatch row, for nesting-depth purposes.
@@ -147,12 +148,12 @@ function findActiveDispatchForCreator(
}
const row = this.db
.prepare(
`SELECT * FROM dispatch_contexts
`SELECT ${DISPATCH_CONTEXT_COLUMN_LIST} FROM dispatch_contexts
WHERE assignee_orca_session_id = ? AND status IN ('pending', 'dispatched')
ORDER BY rowid DESC LIMIT 1`
)
.get(creator.orcaSessionId)
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: SELECT * over this table returns the row shape its schema and row type define, like every row cast in db/.
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: The schema-pinned complete Dispatch projection returns one Dispatch row or undefined; the adapter exposes unknown.
return row as DispatchContextRow | undefined
}
@@ -7,6 +7,7 @@ import { OrchestrationError } from '../../orchestration-error'
import { LEGACY_CONTRACT_VERSION } from '../contract-constants'
import { isEquivalentPaneKey } from '../pane-key-match'
import type { OrchestrationDb } from '../orchestration-db'
import { DISPATCH_CONTEXT_COLUMN_LIST } from '../row-column-lists'
export function resolveLegacyCompatibilityPrincipalByIdentity(
this: OrchestrationDb,
@@ -60,9 +61,10 @@ export function resolveLegacyWorkerCandidate(
? [this.getDispatchContextById(params.dispatchId)].filter(
(row): row is DispatchContextRow => row !== undefined
)
: (this.db
: // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: The existing complete Dispatch projection is pinned to this table's schema by row-column-lists.test.ts.
(this.db
.prepare(
`SELECT * FROM dispatch_contexts
`SELECT ${DISPATCH_CONTEXT_COLUMN_LIST} FROM dispatch_contexts
WHERE run_id = ? AND contract_version = ?
AND status IN ('pending', 'dispatched')
ORDER BY rowid`
@@ -2,6 +2,7 @@ import { parsePaneKey } from '../../../../../shared/stable-pane-id'
import type { DispatchContextRow, MessageType } from '../../types'
import { DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL, paneKeyMatchSuffix } from '../pane-key-match'
import type { OrchestrationDb } from '../orchestration-db'
import { DISPATCH_CONTEXT_COLUMN_LIST } from '../row-column-lists'
import {
ORCHESTRATION_DELIVERY_BATCH_LIMIT,
type ForeignDirectMailboxRoutingPage
@@ -13,25 +14,29 @@ export function findActiveDispatchForDirectMessageOwner(
directHandle: string,
paneKey?: string
): DispatchContextRow | undefined {
const exact = this.db
const rawExact = this.db
.prepare(
`SELECT * FROM dispatch_contexts
`SELECT ${DISPATCH_CONTEXT_COLUMN_LIST} FROM dispatch_contexts
WHERE run_id = ? AND assignee_handle = ? AND status IN ('pending', 'dispatched')
ORDER BY rowid DESC LIMIT 1`
)
.get(runId, directHandle) as DispatchContextRow | undefined
.get(runId, directHandle)
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: The schema-pinned complete Dispatch projection returns one Dispatch row or undefined; the adapter exposes unknown.
const exact = rawExact as DispatchContextRow | undefined
if (exact || !paneKey || !parsePaneKey(paneKey)) {
return exact
}
return this.db
const reminted = this.db
.prepare(
`SELECT * FROM dispatch_contexts
`SELECT ${DISPATCH_CONTEXT_COLUMN_LIST} FROM dispatch_contexts
WHERE run_id = ? AND assignee_pane_key IS NOT NULL
AND status IN ('pending', 'dispatched') AND instr(assignee_pane_key, ':') > 1
AND ${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL} = ?
ORDER BY rowid DESC LIMIT 1`
)
.get(runId, paneKeyMatchSuffix(paneKey)) as DispatchContextRow | undefined
.get(runId, paneKeyMatchSuffix(paneKey))
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: The schema-pinned complete Dispatch projection returns one Dispatch row or undefined; the adapter exposes unknown.
return reminted as DispatchContextRow | undefined
}
export function routeForeignDirectMessagesToOwnedMailboxes(
@@ -0,0 +1,295 @@
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { DatabaseSync } from 'node:sqlite'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { isOrcaSessionId } from '../../../../shared/orca-session-address'
import { OrchestrationDb } from './orchestration-db'
import { createRootDispatch } from './root-dispatch-test-fixture'
import { LEGACY_CONTRACT_VERSION } from './contract-constants'
import { DISPATCH_CONTEXT_COLUMNS } from './row-column-lists'
import { warnStaleDispatches } from '../coordinator-task-dispatch'
const databases: OrchestrationDb[] = []
const directories: string[] = []
const NOW = '2026-10-02T12:00:00.000Z'
const PANE = 'tab_worker:11111111-1111-4111-8111-111111111111'
afterEach(() => {
vi.restoreAllMocks()
vi.useRealTimers()
for (const db of databases.splice(0)) {
db.close()
}
for (const directory of directories.splice(0)) {
rmSync(directory, { recursive: true, force: true })
}
})
function fixture(count = 1, version: 'fresh' | 'v41' = 'fresh') {
const directory = mkdtempSync(join(tmpdir(), 'orca-remaining-dispatch-compilation-'))
directories.push(directory)
const path = join(directory, 'orchestration.db')
let writer = new OrchestrationDb(path)
databases.push(writer)
const run = writer.createRun({
objective: 'dispatch lifecycle',
coordinatorHandle: 'coordinator',
coordinatorPaneKey: 'tab_coordinator:22222222-2222-4222-8222-222222222222'
})
const entries = Array.from({ length: count }, (_, index) => {
const task = writer.createTask({ spec: `work ${index}`, runId: run.id })
const pane =
index === 0
? PANE
: `tab_worker:${String(index).padStart(8, '0')}-1111-4111-8111-111111111111`
const dispatch = createRootDispatch(writer, task.id, `worker_${index}`, pane)
writer.db
.prepare('UPDATE dispatch_contexts SET status = ?, dispatched_at = ?, depth = ? WHERE id = ?')
.run('dispatched', '2026-10-02 10:00:00', 2, dispatch.id)
writer.db.prepare('UPDATE tasks SET status = ? WHERE id = ?').run('dispatched', task.id)
const expected = writer.db
.prepare('SELECT * FROM dispatch_contexts WHERE id = ?')
.get(dispatch.id)
if (!expected) {
throw new Error('Missing dispatch fixture')
}
return { task, dispatch, expected }
})
if (version === 'v41') {
writer.db.exec(`
DROP INDEX idx_runs_coordinator_orca_session_id;
DROP INDEX idx_dispatch_assignee_orca_session_id;
DROP TRIGGER trg_runs_remember_coordinator_insert;
DROP TRIGGER trg_runs_remember_coordinator_update;
DROP TRIGGER trg_messages_route_coordinator_mail;
ALTER TABLE runs DROP COLUMN coordinator_orca_session_id;
ALTER TABLE runs DROP COLUMN coordinator_orca_session_id_generation;
ALTER TABLE dispatch_contexts DROP COLUMN assignee_orca_session_id;
ALTER TABLE dispatch_contexts DROP COLUMN creator_orca_session_id;
`)
writer.db.pragma('user_version = 41')
databases.pop()
writer.close()
writer = new OrchestrationDb(path)
databases.push(writer)
for (const entry of entries) {
const migrated = writer.db
.prepare('SELECT * FROM dispatch_contexts WHERE id = ?')
.get(entry.dispatch.id)
if (!migrated) {
throw new Error('Missing migrated dispatch')
}
entry.expected = migrated
}
}
expect(
writer.db
.prepare('PRAGMA table_info(dispatch_contexts)')
.all()
.map((row) => row.name)
.toSorted()
).toEqual(DISPATCH_CONTEXT_COLUMNS.toSorted())
const reader = new OrchestrationDb(path)
databases.push(reader)
return { writer, reader, run, entries }
}
function fullDispatchReads(calls: [string, ...unknown[]][]): string[] {
return calls
.map(([sql]) => sql)
.filter(
(sql) =>
/SELECT (?:\*|id, run_id, task_id, contract_version,)/.test(sql) &&
sql.includes('FROM dispatch_contexts')
)
}
describe('remaining dispatch metadata statement compilation', () => {
it.each(['fresh', 'v41'] as const)(
'reuses stale and direct-owner reads with exact rows, rebinding, plans and cross-connection freshness (%s)',
(version) => {
const {
writer,
reader,
run,
entries: [{ dispatch, expected }]
} = fixture(1, version)
vi.useFakeTimers()
vi.setSystemTime(new Date(NOW))
const compile = vi.spyOn(DatabaseSync.prototype, 'prepare')
for (let call = 0; call < 10; call += 1) {
expect(reader.getStaleDispatches('2026-10-02T11:00:00Z')).toEqual([expected])
expect(reader.getStaleDispatches('2026-10-02T09:00:00Z')).toEqual([])
expect(reader.findActiveDispatchForDirectMessageOwner(run.id, 'worker_0')).toEqual(expected)
expect(reader.findActiveDispatchForDirectMessageOwner(run.id, 'missing', PANE)).toEqual(
expected
)
expect(
reader.findActiveDispatchForDirectMessageOwner(
run.id,
'missing',
PANE.replace('tab_worker:', 'reminted:')
)
).toEqual(expected)
expect(
reader.findActiveDispatchForDirectMessageOwner(run.id, 'missing', 'bad:legacy')
).toBeUndefined()
}
const warnings: string[] = []
warnStaleDispatches(reader, (warning) => warnings.push(warning))
expect(warnings).toEqual([
`Warning: worker worker_0 on task ${dispatch.task_id} has not sent a heartbeat in ~10 min (dispatch ${dispatch.id})`
])
writer.db
.prepare(
'UPDATE dispatch_contexts SET last_heartbeat_at = ?, assignee_handle = ? WHERE id = ?'
)
.run(NOW, 'updated', dispatch.id)
expect(reader.getStaleDispatches('2026-10-02T11:00:00Z')).toEqual([])
expect(reader.findActiveDispatchForDirectMessageOwner(run.id, 'worker_0')).toBeUndefined()
expect(reader.findActiveDispatchForDirectMessageOwner(run.id, 'updated')).toEqual({
...expected,
last_heartbeat_at: NOW,
assignee_handle: 'updated'
})
const reads = fullDispatchReads(compile.mock.calls)
const staleSql = reads.find((sql) => sql.includes('julianday'))
compile.mockRestore()
if (!staleSql) {
throw new Error('Missing stale query')
}
const wildcardSql = staleSql.replace(/SELECT .*? FROM/, 'SELECT * FROM')
expect(writer.db.prepare(`EXPLAIN QUERY PLAN ${staleSql}`).all(NOW, NOW)).toEqual(
writer.db.prepare(`EXPLAIN QUERY PLAN ${wildcardSql}`).all(NOW, NOW)
)
expect(reads).toHaveLength(3)
}
)
it.each(['fresh', 'v41'] as const)(
'reuses session creator reads without changing self-created parent exclusion or fresh depth (%s)',
(version) => {
const {
writer,
reader,
entries: [{ dispatch }]
} = fixture(1, version)
const sessionId = '00000000-0000-4000-8000-000000000001'
if (!isOrcaSessionId(sessionId)) {
throw new Error('Invalid session fixture')
}
writer.db
.prepare('UPDATE dispatch_contexts SET assignee_orca_session_id = ? WHERE id = ?')
.run(sessionId, dispatch.id)
const compile = vi.spyOn(DatabaseSync.prototype, 'prepare')
for (let call = 0; call < 10; call += 1) {
expect(reader.resolveCreatorDepth({ kind: 'session', orcaSessionId: sessionId })).toBe(2)
expect(reader.resolveCreatorDispatchId({ kind: 'session', orcaSessionId: sessionId })).toBe(
dispatch.id
)
}
writer.db.prepare('UPDATE dispatch_contexts SET depth = ? WHERE id = ?').run(3, dispatch.id)
expect(reader.resolveCreatorDepth({ kind: 'session', orcaSessionId: sessionId })).toBe(3)
writer.db
.prepare('UPDATE dispatch_contexts SET creator_handle = assignee_handle WHERE id = ?')
.run(dispatch.id)
expect(reader.resolveCreatorDepth({ kind: 'session', orcaSessionId: sessionId })).toBe(0)
expect(
reader.resolveCreatorDispatchId({ kind: 'session', orcaSessionId: sessionId })
).toBeNull()
expect(fullDispatchReads(compile.mock.calls)).toHaveLength(1)
}
)
it.each(['fresh', 'v41'] as const)(
'reuses gate ownership and settlement reads while preserving completed rows and blocked tasks (%s)',
(version) => {
vi.useFakeTimers()
vi.setSystemTime(new Date(NOW))
const { reader, entries } = fixture(10, version)
const compile = vi.spyOn(DatabaseSync.prototype, 'prepare')
for (const { task, dispatch, expected } of entries) {
const gate = reader.createGate({ taskId: task.id, question: 'Continue?' })
expect(gate.task_id).toBe(task.id)
expect(gate.status).toBe('pending')
expect(reader.getDispatchContextById(dispatch.id)).toEqual({
...expected,
status: 'completed',
completed_at: NOW,
capability_revoked_at: NOW
})
expect(reader.getTask(task.id)?.status).toBe('blocked')
}
expect(fullDispatchReads(compile.mock.calls)).toHaveLength(12)
}
)
it.each(['fresh', 'v41'] as const)(
'reuses the private failure snapshot while preserving returned committed row order (%s)',
(version) => {
vi.useFakeTimers()
vi.setSystemTime(new Date(NOW))
const { reader, entries } = fixture(10, version)
const compile = vi.spyOn(DatabaseSync.prototype, 'prepare')
for (const { task, dispatch, expected } of entries) {
const expectedFailure = {
...expected,
status: 'failed',
failure_count: 1,
last_failure: 'failed send',
completed_at: NOW,
capability_revoked_at: NOW
}
const failure = reader.failDispatch(dispatch.id, 'failed send')
expect(failure).toEqual(expectedFailure)
expect(JSON.stringify(failure)).toBe(JSON.stringify(expectedFailure))
expect(reader.getTask(task.id)?.status).toBe('ready')
}
const reads = fullDispatchReads(compile.mock.calls)
compile.mockRestore()
const first = entries[0]
if (!first) {
throw new Error('Missing failure fixture')
}
const before = reader.db
.prepare('SELECT * FROM dispatch_contexts WHERE id = ?')
.get(first.dispatch.id)
const early = reader.failDispatch(first.dispatch.id, 'later')
expect(JSON.stringify(early)).toBe(JSON.stringify(before))
expect(early).not.toBe(reader.failDispatch(first.dispatch.id, 'later'))
expect(reads).toHaveLength(21)
}
)
it.each(['fresh', 'v41'] as const)(
'reuses legacy dispatch identity scans with full rows and immediately fences settled rows (%s)',
(version) => {
const {
writer,
reader,
run,
entries: [{ dispatch, expected }]
} = fixture(1, version)
writer.db
.prepare('UPDATE dispatch_contexts SET contract_version = ? WHERE id = ?')
.run(LEGACY_CONTRACT_VERSION, dispatch.id)
const compile = vi.spyOn(DatabaseSync.prototype, 'prepare')
for (let call = 0; call < 10; call += 1) {
expect(
reader.resolveLegacyWorkerCandidate({ runId: run.id, terminalHandle: 'worker_0' })
).toEqual({ dispatch: { ...expected, contract_version: LEGACY_CONTRACT_VERSION } })
expect(
reader.resolveLegacyWorkerCandidate({ runId: run.id, terminalHandle: 'missing' })
).toBeUndefined()
}
writer.db
.prepare('UPDATE dispatch_contexts SET status = ? WHERE id = ?')
.run('completed', dispatch.id)
expect(
reader.resolveLegacyWorkerCandidate({ runId: run.id, terminalHandle: 'worker_0' })
).toBeUndefined()
expect(fullDispatchReads(compile.mock.calls)).toHaveLength(1)
}
)
})