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

fix: recover orphaned code queue runs

Browse Source
This commit is contained in:
Codex
2026-05-17 10:42:51 +00:00
parent 3a6b3ef33c
commit e5749c426f
8 changed files with 64 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/microservices/code-queue-mgr/src/index.ts
+21 -6
View File
@@ -14,6 +14,10 @@ interface RuntimeConfig {
port: number;
databaseUrl: string;
logFile: string;
deployServiceId: string;
deployRepo: string;
deployCommit: string;
deployRequestedCommit: string;
mgrDatabasePoolMax: number;
traceDatabasePoolMax: number;
defaultProviderId: string;
@@ -284,6 +288,10 @@ function configFromEnv(): RuntimeConfig {
port: envNumber("PORT", 4278),
databaseUrl,
logFile: envString("LOG_FILE", "/var/log/unidesk/code-queue-mgr.jsonl"),
deployServiceId: envString("UNIDESK_CODE_QUEUE_MGR_DEPLOY_SERVICE_ID", "code-queue-mgr"),
deployRepo: envString("UNIDESK_CODE_QUEUE_MGR_DEPLOY_REPO", ""),
deployCommit: envString("UNIDESK_CODE_QUEUE_MGR_DEPLOY_COMMIT", ""),
deployRequestedCommit: envString("UNIDESK_CODE_QUEUE_MGR_DEPLOY_REQUESTED_COMMIT", ""),
mgrDatabasePoolMax: Math.max(1, Math.min(2, envNumber("CODE_QUEUE_MGR_DATABASE_POOL_MAX", 2))),
traceDatabasePoolMax: Math.max(1, Math.min(2, envNumber("CODE_QUEUE_TRACE_DATABASE_POOL_MAX", 1))),
defaultProviderId: envString("CODE_QUEUE_MAIN_PROVIDER_ID", "D601"),
@@ -1240,26 +1248,27 @@ async function queueSummary(tasks?: QueueTask[], queueRecords?: QueueRecord[]):
counts[task.status] = Number(counts[task.status] ?? 0) + 1;
summary.counts = counts;
if (terminalTaskUnread(task)) summary.unreadTerminal = Number(summary.unreadTerminal) + 1;
if ((task.status === "running" || task.status === "judging") && summary.activeTaskId === null) summary.activeTaskId = task.id;
if ((task.status === "queued" || task.status === "retry_wait") && summary.runnableTaskId === null) summary.runnableTaskId = task.id;
}
const counts = taskRows.reduce<Record<string, number>>((memo, task) => {
memo[task.status] = (memo[task.status] ?? 0) + 1;
return memo;
}, {});
const activeTaskIds = taskRows.filter((task) => task.status === "running" || task.status === "judging").map((task) => task.id).sort();
const orphanedActiveTaskIds = 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,
defaultQueueId: config.defaultQueueId,
queueCount: queueList.length,
queues: queueList,
activeQueueIds: queueList.filter((queue) => typeof queue.activeTaskId === "string").map((queue) => String(queue.id)),
activeQueueIds: [],
processingQueueIds: [],
activeRunSlotCount: activeTaskIds.length,
activeRunSlotCount: 0,
activeRunSlotWaiters: [],
activeTaskIds,
activeTaskId: activeTaskIds[0] ?? null,
activeTaskIds: [],
activeTaskId: null,
orphanedActiveTaskIds,
orphanedActiveTaskCount: orphanedActiveTaskIds.length,
processing: false,
counts,
unreadTerminal: taskRows.reduce((total, task) => total + (terminalTaskUnread(task) ? 1 : 0), 0),
@@ -2059,6 +2068,12 @@ async function route(req: Request): Promise<Response> {
ok: schemaReady,
service: "code-queue-mgr",
role: "master-control-plane",
deploy: {
serviceId: config.deployServiceId,
repo: config.deployRepo,
commit: config.deployCommit,
requestedCommit: config.deployRequestedCommit,
},
startedAt: serviceStartedAt,
schemaReady,
schemaLastError,
src/components/microservices/code-queue/src/index.ts
+11 -2
View File
@@ -3201,6 +3201,7 @@ function taskHasLocalExecutionRecoveryClaim(task: QueueTask): boolean {
function queueActiveTasksForRestartRetry(reason: string, method: string, options: { onlyActiveRuns?: boolean } = {}): number {
if (!config.schedulerEnabled || !serviceRoleAllowsScheduler(config.serviceRole)) return 0;
let recovered = 0;
const recoveredTaskIds: string[] = [];
for (const task of state.tasks) {
if (task.status !== "running" && task.status !== "judging") continue;
if (options.onlyActiveRuns === true && !taskHasLocalExecutionRecoveryClaim(task)) continue;
@@ -3218,13 +3219,16 @@ function queueActiveTasksForRestartRetry(reason: string, method: string, options
task.nextPrompt = null;
task.lastError = `Max attempts reached (${maxTaskAttempts}) before restart recovery. ${reason}`;
appendOutput(task, "error", `${task.lastError}\n`, method);
recoveredTaskIds.push(task.id);
continue;
}
setAttemptFeedbackPrompt(task.attempts.at(-1), task.nextPrompt, "queue-recovery-retry", task.attempts.length + 1);
task.updatedAt = nowIso();
appendOutput(task, "system", `${reason}; task queued for retry\n`, method);
recoveredTaskIds.push(task.id);
recovered += 1;
}
for (const taskId of recoveredTaskIds) markTaskDirty(taskId);
if (recovered > 0) armIdleNotification();
return recovered;
}
@@ -4929,13 +4933,18 @@ startMemoryWatchdog();
Bun.serve({ hostname: config.host, port: config.port, idleTimeout: 120, fetch: route });
logger("info", "service_listening", { 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", databaseReady });
function startDatabaseBackedRuntime(): void {
async function startDatabaseBackedRuntime(): Promise<void> {
if (serviceReady) return;
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 startupRecovered = config.schedulerEnabled ? queueActiveTasksForRestartRetry("Service restarted while task was active", "startup") : 0;
if (startupRecovered > 0) logger("warn", "startup_requeued_active_tasks", { recovered: startupRecovered });
const startupRecoveryDirtyTaskCount = dirtyDatabaseTaskIds.size;
if (startupRecovered > 0 || startupRecoveryDirtyTaskCount > 0) {
logger("warn", "startup_requeued_active_tasks", { recovered: startupRecovered, dirtyTaskCount: startupRecoveryDirtyTaskCount });
await flushDirtyTasksToDatabase(true);
logger("info", "startup_requeued_active_tasks_flushed", { recovered: startupRecovered, dirtyTaskCount: startupRecoveryDirtyTaskCount });
}
persistState();
serviceReady = true;
void refreshSchedulerTasksFromDatabase("startup").catch((error) => {
src/components/microservices/code-queue/src/queue-api.ts
+9 -6
View File
@@ -375,7 +375,7 @@ function perQueueSummaries(tasks: QueueTask[] = ctx().tasks(), queueRecords: Que
for (const [queueId, summary] of summaries) {
const activeRun = ctx().activeRuns.get(queueId);
const head = ctx().queueHeadTask(queueId, tasks);
summary.activeTaskId = activeRun?.taskId ?? (head !== null && (head.status === "running" || head.status === "judging") ? head.id : null);
summary.activeTaskId = activeRun?.taskId ?? null;
summary.runnableTaskId = head !== null && ctx().queueTaskIsRunnable(head) ? head.id : null;
}
const rows = Array.from(summaries.entries()).sort(([left], [right]) => left.localeCompare(right)).map(([queueId, summary]) => {
@@ -403,11 +403,12 @@ function queueSummary(includeDevReady = true, tasks: QueueTask[] = ctx().tasks()
const unreadTerminal = tasks.reduce((total, task) => total + (terminalTaskUnread(task) ? 1 : 0), 0);
const activeRunSlots = ctx().activeRunSlotQueueIds();
const activeTaskIdSet = new Set<string>(Array.from(ctx().activeRuns.values()).map((run) => run.taskId));
for (const task of tasks) {
if (task.status === "running" || task.status === "judging") activeTaskIdSet.add(task.id);
}
const orphanedActiveTaskIds = tasks
.filter((task) => (task.status === "running" || task.status === "judging") && !activeTaskIdSet.has(task.id))
.map((task) => task.id)
.sort();
const activeTaskIds = Array.from(activeTaskIdSet).sort();
const activeTaskId = activeTaskIds[0] ?? tasks.find((task) => task.status === "running" || task.status === "judging")?.id ?? null;
const activeTaskId = activeTaskIds[0] ?? null;
const queues = perQueueSummaries(tasks, queueRecords);
const summary: Record<string, JsonValue> = {
total: tasks.length,
@@ -420,6 +421,8 @@ function queueSummary(includeDevReady = true, tasks: QueueTask[] = ctx().tasks()
activeRunSlotWaiters: ctx().activeRunSlotWaiterSummaries(),
activeTaskIds,
activeTaskId,
orphanedActiveTaskIds,
orphanedActiveTaskCount: orphanedActiveTaskIds.length,
processing: ctx().processing(),
counts,
unreadTerminal,
@@ -611,7 +614,7 @@ async function queueSummaryForHealth(includeDevReady = true): Promise<JsonValue>
for (const [queueId, queue] of summaries) {
const activeRun = ctx().activeRuns.get(queueId);
const head = ctx().queueHeadTask(queueId, hotTasks);
queue.activeTaskId = activeRun?.taskId ?? (head !== null && (head.status === "running" || head.status === "judging") ? head.id : null);
queue.activeTaskId = activeRun?.taskId ?? null;
queue.runnableTaskId = head !== null && ctx().queueTaskIsRunnable(head) ? head.id : null;
}
const queues = Array.from(summaries.entries()).sort(([left], [right]) => left.localeCompare(right)).map(([id, queue]) => ({