mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 00:03:52 +00:00
fix(orchestration): release final active context
This commit is contained in:
@@ -12,14 +12,13 @@ export type ContextOnlyDispatchReleaseResult = {
|
||||
export function releaseContextOnlyDispatch(
|
||||
db: Database.Database,
|
||||
dispatch: DispatchContextRow,
|
||||
latestDispatchId: string | undefined,
|
||||
requestedState: 'abandoned' | 'stopped'
|
||||
): ContextOnlyDispatchReleaseResult {
|
||||
if (dispatch.status !== 'pending' && dispatch.status !== 'dispatched') {
|
||||
return {
|
||||
state: persistedReleaseState(dispatch),
|
||||
alreadySettled: true,
|
||||
releasedCurrentTask: latestDispatchId === dispatch.id
|
||||
releasedCurrentTask: false
|
||||
}
|
||||
}
|
||||
|
||||
@@ -36,10 +35,12 @@ export function releaseContextOnlyDispatch(
|
||||
WHERE task_id = ? AND status IN ('pending', 'dispatched') LIMIT 1`
|
||||
)
|
||||
.get(dispatch.task_id)
|
||||
const releasedCurrentTask = latestDispatchId === dispatch.id && !remaining
|
||||
if (releasedCurrentTask) {
|
||||
db.prepare("UPDATE tasks SET status = 'blocked' WHERE id = ?").run(dispatch.task_id)
|
||||
}
|
||||
const releasedCurrentTask = Boolean(
|
||||
!remaining &&
|
||||
db
|
||||
.prepare("UPDATE tasks SET status = 'blocked' WHERE id = ? AND status = 'dispatched'")
|
||||
.run(dispatch.task_id).changes
|
||||
)
|
||||
return { state: requestedState, alreadySettled: false, releasedCurrentTask }
|
||||
}
|
||||
|
||||
|
||||
@@ -191,6 +191,39 @@ describe('Task/Dispatch lifecycle guards', () => {
|
||||
}
|
||||
)
|
||||
|
||||
it.each(['stop', 'abandon'] as const)(
|
||||
'%s releases the last context-only sibling after a newer worker start fails',
|
||||
(operation) => {
|
||||
const database = createDatabase()
|
||||
const task = database.createTask({ spec: `${operation} historical sibling` })
|
||||
const contextOnly = database.createDispatchContext(task.id, `term_${operation}`)
|
||||
sqliteFor(database).prepare("UPDATE tasks SET status = 'ready' WHERE id = ?").run(task.id)
|
||||
const failed = database.createStartingWorkerDispatch({ taskId: task.id, startOptions: {} })
|
||||
|
||||
database.failWorkerStart(failed.dispatch.id, 'start_failed', 'worker failed to start')
|
||||
expect(database.getTask(task.id)?.status).toBe('dispatched')
|
||||
|
||||
const released =
|
||||
operation === 'stop'
|
||||
? database.beginWorkerStop(contextOnly.id)
|
||||
: database.abandonWorkerDispatch(contextOnly.id)
|
||||
expect(released).toMatchObject({
|
||||
disposition: 'context_only',
|
||||
alreadySettled: false,
|
||||
releasedCurrentTask: true
|
||||
})
|
||||
expect(database.getTask(task.id)?.status).toBe('blocked')
|
||||
expect(database.getDispatchContextById(contextOnly.id)?.status).toBe('failed')
|
||||
expect(database.getActiveDispatchForTerminal(`term_${operation}`)).toBeUndefined()
|
||||
expect(() =>
|
||||
database.createDispatchContext(
|
||||
database.createTask({ spec: `${operation} later work` }).id,
|
||||
`term_${operation}`
|
||||
)
|
||||
).not.toThrow()
|
||||
}
|
||||
)
|
||||
|
||||
it('rejects gate creation while a supervised worker remains active', () => {
|
||||
const database = createDatabase()
|
||||
const task = database.createTask({ spec: 'worker gate guard' })
|
||||
|
||||
@@ -6383,12 +6383,7 @@ export class OrchestrationDb {
|
||||
throw new OrchestrationError('dispatch_not_found', `Dispatch ${dispatchId} was not found.`)
|
||||
}
|
||||
if (!worker) {
|
||||
const released = releaseContextOnlyDispatch(
|
||||
this.db,
|
||||
dispatch,
|
||||
this.getDispatchContext(dispatch.task_id)?.id,
|
||||
'stopped'
|
||||
)
|
||||
const released = releaseContextOnlyDispatch(this.db, dispatch, 'stopped')
|
||||
if (!released.alreadySettled) {
|
||||
this.closeQuestionsForDispatch(dispatchId)
|
||||
}
|
||||
@@ -6560,12 +6555,7 @@ export class OrchestrationDb {
|
||||
throw new OrchestrationError('dispatch_not_found', `Dispatch ${dispatchId} was not found.`)
|
||||
}
|
||||
if (!worker) {
|
||||
const released = releaseContextOnlyDispatch(
|
||||
this.db,
|
||||
dispatch,
|
||||
this.getDispatchContext(dispatch.task_id)?.id,
|
||||
'abandoned'
|
||||
)
|
||||
const released = releaseContextOnlyDispatch(this.db, dispatch, 'abandoned')
|
||||
if (!released.alreadySettled) {
|
||||
this.closeQuestionsForDispatch(dispatchId)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user