From 85122c815d71521d6b14ce195fdf05096c68a2d5 Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 17 May 2026 08:21:30 +0000 Subject: [PATCH] fix: keep code queue live during db outages --- .../microservices/code-queue/src/index.ts | 90 ++++++++++-- .../k3sctl-adapter/k3s/code-queue.k8s.yaml | 136 +++++++++++++----- 2 files changed, 177 insertions(+), 49 deletions(-) diff --git a/src/components/microservices/code-queue/src/index.ts b/src/components/microservices/code-queue/src/index.ts index d34b0789..6767361c 100644 --- a/src/components/microservices/code-queue/src/index.ts +++ b/src/components/microservices/code-queue/src/index.ts @@ -1641,7 +1641,9 @@ async function findTaskForRead(taskId: string): Promise { } async function findTaskForMutation(taskId: string): Promise { - const task = findTask(taskId) ?? await loadTaskFromDatabase(taskId); + const hotTask = findTask(taskId); + if (!databaseReady) return hotTask; + const task = hotTask ?? await loadTaskFromDatabase(taskId); return task === null ? null : rememberHotTask(task); } @@ -3556,6 +3558,22 @@ function readOnlyRejectResponse(method: string, targetPath: string): Response { }, 405); } +function databaseNotReadyResponse(method: string, targetPath: string): Response { + return jsonResponse({ + ok: false, + error: "Code Queue PostgreSQL storage is not ready", + serviceRole: config.serviceRole, + method, + path: targetPath, + databaseReady, + databaseLastError, + }, 503); +} + +function requireDatabaseReadyForWrite(method: string, targetPath: string): Response | null { + return databaseReady ? null : databaseNotReadyResponse(method, targetPath); +} + function schedulerOnlyRejectResponse(method: string, targetPath: string): Response { return jsonResponse({ ok: false, @@ -3599,6 +3617,8 @@ async function listWorkdirs(url: URL): Promise { } async function createWorkdir(req: Request): Promise { + const notReady = requireDatabaseReadyForWrite(req.method, "/api/workdirs"); + if (notReady !== null) return notReady; const body = await readJson(req); const record = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record : {}; const providerId = normalizeTaskProviderId(record.providerId); @@ -3627,6 +3647,8 @@ async function createWorkdir(req: Request): Promise { } async function deleteWorkdir(providerIdValue: string, executionModeValue: string, pathValue: string): Promise { + const notReady = requireDatabaseReadyForWrite("DELETE", `/api/workdirs/${providerIdValue}/${executionModeValue}/${pathValue}`); + if (notReady !== null) return notReady; const providerId = normalizeTaskProviderId(providerIdValue); const executionMode = normalizeCodeExecutionMode(executionModeValue); const path = normalizeWorkdirPath(pathValue, providerId); @@ -3648,6 +3670,8 @@ async function deleteWorkdir(providerIdValue: string, executionModeValue: string async function createTasks(req: Request): Promise { if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, "/api/tasks"); + const notReady = requireDatabaseReadyForWrite(req.method, "/api/tasks"); + if (notReady !== null) return notReady; const body = await readJson(req); const batchRecord = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record : {}; const batchQueueId = typeof batchRecord.queueId === "string" && batchRecord.queueId.trim().length > 0 ? normalizeQueueId(batchRecord.queueId) : undefined; @@ -3690,6 +3714,8 @@ async function steerTask(task: QueueTask, req: Request): Promise { async function editQueuedTaskPrompt(task: QueueTask, req: Request): Promise { if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, `/api/tasks/${task.id}/edit`); + const notReady = requireDatabaseReadyForWrite(req.method, `/api/tasks/${task.id}/edit`); + if (notReady !== null) return notReady; if (!queuedTaskPromptEditable(task)) { return jsonResponse({ ok: false, @@ -3763,6 +3789,8 @@ async function interruptTask(task: QueueTask, method = "POST"): Promise { if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, `/api/tasks/${task.id}/retry`); + const notReady = requireDatabaseReadyForWrite(req.method, `/api/tasks/${task.id}/retry`); + if (notReady !== null) return notReady; if (task.status !== "failed" && task.status !== "canceled" && task.status !== "succeeded") { return jsonResponse({ ok: false, error: `task is not terminal: ${task.status}`, task: taskForResponse(task) }, 409); } @@ -3790,6 +3818,8 @@ async function manualRetry(task: QueueTask, req: Request): Promise { async function markTaskRead(task: QueueTask): Promise { if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse("POST", `/api/tasks/${task.id}/read`); + const notReady = requireDatabaseReadyForWrite("POST", `/api/tasks/${task.id}/read`); + if (notReady !== null) return notReady; if (!terminalTask(task)) { return jsonResponse({ ok: false, error: `task is not terminal: ${task.status}`, task: taskForResponse(task) }, 409); } @@ -3812,6 +3842,8 @@ function timestampToIso(value: Date | string | null): string | null { async function markTaskReadById(taskId: string): Promise { if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse("POST", `/api/tasks/${taskId}/read`); + const notReady = requireDatabaseReadyForWrite("POST", `/api/tasks/${taskId}/read`); + if (notReady !== null) return notReady; if (!databaseReady) { const task = await findTaskForMutation(taskId); return task === null ? jsonResponse({ ok: false, error: "task not found" }, 404) : markTaskRead(task); @@ -3862,6 +3894,8 @@ async function markTaskReadById(taskId: string): Promise { async function markTerminalTasksRead(url: URL): Promise { if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse("POST", "/api/tasks/read-all"); + const notReady = requireDatabaseReadyForWrite("POST", "/api/tasks/read-all"); + if (notReady !== null) return notReady; const queueFilter = url.searchParams.get("queueId"); const queueId = queueFilter === null || queueFilter.length === 0 ? null : safeQueueId(queueFilter); const readAt = nowIso(); @@ -3916,6 +3950,8 @@ async function markTerminalTasksRead(url: URL): Promise { async function createQueue(req: Request): Promise { if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, "/api/queues"); + const notReady = requireDatabaseReadyForWrite(req.method, "/api/queues"); + if (notReady !== null) return notReady; const body = await readJson(req); const record = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record : {}; const queueId = normalizeQueueId(record.queueId ?? record.id); @@ -3933,6 +3969,8 @@ async function createQueue(req: Request): Promise { async function updateQueue(queueIdValue: string, req: Request): Promise { if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, `/api/queues/${queueIdValue}`); + const notReady = requireDatabaseReadyForWrite(req.method, `/api/queues/${queueIdValue}`); + if (notReady !== null) return notReady; const queueId = normalizeQueueId(queueIdValue); const queue = state.queues.find((item) => item.id === queueId); if (queue === undefined) return jsonResponse({ ok: false, error: "queue not found" }, 404); @@ -4049,6 +4087,9 @@ function deleteQueuesFromState(queueIds: string[]): QueueRecord[] { async function mergeQueues(targetQueueIdValue: string | null, req: Request): Promise { if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, targetQueueIdValue === null ? "/api/queues/merge" : `/api/queues/${targetQueueIdValue}/merge`); + const mergePath = targetQueueIdValue === null ? "/api/queues/merge" : `/api/queues/${targetQueueIdValue}/merge`; + const notReady = requireDatabaseReadyForWrite(req.method, mergePath); + if (notReady !== null) return notReady; const body = await readJson(req); const record = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record : {}; const targetQueueId = normalizeQueueId(targetQueueIdValue ?? record.targetQueueId ?? record.intoQueueId ?? record.into); @@ -4133,6 +4174,8 @@ async function mergeQueues(targetQueueIdValue: string | null, req: Request): Pro async function moveTaskToQueue(task: QueueTask, req: Request): Promise { if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, `/api/tasks/${task.id}/move`); + const notReady = requireDatabaseReadyForWrite(req.method, `/api/tasks/${task.id}/move`); + if (notReady !== null) return notReady; if (task.status === "running" || task.status === "judging") { return jsonResponse({ ok: false, error: `cannot move active task ${task.id} while status=${task.status}`, task: taskForResponse(task) }, 409); } @@ -4406,6 +4449,11 @@ async function route(req: Request): Promise { if (match !== null) { const action = match[2]; const taskId = decodeURIComponent(match[1] ?? ""); + if ((action === "retry" || action === "move" || action === "read" || action === "edit") && req.method !== "GET") { + if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, `/api/tasks/${taskId}/${action}`); + const notReady = requireDatabaseReadyForWrite(req.method, `/api/tasks/${taskId}/${action}`); + if (notReady !== null) return notReady; + } if (action === "read" && req.method === "POST") return await markTaskReadById(taskId); const task = action === undefined && req.method === "GET" ? await findTaskForRead(taskId) @@ -4443,21 +4491,33 @@ prepareCodexHome(); prepareOpenCodeHome(); startCodexSqliteLogExporter(); startMemoryWatchdog(); -await initDatabasePersistenceWithRetry(); Bun.serve({ hostname: config.host, port: config.port, idleTimeout: 120, fetch: route }); -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" }); -{ +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 { + 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 }); + persistState(); + serviceReady = true; + void refreshSchedulerTasksFromDatabase("startup").catch((error) => { + databaseLastError = databaseErrorMessage(error); + logger("warn", "scheduler_database_startup_refresh_failed", { error: errorToJson(error) }); + }); + startSchedulerDatabasePoller(); + if (config.startupOaBackfillEnabled) { + setTimeout(() => { void backfillOaTraceStats(new URL("http://code-queue.local/api/oa/backfill?limit=2000")).catch((error) => logger("warn", "oa_trace_stats_startup_backfill_failed", { error: errorToJson(error) })); }, 1000).unref?.(); + } + scheduleQueue(); + scheduleClaudeQqNotificationDrain(1000); } -const startupRecovered = config.schedulerEnabled ? queueActiveTasksForRestartRetry("Service restarted while task was active", "startup") : 0; -if (startupRecovered > 0) logger("warn", "startup_requeued_active_tasks", { recovered: startupRecovered }); -persistState(); -serviceReady = true; -await refreshSchedulerTasksFromDatabase("startup"); -startSchedulerDatabasePoller(); -if (config.startupOaBackfillEnabled) { - setTimeout(() => { void backfillOaTraceStats(new URL("http://code-queue.local/api/oa/backfill?limit=2000")).catch((error) => logger("warn", "oa_trace_stats_startup_backfill_failed", { error: errorToJson(error) })); }, 1000).unref?.(); -} -scheduleQueue(); -scheduleClaudeQqNotificationDrain(1000); + +void initDatabasePersistenceWithRetry() + .then(() => startDatabaseBackedRuntime()) + .catch((error) => { + databaseLastError = databaseErrorMessage(error); + logger("error", "database_persistence_init_failed", { error: errorToJson(error) }); + }); diff --git a/src/components/microservices/k3sctl-adapter/k3s/code-queue.k8s.yaml b/src/components/microservices/k3sctl-adapter/k3s/code-queue.k8s.yaml index 0f2375f6..0136bb18 100644 --- a/src/components/microservices/k3sctl-adapter/k3s/code-queue.k8s.yaml +++ b/src/components/microservices/k3sctl-adapter/k3s/code-queue.k8s.yaml @@ -689,6 +689,15 @@ data: const rawRoutes = process.env.TCP_EGRESS_ROUTES || ""; const startedAt = new Date().toISOString(); + function positiveInt(value, fallback) { + const parsed = Number(value); + return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback; + } + + const connectAttempts = positiveInt(process.env.TCP_EGRESS_CONNECT_ATTEMPTS, 8); + const connectRetryBackoffMs = positiveInt(process.env.TCP_EGRESS_CONNECT_RETRY_BACKOFF_MS, 250); + const maxEarlyClientBytes = positiveInt(process.env.TCP_EGRESS_MAX_EARLY_CLIENT_BYTES, 1024 * 1024); + function parseRoute(text) { const parts = text.trim().split("="); if (parts.length !== 3) throw new Error(`invalid route: ${text}`); @@ -715,55 +724,110 @@ data: } function pipeViaHttpConnect(client, route) { - const proxy = net.connect({ host: proxyHost, port: proxyPort }); const earlyClientChunks = []; - const proxyHeaderChunks = []; + let earlyClientBytes = 0; + let proxy = null; let tunnelReady = false; let closed = false; + let attempt = 0; + let retryTimer = null; + + const cleanupProxy = () => { + if (proxy === null) return; + proxy.removeAllListeners(); + proxy.destroy(); + proxy = null; + }; const closeBoth = () => { if (closed) return; closed = true; + if (retryTimer !== null) clearTimeout(retryTimer); client.destroy(); - proxy.destroy(); + cleanupProxy(); + }; + + const retryOrClose = (reason, data = {}) => { + cleanupProxy(); + if (!closed && attempt < connectAttempts) { + const delayMs = Math.min(connectRetryBackoffMs * attempt, 3000); + log("warn", "connect_retry", { route: route.name, attempt, attempts: connectAttempts, delayMs, reason, ...data }); + retryTimer = setTimeout(() => { + retryTimer = null; + startProxy(); + }, delayMs); + retryTimer.unref?.(); + return; + } + log("warn", "connect_rejected", { route: route.name, attempt, attempts: connectAttempts, reason, ...data }); + closeBoth(); }; client.on("data", (chunk) => { - if (tunnelReady) proxy.write(chunk); - else earlyClientChunks.push(chunk); + if (tunnelReady && proxy !== null) { + proxy.write(chunk); + return; + } + earlyClientBytes += chunk.length; + if (earlyClientBytes > maxEarlyClientBytes) { + log("warn", "early_client_buffer_limit_exceeded", { route: route.name, bytes: earlyClientBytes, limit: maxEarlyClientBytes }); + closeBoth(); + return; + } + earlyClientChunks.push(chunk); }); client.on("error", closeBoth); client.on("close", closeBoth); - proxy.on("connect", () => { - proxy.write(`CONNECT ${route.targetHost}:${route.targetPort} HTTP/1.1\r\nHost: ${route.targetHost}:${route.targetPort}\r\n\r\n`); - }); - proxy.on("data", (chunk) => { - if (tunnelReady) { - client.write(chunk); - return; - } - proxyHeaderChunks.push(chunk); - const headerBuffer = Buffer.concat(proxyHeaderChunks); - const headerEnd = findHeaderEnd(headerBuffer); - if (headerEnd < 0) return; - const header = headerBuffer.subarray(0, headerEnd).toString("latin1"); - if (!/^HTTP\/1\.[01] 200\b/u.test(header)) { - log("warn", "connect_rejected", { route: route.name, header: header.slice(0, 200) }); + const startProxy = () => { + if (closed) return; + attempt += 1; + tunnelReady = false; + const currentProxy = net.connect({ host: proxyHost, port: proxyPort }); + const proxyHeaderChunks = []; + proxy = currentProxy; + + currentProxy.on("connect", () => { + currentProxy.write(`CONNECT ${route.targetHost}:${route.targetPort} HTTP/1.1\r\nHost: ${route.targetHost}:${route.targetPort}\r\n\r\n`); + }); + currentProxy.on("data", (chunk) => { + if (currentProxy !== proxy) return; + if (tunnelReady) { + client.write(chunk); + return; + } + proxyHeaderChunks.push(chunk); + const headerBuffer = Buffer.concat(proxyHeaderChunks); + const headerEnd = findHeaderEnd(headerBuffer); + if (headerEnd < 0) return; + const header = headerBuffer.subarray(0, headerEnd).toString("latin1"); + if (!/^HTTP\/1\.[01] 200\b/u.test(header)) { + const body = headerBuffer.subarray(headerEnd).toString("utf8").slice(0, 300); + retryOrClose("non_200", { header: header.slice(0, 200), body: body.length > 0 ? body : null }); + return; + } + tunnelReady = true; + const remaining = headerBuffer.subarray(headerEnd); + if (remaining.length > 0) client.write(remaining); + for (const early of earlyClientChunks) currentProxy.write(early); + earlyClientChunks.length = 0; + earlyClientBytes = 0; + }); + currentProxy.on("error", (error) => { + if (currentProxy !== proxy) return; + retryOrClose("proxy_error", { error: String(error?.message || error) }); + }); + currentProxy.on("close", () => { + if (currentProxy !== proxy || closed) return; + if (!tunnelReady) { + retryOrClose("proxy_closed_before_tunnel"); + return; + } closeBoth(); - return; - } - tunnelReady = true; - const remaining = headerBuffer.subarray(headerEnd); - if (remaining.length > 0) client.write(remaining); - for (const early of earlyClientChunks) proxy.write(early); - earlyClientChunks.length = 0; - }); - proxy.on("error", (error) => { - log("warn", "proxy_error", { route: route.name, error: String(error?.message || error) }); - closeBoth(); - }); - proxy.on("close", closeBoth); + }); + }; + + startProxy(); } for (const route of routes) { @@ -776,7 +840,7 @@ data: hostname: "0.0.0.0", port: healthPort, fetch() { - return Response.json({ ok: true, service: "tcp-egress-gateway", startedAt, proxy: `${proxyHost}:${proxyPort}`, routes }); + return Response.json({ ok: true, service: "tcp-egress-gateway", startedAt, proxy: `${proxyHost}:${proxyPort}`, routes, connectAttempts, connectRetryBackoffMs }); }, }); --- @@ -826,6 +890,10 @@ spec: value: "postgres=15432=74.48.78.17:15432,oa-event-flow=4255=74.48.78.17:4255" - name: TCP_EGRESS_HEALTH_PORT value: "18080" + - name: TCP_EGRESS_CONNECT_ATTEMPTS + value: "8" + - name: TCP_EGRESS_CONNECT_RETRY_BACKOFF_MS + value: "250" volumeMounts: - name: script mountPath: /etc/unidesk-tcp-egress