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

fix(code-queue): harden active state recovery

Browse Source
This commit is contained in:
Codex
2026-05-17 11:12:41 +00:00
parent e8d165b1a0
commit 97a647bc8d
4 changed files with 125 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/microservices/code-queue-mgr/src/index.ts
+7 -3
View File
@@ -698,6 +698,7 @@ function normalizeTask(value: unknown): QueueTask {
nextPrompt: typeof record.nextPrompt === "string" ? record.nextPrompt : null,
nextMode: record.nextMode === "initial" || record.nextMode === "retry" ? record.nextMode : null,
};
if (task.status !== "running" && task.status !== "judging") task.activeTurnId = null;
task.stepCount = numberField(record.stepCount ?? record.llmStepCount, task.output.length);
task.llmStepCount = task.stepCount;
task.outputMaxSeq = Math.max(outputMaxSeq(task), numberField(record.outputMaxSeq, 0));
@@ -1254,7 +1255,7 @@ async function queueSummary(tasks?: QueueTask[], queueRecords?: QueueRecord[]):
memo[task.status] = (memo[task.status] ?? 0) + 1;
return memo;
}, {});
const orphanedActiveTaskIds = taskRows.filter((task) => task.status === "running" || task.status === "judging").map((task) => task.id).sort();
const databaseActiveTaskIds = taskRows.filter((task) => task.status === "running" || task.status === "judging").map((task) => task.id).sort();
const queueList = Array.from(summaries.values()).sort((left, right) => String(left.id).localeCompare(String(right.id)));
return {
total: taskRows.length,
@@ -1267,8 +1268,11 @@ async function queueSummary(tasks?: QueueTask[], queueRecords?: QueueRecord[]):
activeRunSlotWaiters: [],
activeTaskIds: [],
activeTaskId: null,
orphanedActiveTaskIds,
orphanedActiveTaskCount: orphanedActiveTaskIds.length,
databaseActiveTaskIds,
databaseActiveTaskCount: databaseActiveTaskIds.length,
executionStateSource: "postgres-control-plane",
orphanedActiveTaskIds: [],
orphanedActiveTaskCount: 0,
processing: false,
counts,
unreadTerminal: taskRows.reduce((total, task) => total + (terminalTaskUnread(task) ? 1 : 0), 0),
src/components/microservices/code-queue/src/index.ts
+115 -5
View File
@@ -10,6 +10,7 @@ import type {
CgroupMemoryUsage,
CodexEventSummary,
CodexRunResult,
JudgeFailureDetails,
JudgeProbeCase,
JudgeResult,
JsonValue,
@@ -388,6 +389,7 @@ function readConfig(): RuntimeConfig {
databaseUrl: envRequiredString("DATABASE_URL"),
databasePoolMax: Math.max(1, Math.min(8, envNumber("CODE_QUEUE_DATABASE_POOL_MAX", envNumber("DATABASE_POOL_MAX", 2)))),
databaseFlushIntervalMs: Math.max(100, Math.min(10_000, envNumber("CODE_QUEUE_DATABASE_FLUSH_INTERVAL_MS", 1000))),
databaseClaimTimeoutMs: Math.max(1000, Math.min(60_000, envNumber("CODE_QUEUE_DATABASE_CLAIM_TIMEOUT_MS", 15_000))),
oaEventFlowBaseUrl: envString("OA_EVENT_FLOW_BASE_URL", "http://oa-event-flow:4255").replace(/\/+$/u, ""),
notifyClaudeQqEnabled: envBool("CODE_QUEUE_NOTIFY_CLAUDEQQ_ENABLED", false),
notifyClaudeQqBaseUrl: envString("CODE_QUEUE_NOTIFY_CLAUDEQQ_BASE_URL", "http://claudeqq.unidesk.svc.cluster.local:3290").replace(/\/+$/u, ""),
@@ -928,6 +930,10 @@ function normalizeTaskMetric(value: unknown): number | null {
return Number.isFinite(parsed) && parsed >= 0 ? Math.floor(parsed) : null;
}
function taskCanHaveActiveTurn(status: TaskStatus): boolean {
return status === "running" || status === "judging";
}
function normalizeTask(task: QueueTask): QueueTask {
task.queueId = safeQueueId(task.queueId);
task.queueEnteredAt = taskQueueEnteredAt(task);
@@ -940,6 +946,7 @@ function normalizeTask(task: QueueTask): QueueTask {
task.readAt = null;
}
task.activeTurnId ??= null;
if (!taskCanHaveActiveTurn(task.status)) task.activeTurnId = null;
task.providerId = normalizeTaskProviderId(task.providerId);
task.model ||= config.defaultModel;
task.executionMode = normalizeCodeExecutionMode(task.executionMode);
@@ -1525,7 +1532,15 @@ async function deleteTaskFromDatabase(taskId: string): Promise<void> {
async function claimTaskInDatabase(task: QueueTask, expectedQueueId: string): Promise<boolean> {
if (!databaseReady) return true;
const claimed = await sql.begin(async (client) => await upsertTaskToDatabase(client, task, { claimQueueId: expectedQueueId }));
const claimed = await withTimeout(
sql.begin(async (client) => {
const timeout = `${config.databaseClaimTimeoutMs}ms`;
await client`SELECT set_config('statement_timeout', ${timeout}, true), set_config('lock_timeout', ${timeout}, true)`;
return await upsertTaskToDatabase(client, task, { claimQueueId: expectedQueueId });
}),
config.databaseClaimTimeoutMs,
`database claim timed out after ${config.databaseClaimTimeoutMs}ms`,
);
if (claimed) return true;
const databaseTask = await loadTaskFromDatabase(task.id);
if (databaseTask !== null) reconcileHotTaskFromDatabase(databaseTask);
@@ -2068,6 +2083,14 @@ async function initDatabasePersistence(): Promise<void> {
AND COALESCE(task_json->>'readAt', '') <> ''
AND (task_json->>'readAt') ~ '^\\d{4}-\\d{2}-\\d{2}T'
`;
await sql`
UPDATE unidesk_code_queue_tasks
SET
active_turn_id = NULL,
task_json = jsonb_set(task_json, '{activeTurnId}', 'null'::jsonb, true)
WHERE status NOT IN ('running', 'judging')
AND active_turn_id IS NOT NULL
`;
await sql`ALTER TABLE unidesk_code_queue_queues ADD COLUMN IF NOT EXISTS name TEXT NOT NULL DEFAULT ''`;
await sql`ALTER TABLE unidesk_code_queue_workdirs ADD COLUMN IF NOT EXISTS execution_mode TEXT NOT NULL DEFAULT 'default'`;
}
@@ -3183,6 +3206,63 @@ function retryBackoffMs(completedAttempts: number): number {
return Math.min(retryBackoffMaxMs, retryBackoffBaseMs * (2 ** exponent));
}
async function withTimeout<T>(promise: Promise<T>, timeoutMs: number, message: string): Promise<T> {
let timer: ReturnType<typeof setTimeout> | null = null;
const timeout = new Promise<never>((_, reject) => {
timer = setTimeout(() => reject(new Error(message)), timeoutMs);
});
try {
return await Promise.race([promise, timeout]);
} finally {
if (timer !== null) clearTimeout(timer);
}
}
function fallbackJudgeFromWatchdog(task: QueueTask, result: CodexRunResult, error: unknown, startedAt: number): JudgeResult {
const durationMs = Date.now() - startedAt;
const message = error instanceof Error ? error.message : String(error);
const detail: JudgeFailureDetails = {
provider: "minimax",
stage: "unknown",
model: config.minimaxModel,
apiBase: config.minimaxApiBase,
occurredAt: nowIso(),
durationMs,
timeoutMs: config.judgeTimeoutMs,
timedOut: durationMs >= config.judgeTimeoutMs,
errorName: error instanceof Error ? error.name : "Error",
errorMessage: message,
};
logger("warn", "judge_watchdog_fallback", {
taskId: task.id,
attempt: task.currentAttempt,
durationMs,
timeoutMs: config.judgeTimeoutMs,
error: safePreview(message, 1000),
});
return {
decision: result.terminalStatus === "completed" ? "complete" : "retry",
confidence: 0.3,
reason: `MiniMax judge watchdog fallback: ${message}`,
source: "fallback",
failureDetails: detail,
raw: { minimaxFailure: detail as unknown as JsonValue },
};
}
async function judgeTaskWithWatchdog(task: QueueTask, result: CodexRunResult): Promise<JudgeResult> {
const startedAt = Date.now();
try {
return await withTimeout(
judgeTask(task, result),
config.judgeTimeoutMs + 5_000,
`judge task timed out after ${config.judgeTimeoutMs + 5_000}ms`,
);
} catch (error) {
return fallbackJudgeFromWatchdog(task, result, error, startedAt);
}
}
async function sleepForRetryBackoff(task: QueueTask, delayMs: number): Promise<void> {
let remaining = delayMs;
while (remaining > 0 && !task.cancelRequested && !shutdownRequested) {
@@ -3198,6 +3278,18 @@ function taskHasLocalExecutionRecoveryClaim(task: QueueTask): boolean {
return activeRunSlotReservations.has(queueIdOf(task));
}
function clearInactiveActiveTurnIds(): number {
let changed = 0;
for (const task of state.tasks) {
if (taskCanHaveActiveTurn(task.status) || task.activeTurnId === null) continue;
task.activeTurnId = null;
task.updatedAt = nowIso();
markTaskDirty(task.id);
changed += 1;
}
return changed;
}
function queueActiveTasksForRestartRetry(reason: string, method: string, options: { onlyActiveRuns?: boolean } = {}): number {
if (!config.schedulerEnabled || !serviceRoleAllowsScheduler(config.serviceRole)) return 0;
let recovered = 0;
@@ -3287,7 +3379,24 @@ async function runTask(task: QueueTask): Promise<void> {
task.readAt = null;
task.finishedAt = null;
task.updatedAt = startedAt;
if (!await claimTaskInDatabase(task, claimQueueId)) {
let claimed = false;
try {
claimed = await claimTaskInDatabase(task, claimQueueId);
} catch (error) {
releaseRunSlot();
task.status = "retry_wait";
task.activeTurnId = null;
task.finishedAt = null;
task.readAt = null;
task.updatedAt = nowIso();
task.lastError = `Database claim failed before agent start: ${error instanceof Error ? error.message : String(error)}`;
appendOutput(task, "error", `${task.lastError}\n`, "database/claim");
persistTaskState(task);
logger("error", "task_claim_failed", { taskId: task.id, queueId: claimQueueId, error: errorToJson(error) });
await sleepForRetryBackoff(task, retryBackoffMs(Math.max(1, task.attempts.length)));
return;
}
if (!claimed) {
releaseRunSlot();
return;
}
@@ -3310,7 +3419,7 @@ async function runTask(task: QueueTask): Promise<void> {
persistTaskState(task);
if (task.cancelRequested) break;
const judge = await judgeTask(task, result);
const judge = await judgeTaskWithWatchdog(task, result);
task.lastJudge = judge;
const judgeOutput = appendOutput(task, judge.decision === "complete" ? "system" : judge.decision === "fail" ? "error" : "system", `judge=${judge.decision} confidence=${judge.confidence.toFixed(2)} source=${judge.source}: ${judge.reason}${judgeFailureDetailsForOutput(judge)}\n`, "judge");
const latestAttempt = task.attempts.at(-1);
@@ -4938,12 +5047,13 @@ async function startDatabaseBackedRuntime(): Promise<void> {
logger("info", "service_started", { port: config.port, instanceId: config.instanceId, role: config.serviceRole, schedulerEnabled: config.schedulerEnabled, schedulerPollIntervalMs: config.schedulerPollIntervalMs, workdir: config.defaultWorkdir, defaultModel: config.defaultModel, judgeConfigured: config.minimaxApiKey.length > 0, storage: "postgres" });
const devReady = collectDevReady() as Record<string, JsonValue>;
logger(devReady.ok === true ? "info" : "warn", "dev_ready_check", devReady);
const inactiveActiveTurnsCleared = clearInactiveActiveTurnIds();
const startupRecovered = config.schedulerEnabled ? queueActiveTasksForRestartRetry("Service restarted while task was active", "startup") : 0;
const startupRecoveryDirtyTaskCount = dirtyDatabaseTaskIds.size;
if (startupRecovered > 0 || startupRecoveryDirtyTaskCount > 0) {
logger("warn", "startup_requeued_active_tasks", { recovered: startupRecovered, dirtyTaskCount: startupRecoveryDirtyTaskCount });
logger("warn", "startup_requeued_active_tasks", { recovered: startupRecovered, inactiveActiveTurnsCleared, dirtyTaskCount: startupRecoveryDirtyTaskCount });
await flushDirtyTasksToDatabase(true);
logger("info", "startup_requeued_active_tasks_flushed", { recovered: startupRecovered, dirtyTaskCount: startupRecoveryDirtyTaskCount });
logger("info", "startup_requeued_active_tasks_flushed", { recovered: startupRecovered, inactiveActiveTurnsCleared, dirtyTaskCount: startupRecoveryDirtyTaskCount });
}
persistState();
serviceReady = true;
src/components/microservices/code-queue/src/types.ts
+1
View File
@@ -75,6 +75,7 @@ export interface RuntimeConfig {
databaseUrl: string;
databasePoolMax: number;
databaseFlushIntervalMs: number;
databaseClaimTimeoutMs: number;
oaEventFlowBaseUrl: string;
notifyClaudeQqEnabled: boolean;
notifyClaudeQqBaseUrl: string;