fix(orchestration): fence stale dispatch failures

2026-10-23 00:03:52 +00:00 · 2026-08-16 18:19:21 -04:00
parent ab7fc1bad4
commit 694b026438
3 changed files with 295 additions and 47 deletions
@@ -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
+}
@@ -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 ──
@@ -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({