From e5749c426f81be1434515fc06d2c608f59a849b3 Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 17 May 2026 10:42:51 +0000 Subject: [PATCH] fix: recover orphaned code queue runs --- docker-compose.yml | 4 +++ docs/reference/deploy.md | 1 + docs/reference/deployment.md | 4 +-- docs/reference/microservices.md | 1 + scripts/src/deploy.ts | 21 ++++++++++----- .../microservices/code-queue-mgr/src/index.ts | 27 ++++++++++++++----- .../microservices/code-queue/src/index.ts | 13 +++++++-- .../microservices/code-queue/src/queue-api.ts | 15 ++++++----- 8 files changed, 64 insertions(+), 22 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index 18733940..2e5482b2 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -223,6 +223,10 @@ services: CODE_QUEUE_WORKDIR: "/workspace" LOG_FILE: "/var/log/unidesk/${UNIDESK_LOG_DAY}/${UNIDESK_LOG_PREFIX}_code-queue-mgr.jsonl" UNIDESK_LOG_RETENTION_BYTES: "${UNIDESK_LOG_RETENTION_BYTES:-1GiB}" + UNIDESK_CODE_QUEUE_MGR_DEPLOY_SERVICE_ID: "${UNIDESK_CODE_QUEUE_MGR_DEPLOY_SERVICE_ID:-code-queue-mgr}" + UNIDESK_CODE_QUEUE_MGR_DEPLOY_REPO: "${UNIDESK_CODE_QUEUE_MGR_DEPLOY_REPO:-}" + UNIDESK_CODE_QUEUE_MGR_DEPLOY_COMMIT: "${UNIDESK_CODE_QUEUE_MGR_DEPLOY_COMMIT:-}" + UNIDESK_CODE_QUEUE_MGR_DEPLOY_REQUESTED_COMMIT: "${UNIDESK_CODE_QUEUE_MGR_DEPLOY_REQUESTED_COMMIT:-}" volumes: - ${UNIDESK_LOG_DIR}:/var/log/unidesk healthcheck: diff --git a/docs/reference/deploy.md b/docs/reference/deploy.md index 8b7387e8..5719ff25 100644 --- a/docs/reference/deploy.md +++ b/docs/reference/deploy.md @@ -67,6 +67,7 @@ Main server targets may build without a proxy unless a service explicitly requir The reconciler selects the executor from `config.json`: - `deployment.mode=unidesk-direct` on `main-server`: build the image on the main server, then use the fixed UniDesk Compose project and `up -d --no-build --no-deps --force-recreate `. +- `deployment.mode=internal-sidecar` on `main-server`: use the same main-server target-side source export, Docker build, image label stamping, fixed Compose project replacement and live commit verification as direct Compose services. This class is for private sidecars such as `code-queue-mgr`; it is still versioned by `deploy.json.commitId`, not by the operator's current worktree. - `deployment.mode=unidesk-direct` on a provider: dispatch `host.ssh` to that provider, build on the provider, then use the service's provider-local compose file and project. The executor resolves the actual Compose project, image name, build context, Dockerfile and target from the running container labels and `docker compose config`; it must not guess an image tag that the service will not actually run. - Control bridges that UniDesk needs in order to inspect or repair an orchestrator must stay in this direct class. In particular, `k3sctl-adapter` is a UniDesk-managed bridge to native k3s and must remain outside k3s; Docker packaging on Docker Desktop/WSL must create an explicit host-local bridge, currently an adapter-container SSH local tunnel, to reach `/etc/rancher/k3s/k3s.yaml` and WSL `127.0.0.1:6443`. - `deployment.mode=k3sctl-managed`: dispatch to the active control target, build on that target, verify or install native k3s on the host OS/WSL distro, import the image into native k3s/containerd, apply the existing Kubernetes manifest, stamp the Deployment and wait for rollout. The executor must use the native kubeconfig and containerd socket, for example `/etc/rancher/k3s/k3s.yaml` and `/run/k3s/containerd/containerd.sock`; running k3s itself in Docker is forbidden for both control-plane and worker nodes. A `rancher/k3s` image or legacy container may only be used as a temporary artifact source during migration, and any active containerized k3s control plane must be stopped before verification succeeds. The executor must preload a valid `rancher/mirrored-pause:3.6` sandbox image into native k3s containerd through the provider-gateway one-shot egress path, verify its entrypoint is `/pause`, and reject fake or sleep-based replacement images. Code Queue's k3s migration executor must also stop/remove the legacy direct Docker `code-queue-backend` after k3s rollout, so there is never a second scheduler running beside the native k3s scheduler. diff --git a/docs/reference/deployment.md b/docs/reference/deployment.md index ab8a164a..7c12cdf1 100644 --- a/docs/reference/deployment.md +++ b/docs/reference/deployment.md @@ -27,7 +27,7 @@ CLI 会优先使用 `docker compose` v2 plugin；当 v2 plugin 不存在时才 Compose v2 安装后仍然必须遵守 UniDesk 的服务控制入口：全栈生命周期用 `server start` / `server stop`，单服务重建用 `server rebuild `。不要因为 v2 可用就直接在生产栈上手工执行未纳入 CLI 的 `up --build`、`down -v` 或跨项目清理命令；所有会影响容器的动作都应保持 job 可观测、Compose project 固定、database named volume 保留。 -版本化用户服务部署优先使用 `bun scripts/cli.ts deploy apply`。`deploy.json` 只声明服务 `id`、`repo` 和 `commitId`；目标节点、Dockerfile、Compose、Kubernetes manifest、健康检查和代理路径继续来自 `config.json` 与现有 manifest。部署必须遵循 target-side build：服务部署到哪台 target，就在哪台 target 从 remote commit 导出源码、一次性代理构建镜像并部署；不得把中心构建镜像作为默认分发路径，也不得用 `docker commit` 或脏 worktree 作为部署输入。完整规则见 `docs/reference/deploy.md`。 +版本化用户服务部署优先使用 `bun scripts/cli.ts deploy apply`。`deploy.json` 只声明服务 `id`、`repo` 和 `commitId`；目标节点、Dockerfile、Compose、Kubernetes manifest、健康检查和代理路径继续来自 `config.json` 与现有 manifest。主 server 直管微服务和内部 sidecar，例如 `code-queue-mgr`，也必须支持这一路径：`deploy apply --service code-queue-mgr` 从 `deploy.json` 指定 commit 导出源码、构建镜像、替换固定 Compose service 并验证运行中镜像/健康信息的 commit。部署必须遵循 target-side build：服务部署到哪台 target，就在哪台 target 从 remote commit 导出源码、一次性代理构建镜像并部署；不得把中心构建镜像作为默认分发路径，也不得用 `docker commit` 或脏 worktree 作为部署输入。完整规则见 `docs/reference/deploy.md`。 ## Start And Stop @@ -35,7 +35,7 @@ Compose v2 安装后仍然必须遵守 UniDesk 的服务控制入口：全栈生 ## Single Service Rebuild -前端、backend-core、本机 provider-gateway 或主 server 承载的 Todo Note/Code Queue Manager/Project Manager/Baidu Netdisk/OA Event Flow 用户服务需要重建时，统一使用 `bun scripts/cli.ts server rebuild `，其中 `` 只能是 `backend-core`、`frontend`、`provider-gateway`、`todo-note`、`code-queue-mgr`、`project-manager`、`baidu-netdisk` 或 `oa-event-flow`。D601 Code Queue 执行面、File Browser、FindJob、Pipeline、MET Nonlinear 和 ClaudeQQ 部署在计算节点，不属于主 server Compose 可重建服务；其中 D601 Code Queue 执行面的正式入口是 `bun scripts/cli.ts codex deploy `。该命令先执行目标服务镜像构建，构建成功后才通过 `up -d --no-deps --force-recreate ` 替换目标容器，避免构建失败导致运行中的服务被提前停掉。 +前端、backend-core、本机 provider-gateway 或主 server 承载的 Todo Note/Code Queue Manager/Project Manager/Baidu Netdisk/OA Event Flow 用户服务需要非版本化本地重建时，统一使用 `bun scripts/cli.ts server rebuild `，其中 `` 只能是 `backend-core`、`frontend`、`provider-gateway`、`todo-note`、`code-queue-mgr`、`project-manager`、`baidu-netdisk` 或 `oa-event-flow`。需要按 commit 上线或恢复到 desired-state 时必须改用 `bun scripts/cli.ts deploy apply --service `；直管微服务也不能把脏工作树或手工重建作为部署真相。D601 Code Queue 执行面、File Browser、FindJob、Pipeline、MET Nonlinear 和 ClaudeQQ 部署在计算节点，不属于主 server Compose 可重建服务；其中 D601 Code Queue 执行面的正式入口是 `bun scripts/cli.ts codex deploy `。该命令先执行目标服务镜像构建，构建成功后才通过 `up -d --no-deps --force-recreate ` 替换目标容器，避免构建失败导致运行中的服务被提前停掉。 frontend 改动必须明确上线到公网：修改 `src/components/frontend/src/`、`src/components/frontend/public/style.css`、frontend 使用的共享 TSX/TS 模块或 WebUI 导航后，必须在同一变更集中执行 `bun scripts/cli.ts server rebuild frontend`，并等待 job 成功。公网 WebUI 的 `/app.js` 是 `unidesk-frontend` 容器启动时从镜像内源码转译生成的运行时 bundle；只改工作区文件、只跑 `bun run check`、只跑 `Bun.build` 或只刷新浏览器都不会替换已经运行的容器。 diff --git a/docs/reference/microservices.md b/docs/reference/microservices.md index 0e91d20c..1c2eacac 100644 --- a/docs/reference/microservices.md +++ b/docs/reference/microservices.md @@ -161,6 +161,7 @@ Baidu Netdisk 在 UniDesk 语境中按纯后端服务管理：不得暴露百度 - 服务拆分语义：`code-queue-read` 只承载 GET/HEAD 查询、overview、任务详情、Trace/output/transcript、统计和只读健康，可多副本滚动更新；它必须设置 `CODE_QUEUE_SERVICE_ROLE=read` 与 `CODE_QUEUE_SCHEDULER_ENABLED=false`，且不得接受入队、queue 变更、已读、重试、移动、追加 prompt 或打断这类 mutation。`code-queue-write` 承载入队、queue 创建/合并/更新、已读、手动重试、移动等命令写入，初期保持单副本和 `CODE_QUEUE_SERVICE_ROLE=write`，只把命令和任务状态写入 PostgreSQL，不启动 agent 子进程。`code-queue-scheduler` 是唯一拥有 scheduler 和 active run 的执行服务，设置 `CODE_QUEUE_SERVICE_ROLE=scheduler` 与 `CODE_QUEUE_SCHEDULER_ENABLED=true`，负责从 PostgreSQL 热任务集轮询新写入任务、推进队列、启动 Codex/OpenCode、处理 running task 的 steer/interrupt、发送终态通知和暴露执行端 `/health`。普通 Service 负载均衡不得把 mutation 打到 read，也不得把 running task 控制打到 write。 - 实例语义：D601 是当前唯一 active 执行节点，`code-queue-scheduler` 以一个 scheduler Pod 承载长生命周期 Codex/OpenCode 子进程并轮询主 PostgreSQL 中由 `code-queue-mgr` 写入的 queued/retry_wait 任务。D518 不属于当前 Code Queue k3s 拓扑；在没有原生 k3s-agent 与稳定 Kubernetes 网络前，不得把 D518 写回 `expectedNodeIds` 或恢复 `code-queue-d518` standby。D601 scheduler 默认关闭 `CODE_QUEUE_STARTUP_OA_BACKFILL_ENABLED`；历史 OA Trace/STEP 回填必须通过显式 `/api/oa/backfill` 运维动作触发，不能在每次 Pod 重启时自动批量发布旧事件。 - 滚动更新边界：master `code-queue-mgr` 保证 D601 抖动或执行面滚动更新期间普通提交、queue 管理和历史读取仍可用；但当前 D601 scheduler Pod 内仍直接承载正在运行的 agent 子进程，scheduler Pod 被替换时 active task 仍会进入 restart-recovery/retry 语义，不能宣称 running task 零中断。真正的长期目标是继续把调度器和执行器拆开：scheduler 只负责 claim task 并创建 Kubernetes Job/Pod 或独立 worker，runner 把输出、状态、attempt、事件和通知写回 PostgreSQL/OA Event Flow/归档；只有这样 controller/scheduler 滚动更新才不会影响正在执行的任务。 +- Restart recovery：D601 scheduler 启动时必须把没有本地 active run 的 `running`/`judging` 任务恢复为 `retry_wait` 并先写回 PostgreSQL，再开启新一轮 scheduler 轮询；否则 PG 中残留的 running 会阻塞队列且不会被执行。health/overview 中的 `activeTaskIds` 只代表当前进程真实持有的 agent run；数据库里仍处于 `running`/`judging` 但没有本地 run 的任务只能作为 `orphanedActiveTaskIds` 暴露，不能计入 active run slot。 - 部署引用：Code Queue 镜像仍复用 `src/components/microservices/code-queue/Dockerfile`，Kubernetes 运行清单为 `src/components/microservices/k3sctl-adapter/k3s/code-queue.k8s.yaml`，`config.json` 对外记录 k3s manifest `src/components/microservices/k3sctl-adapter/k3s/code-queue.k3s.json`；主 server 根目录 `docker-compose.yml` 不包含 `code-queue` service，旧 D601 direct Compose 文件只作为迁移/本地诊断参考，不是正式运行入口。 - 主服务依赖映射：Code Queue 仍以主 PostgreSQL 为权威数据库，但 D601 k3s Pod 不能依赖公网直连 `74.48.78.17:15432/4255`。Pod 内 `DATABASE_URL` 和 `OA_EVENT_FLOW_BASE_URL` 必须指向集群内 `d601-tcp-egress-gateway` Service，再由该 gateway 通过 D601 provider-gateway egress proxy 的 HTTP CONNECT 转发到主 PostgreSQL 和 OA Event Flow；新增 TCP 依赖时扩展 `TCP_EGRESS_ROUTES`，不得在业务容器里新增一次性公网直连或 ad hoc 隧道。D601 active 实例的 `CODE_QUEUE_NOTIFY_CLAUDEQQ_BASE_URL` 必须使用集群内 ClaudeQQ Service `http://claudeqq.unidesk.svc.cluster.local:3290`，并把 `claudeqq`/`claudeqq.unidesk.svc.cluster.local` 加入 `NO_PROXY`，避免任务完成通知被默认出网代理错误转发。旧 `http://host.docker.internal:3290` 只允许作为迁移期诊断，不得作为 Code Queue k3s Pod 的正式通知路径。这些端口映射只服务受控节点运行时，必须用防火墙或等价策略限制来源，不得成为浏览器或任意公网客户端入口。 - K8s 探针与启动维护：Kubernetes liveness/startup probe 必须使用轻量 `/live`，readiness 和用户服务健康使用 `/health`；`/health` 不得执行全量任务聚合、历史回填或长事务索引维护，历史任务总览应由 `/api/tasks/overview` 读取 PostgreSQL。启动时允许后台执行队列元数据 flush、通知 outbox 读取、任务表索引维护和 overview warmup，但这些维护不得阻塞 Bun server、readiness endpoint 或 frontend overview；通知表索引和大批量 OA backfill 不得作为默认启动副作用。 diff --git a/scripts/src/deploy.ts b/scripts/src/deploy.ts index bafc692e..630e213d 100644 --- a/scripts/src/deploy.ts +++ b/scripts/src/deploy.ts @@ -444,6 +444,10 @@ function isCoreDeployService(service: UniDeskMicroserviceConfig): boolean { return service.id === "frontend" && service.providerId === "main-server" && service.backend.proxyMode === "core-direct"; } +function isDirectComposeDeployMode(service: UniDeskMicroserviceConfig): boolean { + return service.deployment.mode === "unidesk-direct" || service.deployment.mode === "internal-sidecar"; +} + function selectServices(config: UniDeskConfig, manifest: DeployManifest, serviceId: string | null): Array<{ desired: DeployManifestService; config: UniDeskMicroserviceConfig; @@ -907,13 +911,18 @@ function imageLabelVerifyScript(service: UniDeskMicroserviceConfig, expectedComm ].join("\n"); } +function directDeployEnvPrefix(service: UniDeskMicroserviceConfig): string { + return `UNIDESK_${service.id.replace(/[^A-Za-z0-9]+/gu, "_").replace(/^_+|_+$/gu, "").toUpperCase()}_DEPLOY`; +} + function directComposeDeployEnv(service: UniDeskMicroserviceConfig, desired: DeployManifestService, resolvedCommit: string): Record { - if (service.id !== "frontend") return {}; + if (!isDirectComposeDeployMode(service)) return {}; + const prefix = directDeployEnvPrefix(service); return { - UNIDESK_FRONTEND_DEPLOY_SERVICE_ID: service.id, - UNIDESK_FRONTEND_DEPLOY_REPO: desired.repo, - UNIDESK_FRONTEND_DEPLOY_COMMIT: resolvedCommit, - UNIDESK_FRONTEND_DEPLOY_REQUESTED_COMMIT: desired.commitId, + [`${prefix}_SERVICE_ID`]: service.id, + [`${prefix}_REPO`]: desired.repo, + [`${prefix}_COMMIT`]: resolvedCommit, + [`${prefix}_REQUESTED_COMMIT`]: desired.commitId, }; } @@ -1669,7 +1678,7 @@ async function applyOneService(config: UniDeskConfig, service: UniDeskMicroservi if (!pushStep(steps, controlManifestSync)) return { ok: false, serviceId: service.id, startedAt, finishedAt: nowIso(), resolvedCommit, before, steps }; } - const buildScript = service.deployment.mode === "unidesk-direct" + const buildScript = isDirectComposeDeployMode(service) ? buildDirectImageScript(service, desired, resolvedCommit) : buildImageScript(service, desired, resolvedCommit); const build = await step(config, service, "docker-build", buildScript, targetWorkDir(service), Math.min(options.timeoutMs, 540_000), !targetIsMain(service)); diff --git a/src/components/microservices/code-queue-mgr/src/index.ts b/src/components/microservices/code-queue-mgr/src/index.ts index f40e06cd..6b8a6a40 100644 --- a/src/components/microservices/code-queue-mgr/src/index.ts +++ b/src/components/microservices/code-queue-mgr/src/index.ts @@ -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>((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 { 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, diff --git a/src/components/microservices/code-queue/src/index.ts b/src/components/microservices/code-queue/src/index.ts index 8747a80d..bf62b556 100644 --- a/src/components/microservices/code-queue/src/index.ts +++ b/src/components/microservices/code-queue/src/index.ts @@ -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 { 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; 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) => { diff --git a/src/components/microservices/code-queue/src/queue-api.ts b/src/components/microservices/code-queue/src/queue-api.ts index b0eda037..390dee0d 100644 --- a/src/components/microservices/code-queue/src/queue-api.ts +++ b/src/components/microservices/code-queue/src/queue-api.ts @@ -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(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 = { 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 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]) => ({