diff --git a/src/main/runtime/orchestration/db-task-dispatch-races.test.ts b/src/main/runtime/orchestration/db-task-dispatch-races.test.ts new file mode 100644 index 00000000000..613e3a3db34 --- /dev/null +++ b/src/main/runtime/orchestration/db-task-dispatch-races.test.ts @@ -0,0 +1,196 @@ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it, vi } from 'vitest' +import type Database from '../../sqlite/sync-database' +import { OrchestrationDb } from './db' + +type DatabaseHarness = { + db: OrchestrationDb + dir: string + path: string +} + +const harnesses: DatabaseHarness[] = [] + +afterEach(() => { + vi.restoreAllMocks() + const closed = harnesses.splice(0) + for (const harness of closed) { + harness.db.close() + } + for (const dir of new Set(closed.map((harness) => harness.dir))) { + rmSync(dir, { recursive: true, force: true }) + } +}) + +describe('Task/Dispatch concurrency', () => { + it('rolls back Dispatch failure when Task requeue fails', () => { + const { db } = createDatabase() + const task = db.createTask({ spec: 'atomic retry failure' }) + const dispatch = db.createDispatchContext(task.id, 'term_worker') + sqliteFor(db).exec(` + CREATE TRIGGER reject_task_requeue + BEFORE UPDATE OF status ON tasks + WHEN OLD.id = '${task.id}' + BEGIN + SELECT RAISE(ABORT, 'forced task requeue failure'); + END; + `) + + expect(() => db.failDispatch(dispatch.id, 'must roll back')).toThrow( + 'forced task requeue failure' + ) + expect(db.getTask(task.id)?.status).toBe('dispatched') + expect(db.getDispatchContextById(dispatch.id)).toMatchObject({ + status: 'dispatched', + failure_count: 0, + last_failure: null, + completed_at: null, + capability_revoked_at: null + }) + }) + + it('does not let stale failure overwrite a completed worker report', () => { + const first = createDatabase() + const concurrent = createDatabase(first.path) + const task = first.db.createTask({ spec: 'worker completion wins' }) + const started = first.db.createStartingWorkerDispatch({ taskId: task.id, startOptions: {} }) + const capability = first.db.prepareStartingWorkerAuthority({ + dispatchId: started.dispatch.id, + handle: 'term_worker', + paneKey: 'tab_worker:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa', + processIncarnation: 'worker:1', + worktreeId: 'repo::worker', + effects: [], + setupState: 'not_applicable', + terminalOwnership: 'created' + }) + first.db.markWorkerDispatchReady(started.dispatch.id) + const sqlite = sqliteFor(first.db) + const prepare = sqlite.prepare.bind(sqlite) + let completionWon = false + vi.spyOn(sqlite, 'prepare').mockImplementation((sql) => { + if ( + !completionWon && + sql.includes('UPDATE dispatch_contexts') && + sql.includes('failure_count') + ) { + completionWon = true + expect( + concurrent.db.settleWorkerReport({ + taskId: task.id, + dispatchId: started.dispatch.id, + outcome: 'succeeded', + result: 'completed concurrently' + }) + ).toMatchObject({ action: 'settled', duplicate: false }) + } + return prepare(sql) + }) + + expect(first.db.failDispatch(started.dispatch.id, 'stale failure')).toMatchObject({ + status: 'completed', + failure_count: 0 + }) + expect(completionWon).toBe(true) + expect(first.db.getTask(task.id)).toMatchObject({ + status: 'completed', + result: 'completed concurrently' + }) + expect(first.db.getWorkerDispatch(started.dispatch.id)?.state).toBe('succeeded') + expect( + first.db.verifyDispatchCapability({ + dispatchId: started.dispatch.id, + capability, + paneKey: 'tab_worker:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa', + processIncarnation: 'worker:1' + }) + ).toMatchObject({ valid: false }) + }) + + it('serializes reminted-pane worker authority claims', () => { + const first = createDatabase() + const concurrent = createDatabase(first.path) + const losingTask = first.db.createTask({ spec: 'losing worker' }) + const winningTask = first.db.createTask({ spec: 'winning worker' }) + const loser = first.db.createStartingWorkerDispatch({ + taskId: losingTask.id, + startOptions: {} + }) + const winner = concurrent.db.createStartingWorkerDispatch({ + taskId: winningTask.id, + startOptions: {} + }) + const sqlite = sqliteFor(first.db) + const exec = sqlite.exec.bind(sqlite) + let winningCapability: string | undefined + vi.spyOn(sqlite, 'exec').mockImplementation((sql) => { + if (!winningCapability && sql === 'BEGIN IMMEDIATE') { + winningCapability = concurrent.db.prepareStartingWorkerAuthority({ + dispatchId: winner.dispatch.id, + handle: 'term_reminted', + paneKey: 'tab_new:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb', + processIncarnation: 'winner:1', + worktreeId: 'repo::winner', + effects: [], + setupState: 'not_applicable', + terminalOwnership: 'created' + }) + } + return exec(sql) + }) + + expect(() => + first.db.prepareStartingWorkerAuthority({ + dispatchId: loser.dispatch.id, + handle: 'term_original', + paneKey: 'tab_old:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb', + processIncarnation: 'loser:1', + worktreeId: 'repo::loser', + effects: [], + setupState: 'not_applicable', + terminalOwnership: 'created' + }) + ).toThrow(`already has an active dispatch (${winner.dispatch.id} for task ${winningTask.id})`) + expect(winningCapability).toBeDefined() + expect(first.db.getDispatchContextById(loser.dispatch.id)).toMatchObject({ + assignee_handle: null, + capability_hash: null + }) + expect(first.db.getWorkerDispatch(loser.dispatch.id)).toMatchObject({ + stage: 'accepted', + agent_terminal_handle: null + }) + expect(first.db.getWorkerTerminalResourceByOwner(loser.dispatch.id)).toBeUndefined() + expect(first.db.getDispatchContextById(winner.dispatch.id)).toMatchObject({ + assignee_handle: 'term_reminted', + capability_hash: expect.any(String) + }) + expect(first.db.getWorkerTerminalResourceByOwner(winner.dispatch.id)).toMatchObject({ + terminal_handle: 'term_reminted', + ownership_state: 'owned' + }) + expect( + sqlite + .prepare( + `SELECT COUNT(*) AS count FROM dispatch_contexts + WHERE status IN ('pending', 'dispatched') AND capability_hash IS NOT NULL` + ) + .get() + ).toEqual({ count: 1 }) + }) +}) + +function createDatabase(path?: string): DatabaseHarness { + const dir = path ? harnesses.find((harness) => harness.path === path)?.dir : undefined + const ownedDir = dir ?? mkdtempSync(join(tmpdir(), 'orca-task-dispatch-races-')) + const dbPath = path ?? join(ownedDir, 'orchestration.db') + const harness = { db: new OrchestrationDb(dbPath), dir: ownedDir, path: dbPath } + harnesses.push(harness) + return harness +} + +function sqliteFor(db: OrchestrationDb): Database.Database { + return (db as unknown as { db: Database.Database }).db +} diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index cb453324637..60d2653d513 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -7245,19 +7245,27 @@ export class OrchestrationDb { WHERE id = ? AND status = 'ready' AND NOT EXISTS ( SELECT 1 FROM dispatch_contexts active - WHERE active.status IN ('pending', 'dispatched') - AND ( - active.assignee_handle = ? - OR (? IS NOT NULL AND active.assignee_pane_key = ?) - OR ( - ? IS NOT NULL AND active.assignee_pane_key IS NOT NULL - AND instr(active.assignee_pane_key, ':') > 1 - AND substr( - active.assignee_pane_key, - instr(active.assignee_pane_key, ':') + 1 - ) = ? - ) - ) + WHERE active.assignee_handle = ? + AND active.status IN ('pending', 'dispatched') + ) + AND ( + ? IS NULL OR NOT EXISTS ( + SELECT 1 FROM dispatch_contexts active + WHERE active.assignee_pane_key = ? + AND active.status IN ('pending', 'dispatched') + ) + ) + AND ( + ? IS NULL OR NOT EXISTS ( + SELECT 1 FROM dispatch_contexts active + WHERE active.assignee_pane_key IS NOT NULL + AND active.status IN ('pending', 'dispatched') + AND instr(active.assignee_pane_key, ':') > 1 + AND substr( + active.assignee_pane_key, + instr(active.assignee_pane_key, ':') + 1 + ) = ? + ) )` ) .run( @@ -7659,41 +7667,44 @@ export class OrchestrationDb { } failDispatch(ctxId: string, error: string): DispatchContextRow | undefined { - const ctx = this.db.prepare('SELECT * FROM dispatch_contexts WHERE id = ?').get(ctxId) as - | DispatchContextRow - | undefined - if (!ctx) { - return undefined + this.db.exec('SAVEPOINT fail_dispatch') + try { + const updated = this.db + .prepare( + `UPDATE dispatch_contexts + SET status = CASE WHEN failure_count + 1 >= 3 THEN 'circuit_broken' ELSE 'failed' END, + failure_count = failure_count + 1, last_failure = ?, + completed_at = COALESCE(completed_at, datetime('now')), + capability_revoked_at = COALESCE(capability_revoked_at, datetime('now')) + WHERE id = ? AND status IN ('pending', 'dispatched')` + ) + .run(error, ctxId) + const ctx = this.db.prepare('SELECT * FROM dispatch_contexts WHERE id = ?').get(ctxId) as + | DispatchContextRow + | undefined + if (updated.changes !== 1 || !ctx) { + this.db.exec('RELEASE fail_dispatch') + return ctx + } + + // Why: back to 'ready' not 'pending' — 'pending' would strand it since promoteReadyTasks only runs when a dep completes. + const taskStatus: TaskStatus = ctx.status === 'circuit_broken' ? 'failed' : 'ready' + this.db + .prepare( + `UPDATE tasks SET status = ? + WHERE id = ? AND status = 'dispatched' AND NOT EXISTS ( + SELECT 1 FROM dispatch_contexts + WHERE task_id = tasks.id AND status IN ('pending', 'dispatched') + )` + ) + .run(taskStatus, ctx.task_id) + this.db.exec('RELEASE fail_dispatch') + return ctx + } catch (error) { + this.db.exec('ROLLBACK TO fail_dispatch') + this.db.exec('RELEASE fail_dispatch') + throw error } - - const newFailureCount = ctx.failure_count + 1 - const newStatus: DispatchStatus = newFailureCount >= 3 ? 'circuit_broken' : 'failed' - - this.db - .prepare( - `UPDATE dispatch_contexts - SET status = ?, failure_count = ?, last_failure = ?, - completed_at = COALESCE(completed_at, datetime('now')), - capability_revoked_at = COALESCE(capability_revoked_at, datetime('now')) - WHERE id = ?` - ) - .run(newStatus, newFailureCount, error, ctxId) - - // Why: back to 'ready' not 'pending' — 'pending' would strand it since promoteReadyTasks only runs when a dep completes. - const taskStatus: TaskStatus = newStatus === 'circuit_broken' ? 'failed' : 'ready' - this.db - .prepare( - `UPDATE tasks SET status = ? - WHERE id = ? AND NOT EXISTS ( - SELECT 1 FROM dispatch_contexts - WHERE task_id = tasks.id AND status IN ('pending', 'dispatched') - )` - ) - .run(taskStatus, ctx.task_id) - - return this.db.prepare('SELECT * FROM dispatch_contexts WHERE id = ?').get(ctxId) as - | DispatchContextRow - | undefined } // ── Decision Gates ── diff --git a/src/main/runtime/orchestration/orchestration-creator-authority-performance.test.ts b/src/main/runtime/orchestration/orchestration-creator-authority-performance.test.ts index 4a6067aa959..d08baf5b47f 100644 --- a/src/main/runtime/orchestration/orchestration-creator-authority-performance.test.ts +++ b/src/main/runtime/orchestration/orchestration-creator-authority-performance.test.ts @@ -50,6 +50,47 @@ describe('creator authority lookup performance', () => { expect(paneDetails).toContain('idx_dispatch_assignee_pane_leaf') }) + it('uses active-assignee indexes for Dispatch occupancy claims', () => { + db = new OrchestrationDb(':memory:') + const plan = sqliteFor(db) + .prepare( + `EXPLAIN QUERY PLAN + SELECT 1 FROM tasks + WHERE id = ? AND status = 'ready' + AND NOT EXISTS ( + SELECT 1 FROM dispatch_contexts active + WHERE active.assignee_handle = ? + AND active.status IN ('pending', 'dispatched') + ) + AND NOT EXISTS ( + SELECT 1 FROM dispatch_contexts active + WHERE active.assignee_pane_key = ? + AND active.status IN ('pending', 'dispatched') + ) + AND NOT EXISTS ( + SELECT 1 FROM dispatch_contexts active + WHERE active.assignee_pane_key IS NOT NULL + AND active.status IN ('pending', 'dispatched') + AND instr(active.assignee_pane_key, ':') > 1 + AND substr( + active.assignee_pane_key, + instr(active.assignee_pane_key, ':') + 1 + ) = ? + )` + ) + .all( + 'task_claimant', + 'term_worker', + 'tab_worker:33333333-3333-4333-8333-333333333333', + '33333333-3333-4333-8333-333333333333' + ) as { detail: string }[] + const details = plan.map((row) => row.detail).join(' | ') + + expect(details).toContain('idx_dispatch_active_assignee_handle') + expect(details).toContain('idx_dispatch_active_assignee_pane_key') + expect(details).toContain('idx_dispatch_assignee_pane_leaf') + }) + it('keeps 300 Task reads bounded with 50,000 retained Runs', () => { db = new OrchestrationDb(':memory:') const run = db.createRun({