From ab8e3ca3d199a8ac8f5184e672cafcf5f182509c Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 17 May 2026 15:56:56 +0000 Subject: [PATCH] fix: tolerate code queue database disconnects --- .../microservices/code-queue/src/index.ts | 93 +++++++++++++++++-- 1 file changed, 83 insertions(+), 10 deletions(-) diff --git a/src/components/microservices/code-queue/src/index.ts b/src/components/microservices/code-queue/src/index.ts index 775b6d3d..ab8ed700 100644 --- a/src/components/microservices/code-queue/src/index.ts +++ b/src/components/microservices/code-queue/src/index.ts @@ -205,12 +205,18 @@ let persistTimer: ReturnType | null = null; let persistDirty = false; let shutdownRequested = false; let serviceReady = false; -const sql: SqlClient = postgres(config.databaseUrl, { +let sql: SqlClient = createSqlClient(); +let sqlRotationInFlight: Promise | null = null; + +function createSqlClient(): SqlClient { + return postgres(config.databaseUrl, { max: config.databasePoolMax, idle_timeout: 20, connect_timeout: 10, connection: { application_name: "unidesk-code-queue" }, -}); + }); +} + let databaseReady = false; let databaseLastError: string | null = null; let databaseFlushTimer: ReturnType | null = null; @@ -1120,6 +1126,54 @@ function errorToJson(error: unknown): JsonValue { return String(error); } +function transientDatabaseErrorMessage(error: unknown): string { + const message = error instanceof Error ? error.message : String(error); + const stack = error instanceof Error ? error.stack ?? "" : ""; + return `${message}\n${stack}`; +} + +function isTransientDatabaseError(error: unknown): boolean { + const text = transientDatabaseErrorMessage(error); + if (/deadlock detected/iu.test(text)) return true; + const looksLikePostgresClient = /node_modules\/postgres|postgres\/src|d601-tcp-egress-gateway|:15432|unidesk_code_queue|set_config/iu.test(text); + if (!looksLikePostgresClient) return false; + return /CONNECTION_CLOSED|ECONNRESET|EPIPE|ETIMEDOUT|Connection terminated|socket connection was closed|socket\.write|null is not an object/iu.test(text); +} + +function transientDatabaseRetryDelayMs(attempt: number): number { + return Math.min(2000, 150 * Math.max(1, attempt)); +} + +async function rotateSqlClientAfterTransientError(reason: string, error: unknown): Promise { + databaseLastError = databaseErrorMessage(error); + if (sqlRotationInFlight !== null) return sqlRotationInFlight; + const previous = sql; + sql = createSqlClient(); + sqlRotationInFlight = previous.end({ timeout: 1 }) + .catch((closeError) => logger("warn", "database_client_close_failed", { reason, error: errorToJson(closeError) })) + .finally(() => { + sqlRotationInFlight = null; + }); + logger("warn", "database_client_rotated", { reason, error: errorToJson(error) }); + return sqlRotationInFlight; +} + +async function withTransientDatabaseRetries(operation: string, action: () => Promise, attempts = 3): Promise { + let lastError: unknown = null; + for (let attempt = 1; attempt <= attempts; attempt += 1) { + try { + return await action(); + } catch (error) { + lastError = error; + if (!isTransientDatabaseError(error) || attempt >= attempts) throw error; + logger("warn", "database_transient_retry", { operation, attempt, attempts, delayMs: transientDatabaseRetryDelayMs(attempt), error: errorToJson(error) }); + await rotateSqlClientAfterTransientError(operation, error); + await Bun.sleep(transientDatabaseRetryDelayMs(attempt)); + } + } + throw lastError; +} + function judgeFailureDetailsForOutput(judge: JudgeResult): string { const details = judge.failureDetails; if (details === undefined || details === null) return ""; @@ -1550,10 +1604,10 @@ async function deleteTaskFromDatabase(taskId: string): Promise { async function claimTaskInDatabase(task: QueueTask, expectedQueueId: string): Promise { if (!databaseReady) return true; const timeout = `${config.databaseClaimTimeoutMs}ms`; - const claimed = await sql.begin(async (client) => { + const claimed = await withTransientDatabaseRetries("claim-task", async () => sql.begin(async (client) => { await client`SELECT set_config('statement_timeout', ${timeout}, true), set_config('lock_timeout', ${timeout}, true)`; return await upsertTaskToDatabase(client, task, { claimQueueId: expectedQueueId }); - }); + })); if (claimed) return true; const databaseTask = await loadTaskFromDatabase(task.id); if (databaseTask !== null) reconcileHotTaskFromDatabase(databaseTask); @@ -2020,24 +2074,26 @@ async function flushDirtyTasksToDatabase(force = false): Promise { scheduleDatabaseFlush(); return; } - const ids = Array.from(dirtyDatabaseTaskIds); - const queueIds = Array.from(dirtyDatabaseQueueIds); + const ids = Array.from(dirtyDatabaseTaskIds).sort(); + const queueIds = Array.from(dirtyDatabaseQueueIds).sort(); if (ids.length === 0 && queueIds.length === 0) return; dirtyDatabaseTaskIds.clear(); dirtyDatabaseQueueIds.clear(); databaseFlushInFlight = true; - const rejectedTaskIds: string[] = []; + let rejectedTaskIds: string[] = []; try { - await sql.begin(async (client) => { + rejectedTaskIds = await withTransientDatabaseRetries("flush-dirty-tasks", async () => sql.begin(async (client) => { + const currentRejectedTaskIds: string[] = []; for (const id of queueIds) { const queue = state.queues.find((item) => item.id === id); if (queue !== undefined) await upsertQueueToDatabase(client, queue); } for (const id of ids) { const task = state.tasks.find((item) => item.id === id); - if (task !== undefined && !await upsertTaskToDatabase(client, task)) rejectedTaskIds.push(id); + if (task !== undefined && !await upsertTaskToDatabase(client, task)) currentRejectedTaskIds.push(id); } - }); + return currentRejectedTaskIds; + })); databaseLastError = null; } catch (error) { for (const id of ids) dirtyDatabaseTaskIds.add(id); @@ -3992,6 +4048,23 @@ function installShutdownHandlers(): void { }; process.once("SIGTERM", stop); process.once("SIGINT", stop); + process.on("uncaughtException", (error) => { + if (isTransientDatabaseError(error)) { + void rotateSqlClientAfterTransientError("uncaught-exception", error).catch((closeError) => logger("warn", "database_client_rotate_uncaught_failed", { error: errorToJson(closeError) })); + logger("error", "transient_database_uncaught_exception_suppressed", { error: errorToJson(error) }); + return; + } + logger("error", "uncaught_exception", { error: errorToJson(error) }); + throw error; + }); + process.on("unhandledRejection", (reason) => { + if (isTransientDatabaseError(reason)) { + void rotateSqlClientAfterTransientError("unhandled-rejection", reason).catch((error) => logger("warn", "database_client_rotate_rejection_failed", { error: errorToJson(error) })); + logger("error", "transient_database_unhandled_rejection_suppressed", { error: errorToJson(reason) }); + return; + } + logger("error", "unhandled_rejection", { error: errorToJson(reason) }); + }); } setInterval(() => {