fix: contain postgres unhandled connection timeouts

This commit is contained in:
lyon
2026-06-21 04:00:50 +08:00
parent e6d7a9ac6b
commit adba4a4354
2 changed files with 61 additions and 1 deletions
+33
View File
@@ -1,8 +1,12 @@
import { createCloudApiServer } from "./server.ts";
import { createHwpodNodeWsRegistry } from "./hwpod-node-ws-registry.ts";
import { classifyRuntimeDbError } from "../db/runtime-store.ts";
const POSTGRES_TRANSIENT_UNHANDLED_REJECTION_BOUNDARY = Symbol.for("hwlab.cloud.postgresTransientUnhandledRejectionBoundaryInstalled");
export async function createCloudApiBunServer(options: any = {}) {
const env = options.env ?? process.env;
installPostgresTransientUnhandledRejectionBoundary({ env, logger: options.logger ?? console });
const hwpodNodeWsRegistry = options.hwpodNodeWsRegistry || createHwpodNodeWsRegistry({ ...options, env });
const innerServer = createCloudApiServer({ ...options, env, hwpodNodeWsRegistry });
await new Promise((resolve) => innerServer.listen(0, "127.0.0.1", resolve));
@@ -54,6 +58,35 @@ export async function createCloudApiBunServer(options: any = {}) {
};
}
export function installPostgresTransientUnhandledRejectionBoundary(options: any = {}) {
const proc: any = options.process ?? process;
if (!proc || typeof proc.on !== "function") return { ok: false, installed: false, reason: "process-unavailable" };
if (proc[POSTGRES_TRANSIENT_UNHANDLED_REJECTION_BOUNDARY]) return { ok: true, installed: false, reason: "already-installed" };
proc[POSTGRES_TRANSIENT_UNHANDLED_REJECTION_BOUNDARY] = true;
const logger = options.logger ?? console;
proc.on("unhandledRejection", (reason: any) => {
const classified = classifyRuntimeDbError(reason);
if (classified.retryable === true && classified.transient === true && classified.connection?.queryResult === "connect_timeout") {
logger?.warn?.({
event: "postgres_runtime_unhandled_rejection_contained",
blocker: classified.blocker,
queryResult: classified.connection?.queryResult,
errorCode: classified.connection?.errorCode ?? "UNKNOWN",
retryable: true,
transient: true,
retryAfterMs: classified.retryAfterMs ?? null,
valuesRedacted: true,
endpointRedacted: true
});
return;
}
queueMicrotask(() => {
throw reason;
});
});
return { ok: true, installed: true };
}
async function proxyHttpToInnerServer(request: Request, innerPort: number) {
const sourceUrl = new URL(request.url);
const targetUrl = new URL(sourceUrl.pathname + sourceUrl.search, `http://127.0.0.1:${innerPort}`);
+28 -1
View File
@@ -1787,6 +1787,9 @@ export class PostgresCloudRuntimeStore {
blocker: classified.blocker,
queryResult: classified.connection?.queryResult ?? "query_blocked",
errorCode: classified.connection?.errorCode ?? "UNKNOWN",
retryable: classified.retryable === true,
transient: classified.transient === true,
retryAfterMs: classified.retryAfterMs ?? null,
valuesRedacted: true,
endpointRedacted: true
});
@@ -1882,6 +1885,9 @@ export class PostgresCloudRuntimeStore {
blocker: classified.blocker,
queryResult: classified.connection?.queryResult ?? "query_blocked",
errorCode: classified.connection?.errorCode ?? "UNKNOWN",
retryable: classified.retryable === true,
transient: classified.transient === true,
retryAfterMs: classified.retryAfterMs ?? null,
valuesRedacted: true,
endpointRedacted: true
});
@@ -2677,8 +2683,22 @@ function runtimeSafety() {
};
}
function classifyRuntimeDbError(error) {
export function classifyRuntimeDbError(error) {
const code = typeof error?.code === "string" ? error.code : "UNKNOWN";
if (isRuntimeDbConnectTimeout(error)) {
return {
blocker: RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED,
reason: "Postgres runtime adapter connection attempt timed out before completing the runtime query",
retryable: true,
transient: true,
retryAfterMs: 5000,
connection: {
queryAttempted: true,
queryResult: "connect_timeout",
errorCode: code
}
};
}
if (code === "HWLAB_PG_DRIVER_MISSING") {
return {
blocker: RUNTIME_DURABLE_ADAPTER_DRIVER_MISSING,
@@ -2734,6 +2754,13 @@ function classifyRuntimeDbError(error) {
};
}
function isRuntimeDbConnectTimeout(error) {
const message = String(error?.message ?? error ?? "").toLowerCase();
if (!message.includes("timeout exceeded when trying to connect")) return false;
const stack = String(error?.stack ?? "").toLowerCase();
return !stack || stack.includes("pg-pool") || stack.includes("node_modules/pg");
}
function durableRuntimeReadBlockedError(method, readiness = {}) {
return new HwlabProtocolError("Postgres durable runtime adapter is blocked for runtime evidence queries", {
code: ERROR_CODES.internalError,