Stabilize provider HTTP tunnel diagnostics

This commit is contained in:
Codex
2026-05-17 06:53:03 +00:00
parent aaf6e74aa4
commit bf364baac8
12 changed files with 518 additions and 50 deletions
@@ -93,3 +93,11 @@ export function closeEgressTcpConnectionsForProvider(providerId: string): void {
connection.socket.destroy();
}
}
export function closeEgressTcpConnectionsForSocket(provider: ProviderSocket): void {
for (const [key, connection] of ctx.activeEgressTcpConnections) {
if (connection.provider !== provider) continue;
ctx.activeEgressTcpConnections.delete(key);
connection.socket.destroy();
}
}
+7 -6
View File
@@ -9,7 +9,7 @@ import { recordRequestPerformance, withPerformanceOperation, getPerformance } fr
import { handleProviderMessage, markProviderOffline, markStaleProvidersOffline } from "./provider-registry";
import { markStaleTasksFailed, dispatchTask } from "./task-dispatcher";
import { handleSshClientMessage, sshRoute } from "./ssh-bridge";
import { closeEgressTcpConnectionsForProvider } from "./egress-tcp";
import { closeEgressTcpConnectionsForProvider, closeEgressTcpConnectionsForSocket } from "./egress-tcp";
import { scheduledTaskRoute, runDueScheduledTasks, recoverScheduledRuns } from "./scheduler";
import { microserviceRoute, getMicroservices } from "./microservice-proxy";
import { getOverview, codexQueueLoadTest } from "./overview";
@@ -171,17 +171,18 @@ const providerServer = Bun.serve<WsData>({
const providerId = ws.data.providerId;
logger("warn", "provider_socket_close", { providerId: providerId ?? null });
if (providerId !== undefined) {
if (ctx.activeProviders.get(providerId) !== ws) {
closeEgressTcpConnectionsForSocket(ws);
logger("info", "provider_socket_close_ignored_replaced", { providerId });
return;
}
closeEgressTcpConnectionsForProvider(providerId);
for (const [requestId, waiter] of ctx.httpTunnelWaiters) {
if (requestId.startsWith(`${providerId}:`)) {
ctx.httpTunnelWaiters.delete(requestId);
waiter(null);
waiter(null, "provider-disconnected");
}
}
if (ctx.activeProviders.get(providerId) !== ws) {
logger("info", "provider_socket_close_ignored_replaced", { providerId });
return;
}
markProviderOffline(providerId).catch((error) => logger("error", "provider_offline_mark_failed", { providerId, error: errorToJson(error) }));
}
},
@@ -1,6 +1,6 @@
import type { JsonValue } from "../../shared/src/index";
import { ctx, config, logger } from "./context";
import type { MicroserviceConfig, MicroserviceProxyCacheEntry, MicroserviceHealthAssessment, MicroserviceAvailabilityEntry, RawTaskRow } from "./types";
import type { HttpTunnelFailureReason, MicroserviceConfig, MicroserviceProxyCacheEntry, MicroserviceHealthAssessment, MicroserviceAvailabilityEntry, RawTaskRow } from "./types";
import { jsonResponse, errorToJson, compactJson, isPlainRecord, truncateText } from "./http";
import { createAndSendTask, waitForTaskTerminal, providerSupports } from "./task-dispatcher";
import { getNodes, getNodeDockerStatuses } from "./db";
@@ -12,6 +12,7 @@ import { getNodes, getNodeDockerStatuses } from "./db";
const microserviceProxyMaxBodyTextLength = 8 * 1024 * 1024;
const microserviceAvailabilityTtlMs = 30_000;
const codeQueueOverviewPathFallbackStaleMs = 30_000;
const providerHttpTunnelMaxAttempts = 3;
const microserviceForwardRequestHeaders = [
"accept",
"content-type",
@@ -456,6 +457,13 @@ function responseFromMicroserviceCache(entry: MicroserviceProxyCacheEntry, state
});
}
function isMicroserviceTransientFailureResponse(response: Response): boolean {
if (response.status !== 502 && response.status !== 503 && response.status !== 504) return false;
return response.headers.get("x-unidesk-transient-error") === "true"
|| response.headers.get("x-unidesk-tunnel-error") !== null
|| response.headers.get("x-unidesk-upstream-proxy-mode") === "provider-gateway-http-fetch";
}
function readMicroserviceCache(key: string): Response | null {
const entry = ctx.microserviceProxyCache.get(key);
if (entry === undefined) return null;
@@ -626,43 +634,248 @@ async function k3sctlAdapterMicroserviceResponse(
return fetchMicroserviceUpstreamResponse(adapter, method, adapterTargetPath, proxyOptions, requestHeaders, bodyText, abortSignal);
}
async function k3sctlManagedDiagnosticsResponse(service: MicroserviceConfig): Promise<Response> {
const adapterServiceId = service.deployment.adapterServiceId ?? "k3sctl-adapter";
const adapter = microserviceById(adapterServiceId);
const checkedAt = new Date().toISOString();
const providerId = adapter?.providerId ?? service.providerId;
const providerOnline = ctx.activeProviders.has(providerId);
const providerTunnelCapable = await providerSupports(providerId, "microservice.http.tunnel");
if (adapter === null) {
return jsonResponse({
ok: false,
serviceId: service.id,
checkedAt,
requestPath: "/diagnostics",
checks: {
providerGateway: { ok: providerOnline, providerId, online: providerOnline },
httpTunnel: { ok: providerTunnelCapable, providerId, capable: providerTunnelCapable },
k3sctlAdapter: { ok: false, serviceId: adapterServiceId, error: `k3sctl adapter microservice not found: ${adapterServiceId}` },
kubernetesApiServiceProxy: { ok: false, skipped: true },
targetService: { ok: false, skipped: true },
},
}, 502);
}
const k3sServiceId = service.id === "code-queue"
? codeQueueK3sServiceIdForRequest("GET", service.backend.healthPath)
: service.deployment.k3sServiceId ?? service.id;
const adapterPath = `/api/services/${encodeURIComponent(k3sServiceId)}/diagnostics`;
const response = await fetchMicroserviceUpstreamResponse(
adapter,
"GET",
adapterPath,
{ query: "", jsonArrayLimits: {} },
{ accept: "application/json" },
"",
);
const contentType = response.headers.get("content-type") ?? "application/json; charset=utf-8";
const bodyText = await response.text();
let adapterBody: JsonValue = bodyText;
try {
adapterBody = JSON.parse(bodyText) as JsonValue;
} catch {
adapterBody = bodyText.slice(0, 4000);
}
const bodyRecord = isPlainRecord(adapterBody) ? adapterBody : {};
const adapterChecks = isPlainRecord(bodyRecord.checks) ? bodyRecord.checks : {};
const checks = {
providerGateway: {
ok: providerOnline,
providerId,
online: providerOnline,
activeSocketCount: ctx.activeProviders.size,
},
httpTunnel: {
ok: response.ok && response.headers.get("x-unidesk-proxy-mode") === "provider-ws-http-tunnel",
providerId,
capable: providerTunnelCapable,
requestId: response.headers.get("x-unidesk-request-id") ?? null,
attempts: response.headers.get("x-unidesk-http-tunnel-attempts") ?? null,
upstreamProxyMode: response.headers.get("x-unidesk-upstream-proxy-mode") ?? null,
proxyStatus: response.status,
},
k3sctlAdapter: {
ok: response.ok,
serviceId: adapter.id,
providerId: adapter.providerId,
status: response.status,
contentType,
},
kubernetesApiServiceProxy: compactJson(adapterChecks.kubernetesApiServiceProxy ?? { ok: false, skipped: true }),
targetService: compactJson(adapterChecks.targetService ?? adapterChecks.managedService ?? { ok: false, skipped: true }),
} satisfies Record<string, JsonValue>;
const httpTunnelCheck = checks.httpTunnel as Record<string, JsonValue>;
return jsonResponse({
ok: response.ok && providerOnline && httpTunnelCheck.ok === true,
serviceId: service.id,
k3sServiceId,
checkedAt,
path: service.backend.healthPath,
chain: "CLI/frontend -> backend-core -> provider-gateway HTTP tunnel -> k3sctl-adapter -> Kubernetes API service proxy -> k3s Service",
checks,
adapter: adapterBody,
}, response.ok ? 200 : response.status);
}
async function microserviceTunnelSelfTestResponse(service: MicroserviceConfig): Promise<Response> {
const tunnelService = isK3sctlManagedMicroservice(service)
? microserviceById(service.deployment.adapterServiceId ?? "k3sctl-adapter")
: service;
if (tunnelService === null) {
return jsonResponse({ ok: false, serviceId: service.id, error: "tunnel service not found", adapterServiceId: service.deployment.adapterServiceId ?? null }, 502);
}
if (!(await providerSupports(tunnelService.providerId, "microservice.http.tunnel"))) {
return jsonResponse({
ok: false,
serviceId: service.id,
providerId: tunnelService.providerId,
error: `provider does not declare microservice.http.tunnel capability: ${tunnelService.providerId}`,
}, 409);
}
const probeService = {
...tunnelService,
backend: {
...tunnelService.backend,
nodeBaseUrl: "http://127.0.0.1:1",
timeoutMs: 1000,
},
};
const response = await providerHttpTunnelMicroserviceResponse(
probeService,
"GET",
"/",
{ query: "", jsonArrayLimits: {} },
{ accept: "application/json" },
"",
);
const headers = {
requestId: response.headers.get("x-unidesk-request-id"),
tunnelError: response.headers.get("x-unidesk-tunnel-error"),
providerId: response.headers.get("x-unidesk-provider-id"),
serviceId: response.headers.get("x-unidesk-service-id"),
transient: response.headers.get("x-unidesk-transient-error"),
};
const bodyText = await response.text();
let body: JsonValue = bodyText.slice(0, 4000);
try {
body = JSON.parse(bodyText) as JsonValue;
} catch {
// Keep bounded text body for malformed JSON diagnostics.
}
const bodyRecord = isPlainRecord(body) ? body : {};
const hasRequestId = typeof bodyRecord.requestId === "string" && bodyRecord.requestId.length > 0;
const hasStage = typeof bodyRecord.stage === "string" && bodyRecord.stage.length > 0;
const ok = response.status === 502 && hasRequestId && hasStage && headers.requestId === bodyRecord.requestId;
return jsonResponse({
ok,
serviceId: service.id,
tunnelServiceId: tunnelService.id,
providerId: tunnelService.providerId,
expectedFailure: true,
status: response.status,
checks: {
expectedStatus: response.status === 502,
bodyHasRequestId: hasRequestId,
bodyHasStage: hasStage,
headerHasRequestId: typeof headers.requestId === "string" && headers.requestId.length > 0,
headerHasTunnelError: typeof headers.tunnelError === "string" && headers.tunnelError.length > 0,
},
headers,
body,
}, ok ? 200 : 502);
}
function providerHttpTunnelRequestId(providerId: string): string {
return `${providerId}:http_${Date.now()}_${Math.random().toString(16).slice(2)}`;
}
function canRetryProviderHttpTunnel(method: string, targetPath: string): boolean {
const normalizedMethod = method.toUpperCase();
if (normalizedMethod !== "GET" && normalizedMethod !== "HEAD") return false;
return !targetPath.endsWith("/stream");
}
function providerHttpTunnelWaitMs(service: MicroserviceConfig, attempt: number, retryable: boolean): number {
const baseTimeoutMs = Math.max(1000, service.backend.timeoutMs);
if (!retryable) return baseTimeoutMs + 3000;
if (attempt === 1) return Math.min(baseTimeoutMs + 3000, Math.max(5000, Math.floor(baseTimeoutMs * 0.45)));
if (attempt === 2) return Math.min(baseTimeoutMs + 3000, Math.max(6000, Math.floor(baseTimeoutMs * 0.7)));
return baseTimeoutMs + 3000;
}
function tunnelErrorBody(
service: MicroserviceConfig,
requestId: string,
error: string,
stage: string,
status: number,
extra: Record<string, JsonValue> = {},
): Response {
const response = jsonResponse({
ok: false,
error,
stage,
providerId: service.providerId,
serviceId: service.id,
requestId,
...extra,
}, status);
response.headers.set("x-unidesk-request-id", requestId);
response.headers.set("x-unidesk-provider-id", service.providerId);
response.headers.set("x-unidesk-service-id", service.id);
response.headers.set("x-unidesk-tunnel-error", stage);
if (status === 502 || status === 503 || status === 504) response.headers.set("x-unidesk-transient-error", "true");
return response;
}
function providerHttpTunnelFailureStatus(reason: HttpTunnelFailureReason | null): number {
if (reason === "aborted") return 499;
if (reason === "provider-disconnected") return 503;
if (reason === "send-failed") return 502;
return 504;
}
function tunnelFailureRetryable(reason: HttpTunnelFailureReason | null): boolean {
return reason === "timeout" || reason === "provider-disconnected" || reason === "send-failed";
}
async function waitForProviderHttpTunnelResponse(
providerId: string,
requestId: string,
timeoutMs: number,
abortSignal?: AbortSignal,
): Promise<{ providerId: string; requestId: string; ok: boolean; result: JsonValue } | null> {
): Promise<{ message: { providerId: string; requestId: string; ok: boolean; result: JsonValue } | null; reason: HttpTunnelFailureReason | null }> {
return await new Promise((resolve) => {
let settled = false;
let abortHandler: (() => void) | null = null;
const timer = setTimeout(() => settle(null), Math.max(1, timeoutMs));
const settle = (message: { providerId: string; requestId: string; ok: boolean; result: JsonValue } | null): void => {
const timer = setTimeout(() => settle(null, "timeout"), Math.max(1, timeoutMs));
const settle = (
message: { providerId: string; requestId: string; ok: boolean; result: JsonValue } | null,
reason: HttpTunnelFailureReason | null = null,
): void => {
if (settled) return;
settled = true;
clearTimeout(timer);
if (abortHandler !== null) abortSignal?.removeEventListener("abort", abortHandler);
ctx.httpTunnelWaiters.delete(requestId);
resolve(message);
resolve({ message, reason });
};
abortHandler = () => settle(null);
abortHandler = () => settle(null, "aborted");
if (abortSignal !== undefined) {
if (abortSignal.aborted) {
settle(null);
settle(null, "aborted");
return;
}
abortSignal.addEventListener("abort", abortHandler, { once: true });
}
ctx.httpTunnelWaiters.set(requestId, (message) => {
ctx.httpTunnelWaiters.set(requestId, (message, reason) => {
if (message !== null && message.providerId !== providerId) {
logger("warn", "http_tunnel_provider_mismatch", { requestId, expectedProviderId: providerId, actualProviderId: message.providerId });
settle(null);
settle(null, "provider-mismatch");
return;
}
settle(message);
settle(message, reason ?? null);
});
});
}
@@ -676,32 +889,116 @@ async function providerHttpTunnelMicroserviceResponse(
bodyText: string,
abortSignal?: AbortSignal,
): Promise<Response> {
const socket = ctx.activeProviders.get(service.providerId);
if (socket === undefined) return jsonResponse({ ok: false, error: `provider is offline: ${service.providerId}` }, 503);
const requestId = providerHttpTunnelRequestId(service.providerId);
const timeoutMs = service.backend.timeoutMs + 3000;
const waiter = waitForProviderHttpTunnelResponse(service.providerId, requestId, timeoutMs, abortSignal);
socket.send(JSON.stringify({
type: "http_tunnel_request",
requestId,
payload: {
source: "microservice-frontend-proxy",
serviceId: service.id,
method,
targetBaseUrl: service.backend.nodeBaseUrl,
path: targetPath,
query: proxyOptions.query,
requestHeaders,
bodyText,
jsonArrayLimits: proxyOptions.jsonArrayLimits,
timeoutMs: service.backend.timeoutMs,
cacheTtlMs: providerMicroserviceCacheTtlMs(service.id, targetPath),
},
}));
const message = await waiter;
if (message === null) return jsonResponse({ ok: false, error: "provider HTTP tunnel timed out or disconnected", providerId: service.providerId, requestId }, 504);
if (!message.ok) return jsonResponse({ ok: false, error: "provider HTTP tunnel failed", providerId: service.providerId, requestId, result: message.result }, 502);
return responseFromProviderMicroserviceResult(dockerStatusRecord(message.result), "provider-ws-http-tunnel");
const retryable = canRetryProviderHttpTunnel(method, targetPath);
const maxAttempts = retryable ? providerHttpTunnelMaxAttempts : 1;
const attempts: JsonValue[] = [];
let lastRequestId = "";
for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
const socket = ctx.activeProviders.get(service.providerId);
const requestId = providerHttpTunnelRequestId(service.providerId);
lastRequestId = requestId;
if (socket === undefined) {
attempts.push({ attempt, requestId, ok: false, reason: "provider-offline" });
return tunnelErrorBody(service, requestId, `provider is offline: ${service.providerId}`, "provider-gateway-online", 503, {
retryable,
attempts,
});
}
const timeoutMs = providerHttpTunnelWaitMs(service, attempt, retryable);
const startedAt = Date.now();
const waiter = waitForProviderHttpTunnelResponse(service.providerId, requestId, timeoutMs, abortSignal);
try {
socket.send(JSON.stringify({
type: "http_tunnel_request",
requestId,
payload: {
source: "microservice-frontend-proxy",
serviceId: service.id,
method,
targetBaseUrl: service.backend.nodeBaseUrl,
path: targetPath,
query: proxyOptions.query,
requestHeaders,
bodyText,
jsonArrayLimits: proxyOptions.jsonArrayLimits,
timeoutMs: service.backend.timeoutMs,
cacheTtlMs: providerMicroserviceCacheTtlMs(service.id, targetPath),
},
}));
} catch (error) {
ctx.httpTunnelWaiters.get(requestId)?.(null, "send-failed");
const durationMs = Date.now() - startedAt;
attempts.push({ attempt, requestId, ok: false, reason: "send-failed", durationMs, error: errorToJson(error) });
if (attempt < maxAttempts) {
logger("warn", "http_tunnel_send_retry", { providerId: service.providerId, serviceId: service.id, requestId, attempt, maxAttempts, error: errorToJson(error) });
await Bun.sleep(Math.min(500, 75 * attempt));
continue;
}
return tunnelErrorBody(service, requestId, "provider HTTP tunnel send failed", "http-tunnel-send", 502, {
retryable,
attempts,
detail: errorToJson(error),
});
}
const { message, reason } = await waiter;
const durationMs = Date.now() - startedAt;
if (message === null) {
attempts.push({ attempt, requestId, ok: false, reason: reason ?? "timeout", durationMs, timeoutMs });
if (retryable && tunnelFailureRetryable(reason) && attempt < maxAttempts) {
logger("warn", "http_tunnel_retry", {
providerId: service.providerId,
serviceId: service.id,
requestId,
attempt,
maxAttempts,
reason: reason ?? "timeout",
durationMs,
timeoutMs,
});
await Bun.sleep(Math.min(750, 100 * attempt));
continue;
}
return tunnelErrorBody(
service,
requestId,
"provider HTTP tunnel timed out or disconnected",
reason === "provider-disconnected" ? "http-tunnel-provider-disconnected" : reason === "aborted" ? "client-aborted" : "http-tunnel-wait",
providerHttpTunnelFailureStatus(reason),
{ retryable, attempts, timeoutMs, failureReason: reason ?? "timeout" },
);
}
attempts.push({ attempt, requestId, ok: message.ok, durationMs });
if (!message.ok) {
const result = dockerStatusRecord(message.result);
const resultError = typeof result.error === "string" ? result.error : "provider HTTP tunnel failed";
logger("warn", "http_tunnel_provider_error", {
providerId: service.providerId,
serviceId: service.id,
requestId,
attempt,
maxAttempts,
durationMs,
result: compactJson(result),
});
if (retryable && attempt < maxAttempts) {
await Bun.sleep(Math.min(750, 100 * attempt));
continue;
}
return tunnelErrorBody(service, requestId, "provider HTTP tunnel failed", "provider-gateway-http-fetch", 502, {
retryable,
attempts,
result: message.result,
providerError: resultError,
});
}
const response = responseFromProviderMicroserviceResult(dockerStatusRecord(message.result), "provider-ws-http-tunnel");
response.headers.set("x-unidesk-request-id", requestId);
response.headers.set("x-unidesk-http-tunnel-attempt", String(attempt));
response.headers.set("x-unidesk-http-tunnel-attempts", String(attempts.length));
response.headers.set("x-unidesk-provider-id", service.providerId);
return response;
}
return tunnelErrorBody(service, lastRequestId, "provider HTTP tunnel exhausted attempts", "http-tunnel-wait", 504, { retryable, attempts });
}
async function fetchMicroserviceUpstreamResponse(
@@ -901,14 +1198,29 @@ export async function microserviceRoute(req: Request, url: URL): Promise<Respons
? "/"
: suffix.startsWith(`${proxyPrefix}/`)
? `/${suffix.slice(proxyPrefix.length + 1)}`
: suffix === "diagnostics"
? "/diagnostics"
: suffix === "tunnel-self-test"
? "/tunnel-self-test"
: "";
if (targetPath.length === 0) return jsonResponse({ ok: false, error: "microservice route must be /status, /health, or /proxy/<path>" }, 404);
if (targetPath.length === 0) return jsonResponse({ ok: false, error: "microservice route must be /status, /health, /diagnostics, /tunnel-self-test, or /proxy/<path>" }, 404);
if (suffix === "health" && method !== "GET" && method !== "HEAD") {
return jsonResponse({ ok: false, error: "microservice health only supports GET/HEAD" }, 405);
}
if (suffix === "diagnostics" && method !== "GET" && method !== "HEAD") {
return jsonResponse({ ok: false, error: "microservice diagnostics only supports GET/HEAD" }, 405);
}
if (suffix === "tunnel-self-test" && method !== "GET" && method !== "HEAD") {
return jsonResponse({ ok: false, error: "microservice tunnel self-test only supports GET/HEAD" }, 405);
}
if (!isMicroserviceMethodAllowed(service, method)) {
return jsonResponse({ ok: false, error: "microservice method is not allowed", serviceId, method, allowedMethods: service.backend.allowedMethods }, 405);
}
if (suffix === "diagnostics") {
if (!isK3sctlManagedMicroservice(service)) return strictMicroserviceHealthResponse(service, method === "HEAD");
return k3sctlManagedDiagnosticsResponse(service);
}
if (suffix === "tunnel-self-test") return microserviceTunnelSelfTestResponse(service);
if (!isMicroservicePathAllowed(service, targetPath)) {
return jsonResponse({ ok: false, error: "microservice path is not allowed", serviceId, targetPath }, 403);
}
@@ -951,6 +1263,14 @@ export async function microserviceRoute(req: Request, url: URL): Promise<Respons
}
}
const response = await fetchMicroserviceUpstreamResponse(service, method, targetPath, proxyOptions, requestHeaders, bodyText, req.signal);
if ((method === "GET" || method === "HEAD") && isMicroserviceTransientFailureResponse(response)) {
const stale = readStaleMicroserviceCache(cacheKey) ?? readMicroservicePathFallback(service, method, targetPath);
if (stale !== null) {
stale.headers.set("x-unidesk-cache", "stale-on-transient-failure");
stale.headers.set("x-unidesk-stale-reason", String(response.status));
return stale;
}
}
if ((method === "GET" || method === "HEAD") && cacheTtlMs > 0) {
const snapshot = await cacheableResponseSnapshot(response);
rememberMicroserviceCache(cacheKey, cacheTtlMs, snapshot);
+8 -1
View File
@@ -168,6 +168,13 @@ export interface EgressTcpConnection {
provider: ProviderSocket;
}
export type HttpTunnelFailureReason =
| "timeout"
| "aborted"
| "provider-disconnected"
| "provider-mismatch"
| "send-failed";
export interface MicroserviceProxyCacheEntry {
expiresAt: number;
staleExpiresAt: number;
@@ -193,6 +200,6 @@ export type HttpTunnelWaiter = (message: {
requestId: string;
ok: boolean;
result: JsonValue;
} | null) => void;
} | null, reason?: HttpTunnelFailureReason) => void;
export type LoggerFn = (level: "debug" | "info" | "warn" | "error", message: string, data?: JsonValue) => void;