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

fix: keep code queue live during db outages

Browse Source
This commit is contained in:
Codex
2026-05-17 08:21:30 +00:00
parent 12368b7091
commit 85122c815d
2 changed files with 177 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/microservices/code-queue/src/index.ts
+75 -15
View File
@@ -1641,7 +1641,9 @@ async function findTaskForRead(taskId: string): Promise<QueueTask | null> {
}
async function findTaskForMutation(taskId: string): Promise<QueueTask | null> {
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<Response> {
}
async function createWorkdir(req: Request): Promise<Response> {
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<string, unknown> : {};
const providerId = normalizeTaskProviderId(record.providerId);
@@ -3627,6 +3647,8 @@ async function createWorkdir(req: Request): Promise<Response> {
}
async function deleteWorkdir(providerIdValue: string, executionModeValue: string, pathValue: string): Promise<Response> {
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<Response> {
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<string, unknown> : {};
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<Response> {
async function editQueuedTaskPrompt(task: QueueTask, req: Request): Promise<Response> {
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<Response
async function manualRetry(task: QueueTask, req: Request): Promise<Response> {
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<Response> {
async function markTaskRead(task: QueueTask): Promise<Response> {
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<Response> {
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<Response> {
async function markTerminalTasksRead(url: URL): Promise<Response> {
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<Response> {
async function createQueue(req: Request): Promise<Response> {
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<string, unknown> : {};
const queueId = normalizeQueueId(record.queueId ?? record.id);
@@ -3933,6 +3969,8 @@ async function createQueue(req: Request): Promise<Response> {
async function updateQueue(queueIdValue: string, req: Request): Promise<Response> {
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<Response> {
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<string, unknown> : {};
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<Response> {
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<Response> {
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<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 });
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) });
});
src/components/microservices/k3sctl-adapter/k3s/code-queue.k8s.yaml
+102 -34
View File
@@ -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