@@ -1,4 +1,5 @@
11import fs from "node:fs";
2+import { request } from "node:http";
23import path from "node:path";
34import type { Command } from "commander";
45import { readSecretFromFile } from "../../acp/secret-file.js";
@@ -111,8 +112,11 @@ const GATEWAY_RUN_BOOLEAN_KEYS = [
111112] as const;
112113113114const SUPERVISED_GATEWAY_LOCK_RETRY_MS = 5000;
115+const SUPERVISED_GATEWAY_LOCK_RETRY_TIMEOUT_MS = 30_000;
116+const SUPERVISED_GATEWAY_HEALTH_PROBE_TIMEOUT_MS = 1000;
114117115118type Awaitable<T> = T | Promise<T>;
119+type GatewayRunLogger = Pick<ReturnType<typeof createSubsystemLogger>, "info" | "warn">;
116120117121/**
118122 * EX_CONFIG (78) from sysexits.h — used for configuration errors so systemd
@@ -356,6 +360,107 @@ function isHealthyGatewayLockError(err: unknown): boolean {
356360);
357361}
358362363+function normalizeGatewayHealthProbeHost(host: string): string {
364+if (host === "0.0.0.0" || host === "::") {
365+return "127.0.0.1";
366+}
367+return host;
368+}
369+370+async function probeGatewayHealthz(params: {
371+host: string;
372+port: number;
373+timeoutMs?: number;
374+}): Promise<boolean> {
375+const timeoutMs = params.timeoutMs ?? SUPERVISED_GATEWAY_HEALTH_PROBE_TIMEOUT_MS;
376+return await new Promise<boolean>((resolve) => {
377+const req = request(
378+{
379+hostname: normalizeGatewayHealthProbeHost(params.host),
380+port: params.port,
381+path: "/healthz",
382+method: "GET",
383+timeout: timeoutMs,
384+},
385+(res) => {
386+res.resume();
387+resolve(typeof res.statusCode === "number" && res.statusCode < 500);
388+},
389+);
390+req.once("timeout", () => {
391+req.destroy();
392+resolve(false);
393+});
394+req.once("error", () => {
395+resolve(false);
396+});
397+req.end();
398+});
399+}
400+401+async function runGatewayLoopWithSupervisedLockRecovery(params: {
402+startLoop: () => Promise<void>;
403+supervisor: ReturnType<typeof detectRespawnSupervisor>;
404+port: number;
405+healthHost: string;
406+log: GatewayRunLogger;
407+now?: () => number;
408+sleep?: (ms: number) => Promise<void>;
409+probeHealth?: (params: { host: string; port: number }) => Promise<boolean>;
410+retryMs?: number;
411+timeoutMs?: number;
412+}) {
413+const supervisor = params.supervisor;
414+if (!supervisor) {
415+await params.startLoop();
416+return;
417+}
418+419+const now = params.now ?? Date.now;
420+const sleep =
421+params.sleep ?? (async (ms: number) => await new Promise((resolve) => setTimeout(resolve, ms)));
422+const probeHealth = params.probeHealth ?? ((probeParams) => probeGatewayHealthz(probeParams));
423+const retryMs = params.retryMs ?? SUPERVISED_GATEWAY_LOCK_RETRY_MS;
424+const timeoutMs = params.timeoutMs ?? SUPERVISED_GATEWAY_LOCK_RETRY_TIMEOUT_MS;
425+const startedAt = now();
426+427+for (;;) {
428+try {
429+await params.startLoop();
430+return;
431+} catch (err) {
432+const isGatewayAlreadyRunning =
433+err instanceof GatewayLockError &&
434+typeof err.message === "string" &&
435+err.message.includes("gateway already running");
436+if (!isGatewayAlreadyRunning) {
437+throw err;
438+}
439+440+if (await probeHealth({ host: params.healthHost, port: params.port })) {
441+params.log.info(
442+`gateway already running under ${supervisor}; existing gateway is healthy, leaving it in control`,
443+);
444+return;
445+}
446+447+const elapsedMs = now() - startedAt;
448+if (elapsedMs >= timeoutMs) {
449+throw new GatewayLockError(
450+`gateway already running under ${supervisor}; existing gateway did not become healthy after ${timeoutMs}ms`,
451+err,
452+);
453+}
454+455+const waitMs = Math.min(retryMs, Math.max(0, timeoutMs - elapsedMs));
456+params.log.warn(
457+`gateway already running under ${supervisor}; waiting ${waitMs}ms before retrying startup`,
458+);
459+await sleep(waitMs);
460+}
461+}
462+}
463+359464function maybeWriteGatewayStartupFailureBundle(err: unknown): void {
360465const result = writeDiagnosticStabilityBundleForFailureSync("gateway.startup_failed", err);
361466if ("message" in result) {
@@ -680,11 +785,12 @@ async function runGatewayCommand(opts: GatewayRunOpts) {
680785681786gatewayLog.info("starting...");
682787startupTrace.mark("cli.gateway-loop");
788+const healthHost = await resolveGatewayBindHost(bind, cfg.gateway?.customBindHost);
683789const startLoop = async () =>
684790await runGatewayLoop({
685791runtime: defaultRuntime,
686792lockPort: port,
687-healthHost: await resolveGatewayBindHost(bind, cfg.gateway?.customBindHost),
793+ healthHost,
688794start: async ({ startupStartedAt } = {}) =>
689795await startGatewayServer(port, {
690796 bind,
@@ -695,25 +801,13 @@ async function runGatewayCommand(opts: GatewayRunOpts) {
695801});
696802697803try {
698-const supervisor = detectRespawnSupervisor(process.env);
699-while (true) {
700-try {
701-await startLoop();
702-break;
703-} catch (err) {
704-const isGatewayAlreadyRunning =
705-err instanceof GatewayLockError &&
706-typeof err.message === "string" &&
707-err.message.includes("gateway already running");
708-if (!supervisor || !isGatewayAlreadyRunning) {
709-throw err;
710-}
711-gatewayLog.warn(
712-`gateway already running under ${supervisor}; waiting ${SUPERVISED_GATEWAY_LOCK_RETRY_MS}ms before retrying startup`,
713-);
714-await new Promise((resolve) => setTimeout(resolve, SUPERVISED_GATEWAY_LOCK_RETRY_MS));
715-}
716-}
804+await runGatewayLoopWithSupervisedLockRecovery({
805+ startLoop,
806+supervisor: detectRespawnSupervisor(process.env),
807+ port,
808+ healthHost,
809+log: gatewayLog,
810+});
717811} catch (err) {
718812if (isGatewayLockError(err)) {
719813const errMessage = formatErrorMessage(err);
@@ -740,6 +834,11 @@ async function runGatewayCommand(opts: GatewayRunOpts) {
740834}
741835}
742836837+export const __testing = {
838+ normalizeGatewayHealthProbeHost,
839+ runGatewayLoopWithSupervisedLockRecovery,
840+};
841+743842export function addGatewayRunCommand(cmd: Command): Command {
744843return cmd
745844.option("--port <port>", "Port for the gateway WebSocket")
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。