mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: recover code queue stale active tasks

Browse Source
This commit is contained in:
Codex
2026-05-17 14:58:29 +00:00
parent 84db3532c9
commit 08a4cf005e
2 changed files with 26 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/microservices/code-queue/src/index.ts
+8 -12
View File
@@ -1549,15 +1549,11 @@ async function deleteTaskFromDatabase(taskId: string): Promise<void> {
async function claimTaskInDatabase(task: QueueTask, expectedQueueId: string): Promise<boolean> { async function claimTaskInDatabase(task: QueueTask, expectedQueueId: string): Promise<boolean> {
if (!databaseReady) return true; if (!databaseReady) return true;
const claimed = await withTimeout( const timeout = `${config.databaseClaimTimeoutMs}ms`;
sql.begin(async (client) => { const claimed = await sql.begin(async (client) => {
const timeout = `${config.databaseClaimTimeoutMs}ms`; await client`SELECT set_config('statement_timeout', ${timeout}, true), set_config('lock_timeout', ${timeout}, true)`;
await client`SELECT set_config('statement_timeout', ${timeout}, true), set_config('lock_timeout', ${timeout}, true)`; return await upsertTaskToDatabase(client, task, { claimQueueId: expectedQueueId });
return await upsertTaskToDatabase(client, task, { claimQueueId: expectedQueueId }); });
}),
config.databaseClaimTimeoutMs,
`database claim timed out after ${config.databaseClaimTimeoutMs}ms`,
);
if (claimed) return true; if (claimed) return true;
const databaseTask = await loadTaskFromDatabase(task.id); const databaseTask = await loadTaskFromDatabase(task.id);
if (databaseTask !== null) reconcileHotTaskFromDatabase(databaseTask); if (databaseTask !== null) reconcileHotTaskFromDatabase(databaseTask);
@@ -2799,6 +2795,7 @@ configureSelfTests({
removeActiveRunSlotWaiter, removeActiveRunSlotWaiter,
resolveReasoningEffort, resolveReasoningEffort,
runDatabaseClaimMoveSelfTest, runDatabaseClaimMoveSelfTest,
taskIsRecoverableOrphanedActive,
tasks: () => state.tasks, tasks: () => state.tasks,
updateProcessingFlag, updateProcessingFlag,
}); });
@@ -3338,10 +3335,8 @@ function taskHasLocalExecutionRecoveryClaim(task: QueueTask): boolean {
} }
function activeTaskHasLocalRunSlotOrWaiter(task: QueueTask): boolean { function activeTaskHasLocalRunSlotOrWaiter(task: QueueTask): boolean {
const queueId = queueIdOf(task);
return activeRunForTask(task) !== null return activeRunForTask(task) !== null
|| activeRunSlotReservations.has(queueId) || activeRunSlotWaiters.some((waiter) => waiter.taskId === task.id);
|| activeRunSlotWaiters.some((waiter) => waiter.taskId === task.id || waiter.queueId === queueId);
} }
function orphanedActiveTaskAgeMs(task: QueueTask): number { function orphanedActiveTaskAgeMs(task: QueueTask): number {
@@ -3497,6 +3492,7 @@ async function runTask(task: QueueTask): Promise<void> {
} }
if (!claimed) { if (!claimed) {
releaseRunSlot(); releaseRunSlot();
await recoverOrphanedActiveTasks("Database claim found stale active task without local run", "database/claim-recovery");
return; return;
} }
publishTaskOaEvent(task, "claim"); publishTaskOaEvent(task, "claim");
src/components/microservices/code-queue/src/self-tests.ts
+18 -10
View File
@@ -28,6 +28,7 @@ export interface SelfTestsContext {
removeActiveRunSlotWaiter: (waiter: ActiveRunSlotWaiter) => void; removeActiveRunSlotWaiter: (waiter: ActiveRunSlotWaiter) => void;
resolveReasoningEffort: (model: string, explicit?: string | null) => string | null; resolveReasoningEffort: (model: string, explicit?: string | null) => string | null;
runDatabaseClaimMoveSelfTest?: () => Promise<JsonValue | null>; runDatabaseClaimMoveSelfTest?: () => Promise<JsonValue | null>;
taskIsRecoverableOrphanedActive: (task: QueueTask, staleMs?: number) => boolean;
tasks: () => QueueTask[]; tasks: () => QueueTask[];
updateProcessingFlag: () => void; updateProcessingFlag: () => void;
} }
@@ -296,16 +297,23 @@ function runQueueOrderingSelfTest(): JsonValue {
ctx().removeActiveRunSlotWaiter(secondWaiter); ctx().removeActiveRunSlotWaiter(secondWaiter);
} }
} }
try { {
ctx().tasks().push(orphanRunning, queuedBehindOrphan); const reservationMarker = "queue_orphan_recovery";
const recovered = ctx().queueActiveTasksForRestartRetry("orphaned active task self-test", "self-test", { taskIds: new Set([orphanRunning.id]) }); const hadSameQueueReservation = ctx().activeRunSlotReservations.has(reservationMarker);
assertReferenceTest(recovered >= 1, "orphaned running task should be recovered for retry"); try {
assertReferenceTest(orphanRunning.status === "retry_wait", "orphaned running task should become retry_wait"); ctx().tasks().push(orphanRunning, queuedBehindOrphan);
assertReferenceTest(orphanRunning.activeTurnId === null, "orphaned running task should clear activeTurnId"); ctx().activeRunSlotReservations.add(reservationMarker);
assertReferenceTest(orphanRunning.lastError === "orphaned active task self-test", "orphaned running task should record recovery reason"); assertReferenceTest(ctx().taskIsRecoverableOrphanedActive(orphanRunning, 0), "same-queue run slot reservation must not hide an orphaned task");
assertReferenceTest(ctx().nextRunnableTaskFrom("queue_orphan_recovery")?.id === orphanRunning.id, "recovered orphan should be the next runnable queue head"); const recovered = ctx().queueActiveTasksForRestartRetry("orphaned active task self-test", "self-test", { taskIds: new Set([orphanRunning.id]) });
} finally { assertReferenceTest(recovered >= 1, "orphaned running task should be recovered for retry");
ctx().tasks().splice(0, ctx().tasks().length, ...beforeTasks); assertReferenceTest(orphanRunning.status === "retry_wait", "orphaned running task should become retry_wait");
assertReferenceTest(orphanRunning.activeTurnId === null, "orphaned running task should clear activeTurnId");
assertReferenceTest(orphanRunning.lastError === "orphaned active task self-test", "orphaned running task should record recovery reason");
assertReferenceTest(ctx().nextRunnableTaskFrom("queue_orphan_recovery")?.id === orphanRunning.id, "recovered orphan should be the next runnable queue head");
} finally {
if (!hadSameQueueReservation) ctx().activeRunSlotReservations.delete(reservationMarker);
ctx().tasks().splice(0, ctx().tasks().length, ...beforeTasks);
}
} }
return { return {