mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-09 15:59:30 +00:00
feat(gateway): add persisted crash-loop breaker and fatal-config exit contract
Gateways that crash-loop under systemd/launchd previously flapped forever with no persisted state and no supervisor signal. The gateway now records every boot outcome in the shared state DB (gateway_boot_lifecycle); three unclean boots within five minutes trip a breaker that boots the gateway in safe mode: full control plane available, channel/provider auto-start suppressed at the channel-manager seam (startup, config hot-reload, secrets.reload) with manual channels.start override, one stability bundle per trip. The breaker re-evaluates each boot and logs recovery when the window drains. Slow shutdowns record forced_stop and never count as crashes; /readyz stays ready and reports suppressed channels; the health monitor treats suppressed accounts as expected-stopped. Fatal invalid-config errors now exit 78 (EX_CONFIG) on both the startup and unhandled-rejection paths, engaging the systemd unit's pre-existing RestartPreventExitStatus=78 so supervisors stop relaunching until the config is fixed.
This commit is contained in:
Ayaan Zaidi
parent
1256aa0a47
commit
a18708c5c1
38 changed files with 1146 additions and 64 deletions
|
|
@ -284,6 +284,8 @@ Do not also let `openclaw doctor --fix` install a user-level gateway service for
|
|||
</Tab>
|
||||
</Tabs>
|
||||
|
||||
Invalid configuration errors exit with code `78`. Linux systemd units use `RestartPreventExitStatus=78` to stop relaunching until the config is fixed. launchd and Windows Task Scheduler do not have an equivalent per-exit-code stop rule, so the Gateway also persists rapid unclean boot history and suppresses channel/provider account auto-start after repeated startup failures. In that safe mode the control plane still starts for inspection and repair, config hot reloads and `secrets.reload` refuse automatic channel restarts, and an explicit operator `channels.start` request can override the suppression.
|
||||
|
||||
## Dev profile quick path
|
||||
|
||||
```bash
|
||||
|
|
|
|||
|
|
@ -126,7 +126,7 @@ describe("gateway-hosted exec approvals", () => {
|
|||
bind: "loopback",
|
||||
auth: { mode: "token", token },
|
||||
controlUiEnabled: false,
|
||||
deferStartupSidecars: true,
|
||||
sidecarStartup: "defer",
|
||||
});
|
||||
cleanup.push(() => server.close());
|
||||
|
||||
|
|
|
|||
|
|
@ -1081,6 +1081,7 @@ describe("runGatewayLoop", () => {
|
|||
const close = vi.fn(async () => {});
|
||||
const startupNeverReturns = new Promise<void>(() => {});
|
||||
const { runtime, exited } = createRuntimeWithExitSignal();
|
||||
const completeBoot = vi.fn();
|
||||
const start = vi.fn(async () => {
|
||||
await startupNeverReturns;
|
||||
return { close };
|
||||
|
|
@ -1090,6 +1091,7 @@ describe("runGatewayLoop", () => {
|
|||
void runGatewayLoop({
|
||||
start: start as unknown as Parameters<typeof runGatewayLoop>[0]["start"],
|
||||
runtime: runtime as unknown as Parameters<typeof runGatewayLoop>[0]["runtime"],
|
||||
completeBoot,
|
||||
});
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
const sigusr1 = captureSignal("SIGUSR1");
|
||||
|
|
@ -1105,6 +1107,10 @@ describe("runGatewayLoop", () => {
|
|||
await vi.advanceTimersByTimeAsync(1);
|
||||
|
||||
await expect(exited).resolves.toBe(1);
|
||||
expect(completeBoot).toHaveBeenCalledWith({
|
||||
outcome: "forced_stop",
|
||||
reason: "gateway.restart_startup_request_timeout",
|
||||
});
|
||||
expect(close).not.toHaveBeenCalled();
|
||||
expect(start).toHaveBeenCalledTimes(1);
|
||||
expect(gatewayLog.error).toHaveBeenCalledWith(
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ import {
|
|||
} from "../../gateway/restart-trace.js";
|
||||
import type { startGatewayServer } from "../../gateway/server.js";
|
||||
import { formatErrorMessage } from "../../infra/errors.js";
|
||||
import type { GatewayBootLifecycleCompletion } from "../../infra/gateway-boot-lifecycle.js";
|
||||
import { acquireGatewayLock } from "../../infra/gateway-lock.js";
|
||||
import { createSubsystemLogger } from "../../logging/subsystem.js";
|
||||
import type { RuntimeEnv } from "../../runtime.js";
|
||||
|
|
@ -106,6 +107,8 @@ export async function runGatewayLoop(params: {
|
|||
lockPort?: number;
|
||||
healthHost?: string;
|
||||
waitForHealthyChild?: (port: number, pid?: number, host?: string) => Promise<boolean>;
|
||||
beginBoot?: (startedAtMs: number) => void | Promise<void>;
|
||||
completeBoot?: (completion: GatewayBootLifecycleCompletion) => void;
|
||||
}) {
|
||||
// macOS/BSD process inspection reports process.title instead of the original
|
||||
// argv. Give the long-running Gateway a verifiable identity for lock readers.
|
||||
|
|
@ -149,6 +152,9 @@ export async function runGatewayLoop(params: {
|
|||
cleanupSignals();
|
||||
params.runtime.exit(code);
|
||||
};
|
||||
const completeForcedStop = (reason: string) => {
|
||||
params.completeBoot?.({ outcome: "forced_stop", reason });
|
||||
};
|
||||
const writeStabilityBundle = async (reason: string, error?: unknown) => {
|
||||
const { writeDiagnosticStabilityBundleForFailureSync } =
|
||||
await loadGatewayLifecycleRuntimeModule();
|
||||
|
|
@ -177,6 +183,10 @@ export async function runGatewayLoop(params: {
|
|||
}
|
||||
};
|
||||
const handleRestartAfterServerClose = async (restartReason?: string) => {
|
||||
params.completeBoot?.({
|
||||
outcome: "planned_restart",
|
||||
reason: restartReason ?? "gateway.restart",
|
||||
});
|
||||
const hadLock = await releaseLockIfHeld();
|
||||
const isUpdateRestart = restartReason === "update.run";
|
||||
const {
|
||||
|
|
@ -323,6 +333,7 @@ export async function runGatewayLoop(params: {
|
|||
restartResolver?.();
|
||||
};
|
||||
const handleStopAfterServerClose = async () => {
|
||||
params.completeBoot?.({ outcome: "clean_stop", reason: "gateway.stop" });
|
||||
await releaseLockIfHeld();
|
||||
exitProcess(0);
|
||||
};
|
||||
|
|
@ -349,6 +360,7 @@ export async function runGatewayLoop(params: {
|
|||
try {
|
||||
await writeStabilityBundle("gateway.restart_startup_request_timeout");
|
||||
} finally {
|
||||
completeForcedStop("gateway.restart_startup_request_timeout");
|
||||
exitProcess(1);
|
||||
}
|
||||
})();
|
||||
|
|
@ -408,6 +420,9 @@ export async function runGatewayLoop(params: {
|
|||
// Keep the in-process watchdog below the supervisor stop budget so this
|
||||
// path wins before launchd/systemd escalates to a hard kill. Exit
|
||||
// non-zero on any timeout so supervised installs restart cleanly.
|
||||
completeForcedStop(
|
||||
isRestart ? "gateway.restart_shutdown_timeout" : "gateway.stop_shutdown_timeout",
|
||||
);
|
||||
exitProcess(1);
|
||||
}
|
||||
})();
|
||||
|
|
@ -846,12 +861,18 @@ export async function runGatewayLoop(params: {
|
|||
for (;;) {
|
||||
await onIteration();
|
||||
restartDrainingMarkPromise = null;
|
||||
startupStartedAt = Date.now();
|
||||
let startupFailedBeforeServerHandle = false;
|
||||
try {
|
||||
await params.beginBoot?.(startupStartedAt);
|
||||
server = await params.start({ startupStartedAt });
|
||||
startupFailedWithoutServerHandle = false;
|
||||
isFirstStart = false;
|
||||
} catch (err) {
|
||||
params.completeBoot?.({
|
||||
outcome: "startup_failed",
|
||||
reason: formatErrorMessage(err).slice(0, 500),
|
||||
});
|
||||
// On initial startup, let the error propagate so the outer handler
|
||||
// can report "Gateway failed to start" and exit non-zero. Only
|
||||
// swallow errors on subsequent in-process restarts to keep the
|
||||
|
|
|
|||
|
|
@ -85,6 +85,27 @@ const writeDiagnosticStabilityBundleForFailureSync = vi.fn((_reason: string, _er
|
|||
message: "wrote stability bundle: /tmp/openclaw-stability.json",
|
||||
path: "/tmp/openclaw-stability.json",
|
||||
}));
|
||||
const bootLifecycle = vi.hoisted(() => ({
|
||||
decisions: [] as Array<{
|
||||
tripped: boolean;
|
||||
uncleanBoots: number;
|
||||
windowMs: number;
|
||||
shouldWriteStabilityBundle: boolean;
|
||||
recovered: boolean;
|
||||
}>,
|
||||
inspect: vi.fn(
|
||||
(_env?: NodeJS.ProcessEnv, _nowMs?: number) =>
|
||||
bootLifecycle.decisions.shift() ?? {
|
||||
tripped: false,
|
||||
uncleanBoots: 0,
|
||||
windowMs: 300_000,
|
||||
shouldWriteStabilityBundle: false,
|
||||
recovered: false,
|
||||
},
|
||||
),
|
||||
record: vi.fn((_env?: NodeJS.ProcessEnv, _nowMs?: number, _reason?: string) => "boot-id"),
|
||||
complete: vi.fn(),
|
||||
}));
|
||||
const controlUiState = vi.hoisted(() => ({
|
||||
root: "/tmp/openclaw-control-ui" as string | null,
|
||||
}));
|
||||
|
|
@ -257,6 +278,17 @@ vi.mock("../../logging/diagnostic-stability-bundle.js", () => ({
|
|||
writeDiagnosticStabilityBundleForFailureSync(reason, error),
|
||||
}));
|
||||
|
||||
vi.mock("../../infra/gateway-boot-lifecycle.js", () => ({
|
||||
GATEWAY_CRASH_LOOP_BREAKER_REASON: "gateway.crash_loop_breaker",
|
||||
GATEWAY_CRASH_LOOP_RECOVERED_REASON: "gateway.crash_loop_recovered",
|
||||
inspectGatewayCrashLoopBreaker: (env?: NodeJS.ProcessEnv, nowMs?: number) =>
|
||||
bootLifecycle.inspect(env, nowMs),
|
||||
recordGatewayBootStart: (env?: NodeJS.ProcessEnv, nowMs?: number, reason?: string) =>
|
||||
bootLifecycle.record(env, nowMs, reason),
|
||||
completeGatewayBootLifecycle: (bootId: string | undefined, completion: unknown) =>
|
||||
bootLifecycle.complete(bootId, completion),
|
||||
}));
|
||||
|
||||
vi.mock("../../logging/subsystem.js", () => ({
|
||||
createSubsystemLogger: () => ({
|
||||
info: (message: string) => {
|
||||
|
|
@ -322,6 +354,10 @@ describe("gateway run option collisions", () => {
|
|||
controlUiState.root = "/tmp/openclaw-control-ui";
|
||||
gatewayLogMessages.length = 0;
|
||||
writeDiagnosticStabilityBundleForFailureSync.mockClear();
|
||||
bootLifecycle.decisions.length = 0;
|
||||
bootLifecycle.inspect.mockClear();
|
||||
bootLifecycle.record.mockClear();
|
||||
bootLifecycle.complete.mockClear();
|
||||
startGatewayServer.mockClear();
|
||||
setGatewayWsLogStyle.mockClear();
|
||||
setVerbose.mockClear();
|
||||
|
|
@ -370,6 +406,7 @@ describe("gateway run option collisions", () => {
|
|||
return callArg(startGatewayServer, index, 1) as {
|
||||
auth?: { mode?: string; token?: string; password?: string };
|
||||
bind?: string;
|
||||
channelAutostartSuppression?: { reason?: string };
|
||||
startupConfigSnapshotRead?: { snapshot?: Record<string, unknown> };
|
||||
startupStartedAt?: number;
|
||||
};
|
||||
|
|
@ -1355,6 +1392,54 @@ describe("gateway run option collisions", () => {
|
|||
expect(secondOptions.startupStartedAt).toBe(2000);
|
||||
});
|
||||
|
||||
it("re-inspects crash-loop breaker state for each boot iteration", async () => {
|
||||
runGatewayLoop.mockImplementationOnce(
|
||||
async ({
|
||||
beginBoot,
|
||||
start,
|
||||
}: {
|
||||
beginBoot?: (startedAtMs: number) => Promise<void> | void;
|
||||
start: GatewayLoopStart;
|
||||
}) => {
|
||||
await beginBoot?.(1000);
|
||||
await start({ startupStartedAt: 1000 });
|
||||
await beginBoot?.(2000);
|
||||
await start({ startupStartedAt: 2000 });
|
||||
},
|
||||
);
|
||||
bootLifecycle.decisions.push(
|
||||
{
|
||||
tripped: true,
|
||||
uncleanBoots: 3,
|
||||
windowMs: 300_000,
|
||||
shouldWriteStabilityBundle: true,
|
||||
recovered: false,
|
||||
},
|
||||
{
|
||||
tripped: false,
|
||||
uncleanBoots: 0,
|
||||
windowMs: 300_000,
|
||||
shouldWriteStabilityBundle: false,
|
||||
recovered: true,
|
||||
},
|
||||
);
|
||||
|
||||
await runGatewayCli(["gateway", "run", "--allow-unconfigured"]);
|
||||
|
||||
expect(bootLifecycle.inspect).toHaveBeenCalledTimes(2);
|
||||
expect(bootLifecycle.inspect.mock.calls.map((call) => call[1])).toEqual([1000, 2000]);
|
||||
expect(bootLifecycle.record.mock.calls.map((call) => call[2])).toEqual([
|
||||
"gateway.crash_loop_breaker",
|
||||
"gateway.crash_loop_recovered",
|
||||
]);
|
||||
expect(writeDiagnosticStabilityBundleForFailureSync).toHaveBeenCalledTimes(1);
|
||||
expect(gatewayStartOptions(0).channelAutostartSuppression).toMatchObject({
|
||||
reason: "crash-loop-breaker",
|
||||
});
|
||||
expect(gatewayStartOptions(1).channelAutostartSuppression).toBeUndefined();
|
||||
expect(gatewayLogMessages.some((message) => message.includes("breaker recovered"))).toBe(true);
|
||||
});
|
||||
|
||||
it("logs when first startup will build missing Control UI assets", async () => {
|
||||
controlUiState.root = null;
|
||||
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ import type {
|
|||
ReadConfigFileSnapshotWithPluginMetadataResult,
|
||||
} from "../../config/config.js";
|
||||
import { ALLOW_OLDER_BINARY_DESTRUCTIVE_ACTIONS_ENV } from "../../config/future-version-guard.js";
|
||||
import { isInvalidConfigError } from "../../config/io.invalid-config.js";
|
||||
import {
|
||||
CONFIG_PATH,
|
||||
normalizeStateDirEnv,
|
||||
|
|
@ -34,6 +35,15 @@ import { setGatewayWsLogStyle } from "../../gateway/ws-logging.js";
|
|||
import { setVerbose } from "../../globals.js";
|
||||
import { isTruthyEnvValue } from "../../infra/env.js";
|
||||
import { formatErrorMessage } from "../../infra/errors.js";
|
||||
import {
|
||||
completeGatewayBootLifecycle,
|
||||
GATEWAY_CRASH_LOOP_BREAKER_REASON,
|
||||
GATEWAY_CRASH_LOOP_RECOVERED_REASON,
|
||||
inspectGatewayCrashLoopBreaker,
|
||||
recordGatewayBootStart,
|
||||
type GatewayCrashLoopBreakerDecision,
|
||||
type GatewayBootLifecycleCompletion,
|
||||
} from "../../infra/gateway-boot-lifecycle.js";
|
||||
import { GatewayLockError } from "../../infra/gateway-lock.js";
|
||||
import { parseStrictPositiveInteger } from "../../infra/parse-finite-number.js";
|
||||
import type { RespawnSupervisor } from "../../infra/supervisor-markers.js";
|
||||
|
|
@ -445,6 +455,10 @@ function resolveGatewayLockErrorExitCode(
|
|||
return isHealthyGatewayLockError(err) ? 0 : 1;
|
||||
}
|
||||
|
||||
function resolveGatewayStartupFailureExitCode(err: unknown): number {
|
||||
return isInvalidConfigError(err) ? EXIT_CONFIG_ERROR : 1;
|
||||
}
|
||||
|
||||
function normalizeGatewayHealthProbeHost(host: string): string {
|
||||
if (host === "0.0.0.0" || host === "::") {
|
||||
return "127.0.0.1";
|
||||
|
|
@ -552,10 +566,13 @@ async function runGatewayLoopWithSupervisedLockRecovery(params: {
|
|||
}
|
||||
}
|
||||
|
||||
async function maybeWriteGatewayStartupFailureBundle(err: unknown): Promise<void> {
|
||||
async function maybeWriteGatewayStartupFailureBundle(
|
||||
err: unknown,
|
||||
reason = "gateway.startup_failed",
|
||||
): Promise<void> {
|
||||
const { writeDiagnosticStabilityBundleForFailureSync } =
|
||||
await import("../../logging/diagnostic-stability-bundle.js");
|
||||
const result = writeDiagnosticStabilityBundleForFailureSync("gateway.startup_failed", err);
|
||||
const result = writeDiagnosticStabilityBundleForFailureSync(reason, err);
|
||||
if ("message" in result) {
|
||||
gatewayLog.warn(result.message);
|
||||
}
|
||||
|
|
@ -704,7 +721,7 @@ export async function runGatewayCommand(opts: GatewayRunOpts, hooks: GatewayRunR
|
|||
const port = portOverride ?? resolveGatewayPort(cfg);
|
||||
if (!Number.isFinite(port) || port <= 0 || port > 65_535) {
|
||||
defaultRuntime.error(formatInvalidConfigPort("gateway.port"));
|
||||
defaultRuntime.exit(1);
|
||||
defaultRuntime.exit(EXIT_CONFIG_ERROR);
|
||||
return;
|
||||
}
|
||||
// Only capture the *explicit* bind value here. The container-aware
|
||||
|
|
@ -944,14 +961,56 @@ export async function runGatewayCommand(opts: GatewayRunOpts, hooks: GatewayRunR
|
|||
gatewayLog.info("starting...");
|
||||
startupTrace.mark("cli.gateway-loop");
|
||||
let startupConfigSnapshotReadForNextStart = startupConfigSnapshotRead;
|
||||
const deferStartupSidecars =
|
||||
const envSidecarStartupMode =
|
||||
isTruthyEnvValue(process.env.OPENCLAW_SKIP_CHANNELS) ||
|
||||
isTruthyEnvValue(process.env.OPENCLAW_SKIP_PROVIDERS);
|
||||
isTruthyEnvValue(process.env.OPENCLAW_SKIP_PROVIDERS)
|
||||
? "defer"
|
||||
: "start";
|
||||
let crashLoopDecision: GatewayCrashLoopBreakerDecision | undefined;
|
||||
let channelAutostartSuppression: { reason: "crash-loop-breaker"; message: string } | undefined;
|
||||
let activeBootId: string | undefined;
|
||||
const beginBoot = async (startedAtMs: number) => {
|
||||
// run-loop calls beginBoot before every startGatewayServer invocation, so
|
||||
// in-process restarts re-evaluate breaker state instead of reusing stale mode.
|
||||
crashLoopDecision = inspectGatewayCrashLoopBreaker(process.env, startedAtMs);
|
||||
const bootStartReason = crashLoopDecision.tripped
|
||||
? crashLoopDecision.shouldWriteStabilityBundle
|
||||
? GATEWAY_CRASH_LOOP_BREAKER_REASON
|
||||
: undefined
|
||||
: crashLoopDecision.recovered
|
||||
? GATEWAY_CRASH_LOOP_RECOVERED_REASON
|
||||
: undefined;
|
||||
activeBootId = recordGatewayBootStart(process.env, startedAtMs, bootStartReason);
|
||||
channelAutostartSuppression = undefined;
|
||||
if (crashLoopDecision.recovered) {
|
||||
gatewayLog.info("gateway restart-loop breaker recovered; channel auto-start restored");
|
||||
}
|
||||
if (!crashLoopDecision.tripped) {
|
||||
return;
|
||||
}
|
||||
const message =
|
||||
`gateway restart-loop breaker tripped: ${crashLoopDecision.uncleanBoots} unclean boot(s) within ${crashLoopDecision.windowMs}ms; ` +
|
||||
"suppressing channel/provider account auto-start. Inspect the stability bundle and fix the startup crash before restarting the service.";
|
||||
channelAutostartSuppression = { reason: "crash-loop-breaker", message };
|
||||
gatewayLog.error(message);
|
||||
if (crashLoopDecision.shouldWriteStabilityBundle) {
|
||||
await maybeWriteGatewayStartupFailureBundle(
|
||||
new Error(message),
|
||||
GATEWAY_CRASH_LOOP_BREAKER_REASON,
|
||||
);
|
||||
}
|
||||
};
|
||||
const completeBoot = (completion: GatewayBootLifecycleCompletion) => {
|
||||
completeGatewayBootLifecycle(activeBootId, completion, process.env);
|
||||
activeBootId = undefined;
|
||||
};
|
||||
const startLoop = async () =>
|
||||
await runGatewayLoop({
|
||||
runtime: defaultRuntime,
|
||||
lockPort: port,
|
||||
healthHost,
|
||||
beginBoot,
|
||||
completeBoot,
|
||||
start: async ({ startupStartedAt } = {}) => {
|
||||
const startupConfigSnapshotReadForThisStart = startupConfigSnapshotReadForNextStart;
|
||||
startupConfigSnapshotReadForNextStart = undefined;
|
||||
|
|
@ -963,7 +1022,8 @@ export async function runGatewayCommand(opts: GatewayRunOpts, hooks: GatewayRunR
|
|||
...(startupConfigSnapshotReadForThisStart
|
||||
? { startupConfigSnapshotRead: startupConfigSnapshotReadForThisStart }
|
||||
: {}),
|
||||
...(deferStartupSidecars ? { deferStartupSidecars: true } : {}),
|
||||
...(envSidecarStartupMode !== "start" ? { sidecarStartup: envSidecarStartupMode } : {}),
|
||||
...(channelAutostartSuppression ? { channelAutostartSuppression } : {}),
|
||||
});
|
||||
},
|
||||
});
|
||||
|
|
@ -1004,13 +1064,14 @@ export async function runGatewayCommand(opts: GatewayRunOpts, hooks: GatewayRunR
|
|||
defaultRuntime.error(
|
||||
`Gateway failed to start: ${formatErrorMessage(err)}. Run ${formatCliCommand("openclaw gateway status --deep")} for diagnostics.`,
|
||||
);
|
||||
defaultRuntime.exit(1);
|
||||
defaultRuntime.exit(resolveGatewayStartupFailureExitCode(err));
|
||||
}
|
||||
}
|
||||
|
||||
export const testing = {
|
||||
normalizeGatewayHealthProbeHost,
|
||||
resolveGatewayLockErrorExitCode,
|
||||
resolveGatewayStartupFailureExitCode,
|
||||
runGatewayLoopWithSupervisedLockRecovery,
|
||||
};
|
||||
export { testing as __testing };
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import {
|
|||
createInvalidConfigError,
|
||||
formatInvalidConfigDetails,
|
||||
formatInvalidConfigLogMessage,
|
||||
isInvalidConfigError,
|
||||
logInvalidConfigOnce,
|
||||
throwInvalidConfig,
|
||||
} from "./io.invalid-config.js";
|
||||
|
|
@ -40,8 +41,14 @@ describe("config io invalid config formatting", () => {
|
|||
};
|
||||
|
||||
expect(err.message).toBe("Invalid config at /tmp/openclaw.json:\n- gateway.port: bad");
|
||||
expect(err.name).toBe("InvalidConfigError");
|
||||
expect(err.code).toBe("INVALID_CONFIG");
|
||||
expect(err.details).toBe("- gateway.port: bad");
|
||||
expect(isInvalidConfigError(err)).toBe(true);
|
||||
expect(
|
||||
isInvalidConfigError(Object.assign(new Error(err.message), { code: "INVALID_CONFIG" })),
|
||||
).toBe(true);
|
||||
expect(isInvalidConfigError(new Error(err.message))).toBe(false);
|
||||
});
|
||||
|
||||
it("logs invalid config details only once per path", () => {
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@
|
|||
* All terminal-facing text is sanitized here so callers can reuse the same failure surface.
|
||||
*/
|
||||
import { sanitizeTerminalText } from "../../packages/terminal-core/src/safe-text.js";
|
||||
import { extractErrorCode } from "../infra/errors.js";
|
||||
|
||||
/** Minimal validation issue shape accepted from schema and mutation validation paths. */
|
||||
type ConfigValidationIssueLike = {
|
||||
|
|
@ -45,11 +46,19 @@ export function logInvalidConfigOnce(params: {
|
|||
export function createInvalidConfigError(configPath: string, details: string): Error {
|
||||
const error = new Error(`Invalid config at ${configPath}:\n${details}`);
|
||||
// Keep metadata non-class-based so cross-module callers can inspect plain Error instances.
|
||||
(error as { code?: string; details?: string }).code = "INVALID_CONFIG";
|
||||
(error as { code?: string; details?: string }).details = details;
|
||||
error.name = "InvalidConfigError";
|
||||
(error as { code?: "INVALID_CONFIG"; details?: string }).code = "INVALID_CONFIG";
|
||||
(error as { code?: "INVALID_CONFIG"; details?: string }).details = details;
|
||||
return error;
|
||||
}
|
||||
|
||||
export function isInvalidConfigError(err: unknown): err is Error & {
|
||||
code: "INVALID_CONFIG";
|
||||
details?: string;
|
||||
} {
|
||||
return extractErrorCode(err) === "INVALID_CONFIG";
|
||||
}
|
||||
|
||||
/** Logs and throws the standard invalid-config error for a validation result. */
|
||||
export function throwInvalidConfig(params: {
|
||||
configPath: string;
|
||||
|
|
|
|||
|
|
@ -14,6 +14,8 @@ function createMockChannelManager(overrides?: Partial<ChannelManager>): ChannelM
|
|||
startChannels: vi.fn(async () => {}),
|
||||
startChannel: vi.fn(async () => {}),
|
||||
stopChannel: vi.fn(async () => {}),
|
||||
setAutostartSuppression: vi.fn(),
|
||||
getAutostartSuppression: vi.fn(() => null),
|
||||
markChannelLoggedOut: vi.fn(),
|
||||
isHealthMonitorEnabled: vi.fn(() => true),
|
||||
isManuallyStopped: vi.fn(() => false),
|
||||
|
|
@ -251,6 +253,38 @@ describe("channel-health-monitor", () => {
|
|||
monitor.stop();
|
||||
});
|
||||
|
||||
it("treats crash-loop suppressed accounts as expected stopped", async () => {
|
||||
let suppressed = true;
|
||||
const suppression = { reason: "crash-loop-breaker" as const, message: "safe mode" };
|
||||
const manager = createSnapshotManager(
|
||||
{
|
||||
discord: {
|
||||
default: managedStoppedAccount("safe mode"),
|
||||
},
|
||||
},
|
||||
{
|
||||
getAutostartSuppression: vi.fn(() => (suppressed ? suppression : null)),
|
||||
},
|
||||
);
|
||||
const monitor = startDefaultMonitor(manager, {
|
||||
checkIntervalMs: 100,
|
||||
cooldownCycles: 0,
|
||||
maxRestartsPerHour: 1,
|
||||
});
|
||||
|
||||
await vi.advanceTimersByTimeAsync(350);
|
||||
|
||||
expect(manager.resetRestartAttempts).not.toHaveBeenCalled();
|
||||
expect(manager.startChannel).not.toHaveBeenCalled();
|
||||
|
||||
suppressed = false;
|
||||
await vi.advanceTimersByTimeAsync(101);
|
||||
|
||||
expect(manager.resetRestartAttempts).toHaveBeenCalledWith("discord", "default");
|
||||
expect(manager.startChannel).toHaveBeenCalledWith("discord", "default");
|
||||
monitor.stop();
|
||||
});
|
||||
|
||||
it("skips disabled channels", async () => {
|
||||
const manager = createSnapshotManager({
|
||||
imessage: {
|
||||
|
|
|
|||
|
|
@ -93,6 +93,7 @@ export function startChannelHealthMonitor(deps: ChannelHealthMonitorDeps): Chann
|
|||
let stopped = false;
|
||||
let checkInFlight = false;
|
||||
let timer: ReturnType<typeof setInterval> | null = null;
|
||||
const suppressedAccounts = new Set<string>();
|
||||
|
||||
const rKey = (channelId: string, accountId: string) => `${channelId}:${accountId}`;
|
||||
|
||||
|
|
@ -113,6 +114,10 @@ export function startChannelHealthMonitor(deps: ChannelHealthMonitorDeps): Chann
|
|||
}
|
||||
|
||||
const snapshot = channelManager.getRuntimeSnapshot();
|
||||
const autostartSuppression = channelManager.getAutostartSuppression();
|
||||
if (!autostartSuppression) {
|
||||
suppressedAccounts.clear();
|
||||
}
|
||||
|
||||
for (const [channelId, accounts] of Object.entries(snapshot.channelAccounts)) {
|
||||
if (!accounts) {
|
||||
|
|
@ -128,6 +133,17 @@ export function startChannelHealthMonitor(deps: ChannelHealthMonitorDeps): Chann
|
|||
if (channelManager.isManuallyStopped(channelId as ChannelId, accountId)) {
|
||||
continue;
|
||||
}
|
||||
const key = rKey(channelId, accountId);
|
||||
if (autostartSuppression) {
|
||||
if (status.running !== true && !suppressedAccounts.has(key)) {
|
||||
log.info?.(
|
||||
`[${channelId}:${accountId}] health-monitor: channel autostart suppressed; treating as expected stopped`,
|
||||
);
|
||||
suppressedAccounts.add(key);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
suppressedAccounts.delete(key);
|
||||
const healthPolicy: ChannelHealthPolicy = {
|
||||
channelId,
|
||||
now,
|
||||
|
|
@ -145,7 +161,6 @@ export function startChannelHealthMonitor(deps: ChannelHealthMonitorDeps): Chann
|
|||
continue;
|
||||
}
|
||||
|
||||
const key = rKey(channelId, accountId);
|
||||
const record = restartRecords.get(key) ?? {
|
||||
lastRestartAt: 0,
|
||||
restartsThisHour: [],
|
||||
|
|
|
|||
|
|
@ -70,7 +70,7 @@ async function startLoopbackTokenGateway(token: string) {
|
|||
bind: "loopback",
|
||||
auth: { mode: "token", token },
|
||||
controlUiEnabled: false,
|
||||
deferStartupSidecars: true,
|
||||
sidecarStartup: "defer",
|
||||
});
|
||||
return { port, server };
|
||||
}
|
||||
|
|
|
|||
|
|
@ -85,7 +85,7 @@ describe("operator approval gateway client runtime token source", () => {
|
|||
bind: "loopback",
|
||||
auth: { mode: "token", token },
|
||||
controlUiEnabled: false,
|
||||
deferStartupSidecars: true,
|
||||
sidecarStartup: "defer",
|
||||
});
|
||||
cleanup.push(() => server.close());
|
||||
|
||||
|
|
|
|||
|
|
@ -138,6 +138,7 @@ type SecretsReloadHarnessParams = {
|
|||
clients?: GatewayAuxHandlerParams["clients"];
|
||||
startChannel?: GatewayAuxHandlerParams["startChannel"];
|
||||
stopChannel?: GatewayAuxHandlerParams["stopChannel"];
|
||||
getChannelAutostartSuppression?: GatewayAuxHandlerParams["getChannelAutostartSuppression"];
|
||||
logChannelsInfo?: GatewayAuxHandlerParams["logChannels"]["info"];
|
||||
respond?: ReturnType<typeof vi.fn>;
|
||||
};
|
||||
|
|
@ -157,6 +158,7 @@ function createSecretsReloadHarness(params: SecretsReloadHarnessParams) {
|
|||
clients: params.clients ?? [],
|
||||
startChannel: params.startChannel ?? (async () => {}),
|
||||
stopChannel: params.stopChannel ?? (async () => {}),
|
||||
getChannelAutostartSuppression: params.getChannelAutostartSuppression,
|
||||
logChannels: { info: params.logChannelsInfo ?? vi.fn() },
|
||||
});
|
||||
|
||||
|
|
@ -201,6 +203,30 @@ afterEach(() => {
|
|||
});
|
||||
|
||||
describe("gateway aux handlers", () => {
|
||||
it("refuses secrets.reload channel restarts while crash-loop safe mode suppresses autostart", async () => {
|
||||
const buildReloadPlan = buildRestartChannelsPlan("slack");
|
||||
activateSnapshot(slackConfig("old-slack-secret"));
|
||||
const activateRuntimeSecrets = mockResolvedSecrets(slackConfig("new-slack-secret"));
|
||||
const { reload, respond, startChannel, stopChannel } =
|
||||
createSecretsReloadHarnessWithChannelMocks({
|
||||
activateRuntimeSecrets,
|
||||
buildReloadPlan,
|
||||
getChannelAutostartSuppression: () => ({
|
||||
reason: "crash-loop-breaker",
|
||||
message: "safe mode",
|
||||
}),
|
||||
});
|
||||
|
||||
await reload();
|
||||
|
||||
expect(stopChannel).not.toHaveBeenCalled();
|
||||
expect(startChannel).not.toHaveBeenCalled();
|
||||
const [okFlag, successPayload, errorPayload] = firstRespondCall(respond);
|
||||
expect(okFlag).toBe(false);
|
||||
expect(successPayload).toBeUndefined();
|
||||
expect(errorPayload?.message ?? "").toBe("secrets.reload failed");
|
||||
});
|
||||
|
||||
it("restarts only channels whose resolved secret-backed config changed on secrets.reload", async () => {
|
||||
const buildReloadPlanCalls: string[][] = [];
|
||||
const buildReloadPlan = (changedPaths: string[]) => {
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ import {
|
|||
} from "./config-reload-plan.js";
|
||||
import { createExecApprovalIosPushDelivery } from "./exec-approval-ios-push.js";
|
||||
import { ExecApprovalManager } from "./exec-approval-manager.js";
|
||||
import type { ChannelAutostartSuppression } from "./server-channels.js";
|
||||
import type { GatewayRequestHandler, GatewayRequestHandlers } from "./server-methods/types.js";
|
||||
import {
|
||||
disconnectStaleSharedGatewayAuthClients,
|
||||
|
|
@ -72,6 +73,7 @@ export function createGatewayAuxHandlers(params: {
|
|||
clients: Iterable<SharedGatewayAuthClient>;
|
||||
startChannel: (name: ChannelKind) => Promise<void>;
|
||||
stopChannel: (name: ChannelKind) => Promise<void>;
|
||||
getChannelAutostartSuppression?: () => ChannelAutostartSuppression | null;
|
||||
logChannels: { info: (msg: string) => void };
|
||||
}) {
|
||||
const execApprovalManager = new ExecApprovalManager();
|
||||
|
|
@ -177,6 +179,11 @@ export function createGatewayAuxHandlers(params: {
|
|||
`secrets.reload requires restarting channels: ${restartChannels.join(", ")}`,
|
||||
);
|
||||
}
|
||||
if (params.getChannelAutostartSuppression?.()) {
|
||||
throw new Error(
|
||||
`secrets.reload requires restarting channels but channel autostart is suppressed by crash-loop breaker: ${restartChannels.join(", ")}`,
|
||||
);
|
||||
}
|
||||
const restartFailures: ChannelKind[] = [];
|
||||
for (const channel of restartChannels) {
|
||||
params.logChannels.info(`restarting ${channel} channel after secrets reload`);
|
||||
|
|
|
|||
|
|
@ -944,6 +944,30 @@ describe("server-channels auto restart", () => {
|
|||
expect((ctx?.log as SubsystemLogger | undefined)?.subsystem).toBe("channels/slack");
|
||||
});
|
||||
|
||||
it("suppresses automatic channel starts while allowing manual starts", async () => {
|
||||
const startAccount = vi.fn(async (_ctx: ChannelGatewayContext<TestAccount>) => {});
|
||||
installTestRegistry(createTestPlugin({ startAccount }));
|
||||
const manager = createManager();
|
||||
|
||||
manager.setAutostartSuppression({
|
||||
reason: "crash-loop-breaker",
|
||||
message: "safe mode",
|
||||
});
|
||||
|
||||
await manager.startChannel("discord", DEFAULT_ACCOUNT_ID);
|
||||
await flushMicrotasks();
|
||||
expect(startAccount).not.toHaveBeenCalled();
|
||||
expect(manager.getRuntimeSnapshot().channelAccounts.discord?.default?.lastError).toBe(
|
||||
"safe mode",
|
||||
);
|
||||
|
||||
await manager.startChannel("discord", DEFAULT_ACCOUNT_ID, { manual: true });
|
||||
await flushMicrotasks();
|
||||
|
||||
expect(startAccount).toHaveBeenCalledTimes(1);
|
||||
expect(manager.getAutostartSuppression()?.reason).toBe("crash-loop-breaker");
|
||||
});
|
||||
|
||||
it("deduplicates concurrent start requests for the same account", async () => {
|
||||
const startupGate = createDeferred();
|
||||
const isConfigured = vi.fn(async () => {
|
||||
|
|
|
|||
|
|
@ -94,6 +94,11 @@ type ChannelHealthMonitorConfig = HealthMonitorConfig & {
|
|||
accounts?: Record<string, HealthMonitorConfig>;
|
||||
};
|
||||
|
||||
export type ChannelAutostartSuppression = {
|
||||
reason: "crash-loop-breaker";
|
||||
message: string;
|
||||
};
|
||||
|
||||
type GatewayStartupTrace = {
|
||||
measure: <T>(name: string, run: () => T | Promise<T>) => Promise<T>;
|
||||
};
|
||||
|
|
@ -208,10 +213,11 @@ type ChannelManagerOptions = {
|
|||
deferStartupAccountStartsUntil?: Promise<void>;
|
||||
};
|
||||
|
||||
type StartChannelOptions = {
|
||||
export type StartChannelOptions = {
|
||||
preserveRestartAttempts?: boolean;
|
||||
preserveManualStop?: boolean;
|
||||
deferAccountStartUntil?: Promise<void>;
|
||||
manual?: boolean;
|
||||
};
|
||||
|
||||
type StopChannelOptions = {
|
||||
|
|
@ -236,8 +242,14 @@ async function waitForDeferredAccountStart(
|
|||
export type ChannelManager = {
|
||||
getRuntimeSnapshot: () => ChannelRuntimeSnapshot;
|
||||
startChannels: () => Promise<void>;
|
||||
startChannel: (channel: ChannelId, accountId?: string) => Promise<void>;
|
||||
startChannel: (
|
||||
channel: ChannelId,
|
||||
accountId?: string,
|
||||
opts?: StartChannelOptions,
|
||||
) => Promise<void>;
|
||||
stopChannel: (channel: ChannelId, accountId?: string, opts?: StopChannelOptions) => Promise<void>;
|
||||
setAutostartSuppression: (suppression: ChannelAutostartSuppression | null) => void;
|
||||
getAutostartSuppression: () => ChannelAutostartSuppression | null;
|
||||
markChannelLoggedOut: (channelId: ChannelId, cleared: boolean, accountId?: string) => void;
|
||||
isManuallyStopped: (channelId: ChannelId, accountId: string) => boolean;
|
||||
resetRestartAttempts: (channelId: ChannelId, accountId: string) => void;
|
||||
|
|
@ -263,6 +275,7 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
|
|||
const manuallyStopped = new Set<string>();
|
||||
const recoveryStopTimedOut = new Set<string>();
|
||||
const recoveryStartRequested = new Set<string>();
|
||||
let autostartSuppression: ChannelAutostartSuppression | null = null;
|
||||
|
||||
const restartKey = (channelId: ChannelId, accountId: string) => `${channelId}:${accountId}`;
|
||||
const ensureChannelLog = (channelId: ChannelId): SubsystemLogger => {
|
||||
|
|
@ -443,6 +456,21 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
|
|||
if (accountIds.length === 0) {
|
||||
return;
|
||||
}
|
||||
if (autostartSuppression && optsValue.manual !== true) {
|
||||
// Safe mode must block every automatic channel start surface; otherwise
|
||||
// config reloads can undo the crash-loop breaker while operators inspect.
|
||||
const suffix = accountId ? ` account ${accountId}` : "";
|
||||
ensureChannelLog(channelId).warn?.(
|
||||
`channel autostart suppressed by crash-loop breaker; refusing automatic start for ${channelId}${suffix}. Use channels.start to override.`,
|
||||
);
|
||||
for (const id of accountIds) {
|
||||
setStoppedRuntime(channelId, id, {
|
||||
restartPending: false,
|
||||
lastError: autostartSuppression.message,
|
||||
});
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
const startup = await runTasksWithConcurrency({
|
||||
limit: CHANNEL_STARTUP_CONCURRENCY,
|
||||
|
|
@ -812,8 +840,12 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
|
|||
}
|
||||
};
|
||||
|
||||
const startChannel = async (channelId: ChannelId, accountId?: string) => {
|
||||
await startChannelInternal(channelId, accountId);
|
||||
const startChannel = async (
|
||||
channelId: ChannelId,
|
||||
accountId?: string,
|
||||
optsValue: StartChannelOptions = {},
|
||||
) => {
|
||||
await startChannelInternal(channelId, accountId, optsValue);
|
||||
};
|
||||
|
||||
const stopChannel = async (
|
||||
|
|
@ -1026,6 +1058,10 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
|
|||
startChannels,
|
||||
startChannel,
|
||||
stopChannel,
|
||||
setAutostartSuppression: (suppression) => {
|
||||
autostartSuppression = suppression;
|
||||
},
|
||||
getAutostartSuppression: () => autostartSuppression,
|
||||
markChannelLoggedOut,
|
||||
isManuallyStopped: isManuallyStoppedFlag,
|
||||
resetRestartAttempts: resetRestartAttemptsForTest,
|
||||
|
|
|
|||
|
|
@ -112,7 +112,7 @@ describe("channelsHandlers channels.start", () => {
|
|||
expect(mocks.applyPluginAutoEnable).toHaveBeenCalledWith({
|
||||
config: {},
|
||||
});
|
||||
expect(startChannel).toHaveBeenCalledWith("whatsapp", "default-account");
|
||||
expect(startChannel).toHaveBeenCalledWith("whatsapp", "default-account", { manual: true });
|
||||
expect(respond).toHaveBeenCalledWith(
|
||||
true,
|
||||
{
|
||||
|
|
@ -127,7 +127,7 @@ describe("channelsHandlers channels.start", () => {
|
|||
it("reports started=false when the channel runtime remains stopped", async () => {
|
||||
const { respond, startChannel } = await runChannelsStart(false);
|
||||
|
||||
expect(startChannel).toHaveBeenCalledWith("whatsapp", "default-account");
|
||||
expect(startChannel).toHaveBeenCalledWith("whatsapp", "default-account", { manual: true });
|
||||
expect(respond).toHaveBeenCalledWith(
|
||||
true,
|
||||
{
|
||||
|
|
|
|||
|
|
@ -293,7 +293,7 @@ export async function startChannelAccount(params: {
|
|||
throw new Error(`Channel ${params.channelId} does not support runtime start`);
|
||||
}
|
||||
const resolvedAccountId = resolveChannelGatewayAccountId(params);
|
||||
await params.context.startChannel(params.channelId, resolvedAccountId);
|
||||
await params.context.startChannel(params.channelId, resolvedAccountId, { manual: true });
|
||||
const runtime = params.context.getRuntimeSnapshot();
|
||||
const started =
|
||||
resolveRuntimeAccountSnapshot({
|
||||
|
|
|
|||
|
|
@ -63,7 +63,7 @@ describe("plugin.approval.request turn-source routing (real gateway)", () => {
|
|||
bind: "loopback",
|
||||
auth: { mode: "token", token },
|
||||
controlUiEnabled: false,
|
||||
deferStartupSidecars: true,
|
||||
sidecarStartup: "defer",
|
||||
});
|
||||
|
||||
// No operator approval client; only a requester with APPROVALS_SCOPE.
|
||||
|
|
|
|||
|
|
@ -143,6 +143,7 @@ export type GatewayRequestContext = {
|
|||
startChannel: (
|
||||
channel: import("../../channels/plugins/types.public.js").ChannelId,
|
||||
accountId?: string,
|
||||
opts?: import("../server-channels.js").StartChannelOptions,
|
||||
) => Promise<void>;
|
||||
stopChannel: (
|
||||
channel: import("../../channels/plugins/types.public.js").ChannelId,
|
||||
|
|
|
|||
|
|
@ -985,6 +985,59 @@ describe("gateway channel hot reload handlers", () => {
|
|||
}
|
||||
}
|
||||
|
||||
it("refuses channel restarts while crash-loop safe mode suppresses autostart", async () => {
|
||||
const logChannels = { info: vi.fn(), error: vi.fn() };
|
||||
const channels = {
|
||||
start: vi.fn(async () => {}),
|
||||
stop: vi.fn(async () => {}),
|
||||
};
|
||||
const { applyHotReload } = createGatewayReloadHandlers({
|
||||
deps: {} as never,
|
||||
broadcast: vi.fn(),
|
||||
getState: () => ({
|
||||
hooksConfig: {} as never,
|
||||
hookClientIpConfig: {} as never,
|
||||
heartbeatRunner: { stop: vi.fn(), updateConfig: vi.fn() } as never,
|
||||
cronState: {
|
||||
cron: { start: vi.fn(async () => {}), stop: vi.fn() },
|
||||
storePath: "/tmp/cron.json",
|
||||
cronEnabled: false,
|
||||
} as never,
|
||||
channelHealthMonitor: null,
|
||||
}),
|
||||
setState: vi.fn(),
|
||||
startChannel: channels.start,
|
||||
stopChannel: channels.stop,
|
||||
getChannelAutostartSuppression: () => ({
|
||||
reason: "crash-loop-breaker",
|
||||
message: "safe mode",
|
||||
}),
|
||||
stopPostReadySidecars: vi.fn(),
|
||||
reloadPlugins: vi.fn(
|
||||
async (): Promise<GatewayPluginReloadResult> => ({
|
||||
restartChannels: new Set(),
|
||||
activeChannels: new Set(),
|
||||
}),
|
||||
),
|
||||
logHooks: { info: vi.fn(), warn: vi.fn(), error: vi.fn() },
|
||||
logChannels,
|
||||
logCron: { error: vi.fn() },
|
||||
logReload: { info: vi.fn(), warn: vi.fn() },
|
||||
createHealthMonitor: () => null,
|
||||
});
|
||||
|
||||
await withChannelReloadsEnabled(() => applyHotReload(createChannelReloadPlan(["discord"]), {}));
|
||||
|
||||
expect(channels.stop).toHaveBeenCalledWith("discord", undefined, { manual: false });
|
||||
expect(channels.start).not.toHaveBeenCalled();
|
||||
expect(logChannels.info).toHaveBeenCalledWith(
|
||||
"stopping discord channel before suppressed hot reload",
|
||||
);
|
||||
expect(logChannels.info).toHaveBeenCalledWith(
|
||||
"channel restart during hot reload suppressed by crash-loop breaker for channels: discord",
|
||||
);
|
||||
});
|
||||
|
||||
it("restarts WhatsApp when the planner receives a selfChatMode change", async () => {
|
||||
const whatsappPlugin = {
|
||||
...createChannelTestPluginBase({ id: "whatsapp" }),
|
||||
|
|
@ -1934,6 +1987,108 @@ describe("gateway plugin hot reload handlers", () => {
|
|||
expect(events).toEqual(["reload:start", "stop", "registry:replace"]);
|
||||
expect(setState).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("stops manually started channels before plugin replacement while autostart is suppressed", async () => {
|
||||
const previousSkipChannels = process.env.OPENCLAW_SKIP_CHANNELS;
|
||||
const previousSkipProviders = process.env.OPENCLAW_SKIP_PROVIDERS;
|
||||
delete process.env.OPENCLAW_SKIP_CHANNELS;
|
||||
delete process.env.OPENCLAW_SKIP_PROVIDERS;
|
||||
const cron = { start: vi.fn(async () => {}), stop: vi.fn() };
|
||||
const heartbeatRunner = {
|
||||
stop: vi.fn(),
|
||||
updateConfig: vi.fn(),
|
||||
};
|
||||
const setState = vi.fn();
|
||||
const logChannels = { info: vi.fn(), error: vi.fn() };
|
||||
const events: string[] = [];
|
||||
const startChannel = vi.fn(async (channel: ChannelKind) => {
|
||||
events.push(`start:${channel}`);
|
||||
});
|
||||
const stopChannel = vi.fn(async (channel: ChannelKind) => {
|
||||
events.push(`stop:${channel}`);
|
||||
});
|
||||
const reloadPlugins = vi.fn(
|
||||
async (params: {
|
||||
beforeReplace: (channels: ReadonlySet<ChannelKind>) => Promise<void>;
|
||||
}): Promise<GatewayPluginReloadResult> => {
|
||||
events.push("reload:start");
|
||||
await params.beforeReplace(new Set(["discord"]));
|
||||
events.push("registry:replace");
|
||||
return {
|
||||
restartChannels: new Set(["discord"]),
|
||||
activeChannels: new Set(["discord"]),
|
||||
};
|
||||
},
|
||||
);
|
||||
const { applyHotReload } = createGatewayReloadHandlers({
|
||||
deps: {} as never,
|
||||
broadcast: vi.fn(),
|
||||
getState: () => ({
|
||||
hooksConfig: {} as never,
|
||||
hookClientIpConfig: {} as never,
|
||||
heartbeatRunner: heartbeatRunner as never,
|
||||
cronState: { cron, storePath: "/tmp/cron.json", cronEnabled: false } as never,
|
||||
channelHealthMonitor: null,
|
||||
}),
|
||||
setState,
|
||||
startChannel,
|
||||
stopChannel,
|
||||
reloadPlugins,
|
||||
getChannelAutostartSuppression: () => ({
|
||||
reason: "crash-loop-breaker",
|
||||
message: "safe mode",
|
||||
}),
|
||||
logHooks: { info: vi.fn(), warn: vi.fn(), error: vi.fn() },
|
||||
logChannels,
|
||||
logCron: { error: vi.fn() },
|
||||
logReload: { info: vi.fn(), warn: vi.fn() },
|
||||
createHealthMonitor: () => null,
|
||||
});
|
||||
|
||||
try {
|
||||
await applyHotReload(
|
||||
{
|
||||
changedPaths: ["plugins.enabled"],
|
||||
restartGateway: false,
|
||||
restartReasons: [],
|
||||
hotReasons: ["plugins.enabled"],
|
||||
reloadHooks: false,
|
||||
restartGmailWatcher: false,
|
||||
restartCron: false,
|
||||
restartHeartbeat: false,
|
||||
restartHealthMonitor: false,
|
||||
reloadPlugins: true,
|
||||
restartChannels: new Set(),
|
||||
disposeMcpRuntimes: false,
|
||||
noopPaths: [],
|
||||
},
|
||||
{
|
||||
plugins: {
|
||||
enabled: false,
|
||||
},
|
||||
},
|
||||
);
|
||||
} finally {
|
||||
if (previousSkipChannels === undefined) {
|
||||
delete process.env.OPENCLAW_SKIP_CHANNELS;
|
||||
} else {
|
||||
process.env.OPENCLAW_SKIP_CHANNELS = previousSkipChannels;
|
||||
}
|
||||
if (previousSkipProviders === undefined) {
|
||||
delete process.env.OPENCLAW_SKIP_PROVIDERS;
|
||||
} else {
|
||||
process.env.OPENCLAW_SKIP_PROVIDERS = previousSkipProviders;
|
||||
}
|
||||
}
|
||||
|
||||
expect(stopChannel).toHaveBeenCalledWith("discord", undefined, { manual: false });
|
||||
expect(startChannel).not.toHaveBeenCalled();
|
||||
expect(events).toEqual(["reload:start", "stop:discord", "registry:replace"]);
|
||||
expect(logChannels.info).toHaveBeenCalledWith(
|
||||
"channel restart during hot reload suppressed by crash-loop breaker for channels: discord",
|
||||
);
|
||||
expect(setState).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe("deferred channel reload abort generation", () => {
|
||||
|
|
|
|||
|
|
@ -181,6 +181,7 @@ type GatewayReloadHandlerParams = {
|
|||
setState: (state: GatewayHotReloadState) => void;
|
||||
startChannel: GatewayChannelManager["startChannel"];
|
||||
stopChannel: GatewayChannelManager["stopChannel"];
|
||||
getChannelAutostartSuppression?: GatewayChannelManager["getAutostartSuppression"];
|
||||
stopPostReadySidecars?: () => Promise<void> | void;
|
||||
reloadPlugins: (params: {
|
||||
nextConfig: OpenClawConfig;
|
||||
|
|
@ -386,6 +387,19 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams)
|
|||
const shouldSkipChannelRestart = () =>
|
||||
isTruthyEnvValue(process.env.OPENCLAW_SKIP_CHANNELS) ||
|
||||
isTruthyEnvValue(process.env.OPENCLAW_SKIP_PROVIDERS);
|
||||
const getChannelAutostartSuppression = () => params.getChannelAutostartSuppression?.() ?? null;
|
||||
const logSuppressedChannelRestart = (
|
||||
channels: ReadonlySet<ChannelKind>,
|
||||
action: string,
|
||||
): void => {
|
||||
const suppression = getChannelAutostartSuppression();
|
||||
if (!suppression) {
|
||||
return;
|
||||
}
|
||||
params.logChannels.info(
|
||||
`${action} suppressed by crash-loop breaker for channels: ${[...channels].join(", ")}`,
|
||||
);
|
||||
};
|
||||
if (plan.reloadPlugins) {
|
||||
const stopChannelsBeforePluginReplace = async (channels: ReadonlySet<ChannelKind>) => {
|
||||
for (const channel of channels) {
|
||||
|
|
@ -532,6 +546,44 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams)
|
|||
params.logChannels.info(
|
||||
"skipping channel reload (OPENCLAW_SKIP_CHANNELS=1 or OPENCLAW_SKIP_PROVIDERS=1)",
|
||||
);
|
||||
} else if (getChannelAutostartSuppression()) {
|
||||
let cancelledByRestart = pluginReloadAborted;
|
||||
if (!plan.reloadPlugins && !cancelledByRestart) {
|
||||
cancelledByRestart = await waitForActiveWorkBeforeChannelReload(
|
||||
channelsToRestart,
|
||||
nextConfig,
|
||||
);
|
||||
}
|
||||
if (cancelledByRestart) {
|
||||
params.logChannels.info("channel restart cancelled by in-process restart");
|
||||
} else {
|
||||
const stopFailures = await collectChannelOperationFailures({
|
||||
channels: channelsToRestart,
|
||||
run: async (channel) => {
|
||||
if (plan.reloadPlugins && activePluginChannelsAfterReload?.has(channel) === false) {
|
||||
return;
|
||||
}
|
||||
if (channelsStoppedBeforePluginReload.has(channel)) {
|
||||
return;
|
||||
}
|
||||
params.logChannels.info(`stopping ${channel} channel before suppressed hot reload`);
|
||||
await params.stopChannel(channel, undefined, { manual: false });
|
||||
},
|
||||
onFailure: (channel, err) => {
|
||||
params.logChannels.error(
|
||||
`failed to stop ${channel} channel during suppressed hot reload: ${formatErrorMessage(
|
||||
err,
|
||||
)}`,
|
||||
);
|
||||
},
|
||||
});
|
||||
if (stopFailures.length > 0) {
|
||||
throw new Error(
|
||||
`failed to stop channels during suppressed hot reload: ${stopFailures.join(", ")}`,
|
||||
);
|
||||
}
|
||||
logSuppressedChannelRestart(channelsToRestart, "channel restart during hot reload");
|
||||
}
|
||||
} else {
|
||||
let cancelledByRestart = pluginReloadAborted;
|
||||
if (!plan.reloadPlugins && !cancelledByRestart) {
|
||||
|
|
@ -711,6 +763,7 @@ export function startManagedGatewayConfigReloader(params: ManagedGatewayConfigRe
|
|||
setState: params.setState,
|
||||
startChannel: params.startChannel,
|
||||
stopChannel: params.stopChannel,
|
||||
getChannelAutostartSuppression: params.getChannelAutostartSuppression,
|
||||
stopPostReadySidecars: params.stopPostReadySidecars,
|
||||
reloadPlugins: params.reloadPlugins,
|
||||
logHooks: params.logHooks,
|
||||
|
|
|
|||
|
|
@ -480,7 +480,7 @@ describe("gateway startup config validation", () => {
|
|||
vi.mocked(configIo.readConfigFileSnapshot).mockResolvedValueOnce(invalidSnapshot);
|
||||
|
||||
await expectStartupRejects(
|
||||
`Invalid config at ${configPath}.\ngateway.mode: Expected 'local' or 'remote'\nRun "openclaw doctor --fix" to repair, then retry.\nIf startup is still blocked, inspect the adjacent .bak backup before restoring it manually.`,
|
||||
`Invalid config at ${configPath}:\ngateway.mode: Expected 'local' or 'remote'\nRun "openclaw doctor --fix" to repair, then retry.\nIf startup is still blocked, inspect the adjacent .bak backup before restoring it manually.`,
|
||||
);
|
||||
});
|
||||
|
||||
|
|
@ -507,7 +507,7 @@ describe("gateway startup config validation", () => {
|
|||
|
||||
const start = loadTestStartup({});
|
||||
await expect(start).rejects.toThrow(
|
||||
`Invalid config at ${configPath}.\nplugins.slots.memory: plugin not found: source-only-pack\nThis is a plugin packaging issue, not a local config problem.\nUpdate or reinstall the plugin after the publisher ships compiled JavaScript, or disable/uninstall the plugin until then.`,
|
||||
`Invalid config at ${configPath}:\nplugins.slots.memory: plugin not found: source-only-pack\nThis is a plugin packaging issue, not a local config problem.\nUpdate or reinstall the plugin after the publisher ships compiled JavaScript, or disable/uninstall the plugin until then.`,
|
||||
);
|
||||
await start.catch((error: unknown) => {
|
||||
expect(String(error)).not.toContain("openclaw doctor --fix");
|
||||
|
|
@ -585,7 +585,7 @@ describe("gateway startup config validation", () => {
|
|||
],
|
||||
});
|
||||
vi.mocked(configIo.readConfigFileSnapshot).mockResolvedValueOnce(invalidSnapshot);
|
||||
await expectStartupRejects(`Invalid config at ${configPath}.`);
|
||||
await expectStartupRejects(`Invalid config at ${configPath}:`);
|
||||
});
|
||||
|
||||
it("keeps mixed plugin and core startup invalidity fatal", async () => {
|
||||
|
|
@ -606,7 +606,7 @@ describe("gateway startup config validation", () => {
|
|||
});
|
||||
vi.mocked(configIo.readConfigFileSnapshot).mockResolvedValueOnce(invalidSnapshot);
|
||||
|
||||
await expectStartupRejects(`Invalid config at ${configPath}.`);
|
||||
await expectStartupRejects(`Invalid config at ${configPath}:`);
|
||||
});
|
||||
|
||||
it("rejects stale model provider api enum values during startup", async () => {
|
||||
|
|
@ -655,7 +655,7 @@ describe("gateway startup config validation", () => {
|
|||
],
|
||||
});
|
||||
vi.mocked(configIo.readConfigFileSnapshot).mockResolvedValueOnce(invalidSnapshot);
|
||||
await expectStartupRejects(`Invalid config at ${configPath}.`, false);
|
||||
await expectStartupRejects(`Invalid config at ${configPath}:`, false);
|
||||
|
||||
expect(configMutate.replaceConfigFile).not.toHaveBeenCalled();
|
||||
});
|
||||
|
|
@ -667,6 +667,6 @@ describe("gateway startup config validation", () => {
|
|||
});
|
||||
vi.mocked(configIo.readConfigFileSnapshot).mockResolvedValueOnce(invalidSnapshot);
|
||||
|
||||
await expectStartupRejects(`Invalid config at ${configPath}.`);
|
||||
await expectStartupRejects(`Invalid config at ${configPath}:`);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import {
|
|||
formatInvalidConfigRecoveryHint,
|
||||
formatPluginPackagingRuntimeOutputRecoveryHint,
|
||||
} from "../cli/config-recovery-hints.js";
|
||||
import { createInvalidConfigError } from "../config/io.invalid-config.js";
|
||||
import {
|
||||
type ReadConfigFileSnapshotWithPluginMetadataResult,
|
||||
readConfigFileSnapshotWithPluginMetadata,
|
||||
|
|
@ -120,7 +121,8 @@ export async function loadGatewayStartupConfigSnapshot(params: {
|
|||
const pluginMetadataSnapshot = snapshotRead.pluginMetadataSnapshot;
|
||||
const wroteConfig = false;
|
||||
if (configSnapshot.legacyIssues.length > 0 && isNixMode) {
|
||||
throw new Error(
|
||||
throw createInvalidConfigError(
|
||||
configSnapshot.path,
|
||||
"Legacy config entries detected while running in Nix mode. Update your Nix config to the latest schema and restart.",
|
||||
);
|
||||
}
|
||||
|
|
@ -423,7 +425,7 @@ export function assertValidGatewayStartupConfigSnapshot(
|
|||
: options.includeDoctorHint
|
||||
? `\n${formatInvalidConfigRecoveryHint()}`
|
||||
: "";
|
||||
throw new Error(`Invalid config at ${snapshot.path}.\n${issues}${recoveryHint}`);
|
||||
throw createInvalidConfigError(snapshot.path, `${issues}${recoveryHint}`);
|
||||
}
|
||||
|
||||
/** Prepare the effective Gateway startup config after auth, overrides, and secrets activation. */
|
||||
|
|
|
|||
|
|
@ -1016,7 +1016,7 @@ describe("startGatewayPostAttachRuntime", () => {
|
|||
await startGatewayPostAttachRuntime({
|
||||
...createPostAttachParams(),
|
||||
log,
|
||||
deferSidecars: true,
|
||||
sidecarStartup: "defer",
|
||||
providerAuthPrewarm: { enabled: true, delayMs: 1_000 },
|
||||
onPostReadySidecars,
|
||||
onGatewayLifetimeSidecars,
|
||||
|
|
@ -1054,7 +1054,7 @@ describe("startGatewayPostAttachRuntime", () => {
|
|||
try {
|
||||
await startGatewayPostAttachRuntime({
|
||||
...createPostAttachParams(),
|
||||
deferSidecars: true,
|
||||
sidecarStartup: "defer",
|
||||
providerAuthPrewarm: {},
|
||||
onGatewayLifetimeSidecars,
|
||||
});
|
||||
|
|
@ -1134,7 +1134,7 @@ describe("startGatewayPostAttachRuntime", () => {
|
|||
} as never,
|
||||
}),
|
||||
log,
|
||||
deferSidecars: true,
|
||||
sidecarStartup: "defer",
|
||||
providerAuthPrewarm: { enabled: true, delayMs: 1_000 },
|
||||
onPostReadySidecars,
|
||||
onGatewayLifetimeSidecars,
|
||||
|
|
@ -1377,7 +1377,7 @@ describe("startGatewayPostAttachRuntime", () => {
|
|||
|
||||
await startGatewayPostAttachRuntime({
|
||||
...createPostAttachParams({
|
||||
deferSidecars: true,
|
||||
sidecarStartup: "defer",
|
||||
onChannelsStarted: async () => {
|
||||
events.push("channels-started");
|
||||
},
|
||||
|
|
@ -1431,7 +1431,7 @@ describe("startGatewayPostAttachRuntime", () => {
|
|||
|
||||
await startGatewayPostAttachRuntime({
|
||||
...createPostAttachParams({
|
||||
deferSidecars: true,
|
||||
sidecarStartup: "defer",
|
||||
onPluginServices,
|
||||
onSidecarsReady,
|
||||
}),
|
||||
|
|
@ -1536,7 +1536,7 @@ describe("startGatewayPostAttachRuntime", () => {
|
|||
const runtimePromise = startGatewayPostAttachRuntime(
|
||||
{
|
||||
...createPostAttachParams({
|
||||
deferSidecars: true,
|
||||
sidecarStartup: "defer",
|
||||
onPluginServices,
|
||||
onSidecarsReady,
|
||||
}),
|
||||
|
|
@ -1986,7 +1986,7 @@ describe("startGatewayPostAttachRuntime", () => {
|
|||
{
|
||||
...createPostAttachParams(),
|
||||
unavailableGatewayMethods,
|
||||
deferSidecars: true,
|
||||
sidecarStartup: "defer",
|
||||
},
|
||||
createPostAttachRuntimeDeps({ startGatewaySidecars: startGatewaySidecarsValue }),
|
||||
);
|
||||
|
|
@ -2028,7 +2028,7 @@ describe("startGatewayPostAttachRuntime", () => {
|
|||
await startGatewayPostAttachRuntime(
|
||||
{
|
||||
...createPostAttachParams({
|
||||
deferSidecars: true,
|
||||
sidecarStartup: "defer",
|
||||
loadStartupPlugins,
|
||||
onStartupPluginsLoaded,
|
||||
}),
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@ import { STARTUP_UNAVAILABLE_GATEWAY_METHODS } from "./methods/core-descriptors.
|
|||
import type { refreshLatestUpdateRestartSentinel } from "./server-restart-sentinel.js";
|
||||
import type { logGatewayStartup } from "./server-startup-log.js";
|
||||
import type { startGatewayTailscaleExposure } from "./server-tailscale.js";
|
||||
import type { GatewaySidecarStartupMode } from "./server.impl.js";
|
||||
|
||||
const ACP_BACKEND_READY_TIMEOUT_MS = 5_000;
|
||||
const ACP_BACKEND_READY_POLL_MS = 50;
|
||||
|
|
@ -1111,8 +1112,7 @@ export async function startGatewayPostAttachRuntime(
|
|||
onSidecarsReady?: () => void;
|
||||
isClosing?: () => boolean;
|
||||
startupTrace?: GatewayStartupTrace;
|
||||
deferSidecars?: boolean;
|
||||
logReadyOnSidecars?: boolean;
|
||||
sidecarStartup?: GatewaySidecarStartupMode;
|
||||
providerAuthPrewarm?: {
|
||||
enabled?: boolean;
|
||||
delayMs?: number;
|
||||
|
|
@ -1213,7 +1213,7 @@ export async function startGatewayPostAttachRuntime(
|
|||
};
|
||||
const waitForSidecarStartTurn = () =>
|
||||
new Promise<void>((resolve) => {
|
||||
if (params.deferSidecars === true) {
|
||||
if (params.sidecarStartup === "defer") {
|
||||
// Give startup logging and bind observers a deterministic head start
|
||||
// when tests or callers request deferred sidecar startup.
|
||||
const timer = setTimeout(resolve, DEFERRED_SIDECAR_START_DELAY_MS);
|
||||
|
|
@ -1318,7 +1318,7 @@ export async function startGatewayPostAttachRuntime(
|
|||
["postReadySidecarCount", postReadySidecars.length + gatewayLifetimeSidecars.length],
|
||||
]);
|
||||
params.startupTrace?.mark("sidecars.ready");
|
||||
if (params.logReadyOnSidecars !== false) {
|
||||
if (params.sidecarStartup !== "defer") {
|
||||
params.log.info("gateway ready");
|
||||
}
|
||||
return { ...result, postReadySidecars, gatewayLifetimeSidecars, pluginRegistry };
|
||||
|
|
@ -1364,7 +1364,7 @@ export async function startGatewayPostAttachRuntime(
|
|||
params.log.warn(`gateway sidecars failed to start: ${String(err)}`);
|
||||
});
|
||||
|
||||
if (params.deferSidecars !== true) {
|
||||
if (params.sidecarStartup !== "defer") {
|
||||
const [, tailscaleCleanup, sidecarsResult] = await Promise.all([
|
||||
startupLogPromise,
|
||||
tailscaleCleanupPromise,
|
||||
|
|
|
|||
|
|
@ -93,6 +93,7 @@ import {
|
|||
resumeGatewayRestartTraceFromHandoff,
|
||||
} from "./restart-trace.js";
|
||||
import { resolveGatewayPluginConfig } from "./runtime-plugin-config.js";
|
||||
import type { ChannelAutostartSuppression } from "./server-channels.js";
|
||||
import { resolveGatewayControlUiRootState } from "./server-control-ui-root.js";
|
||||
import { createLazyGatewayCronState } from "./server-cron-lazy.js";
|
||||
import { applyGatewayLaneConcurrency } from "./server-lanes.js";
|
||||
|
|
@ -454,6 +455,8 @@ export type GatewayServer = {
|
|||
close: (opts?: GatewayCloseOptions) => Promise<void>;
|
||||
};
|
||||
|
||||
export type GatewaySidecarStartupMode = "start" | "defer";
|
||||
|
||||
export type GatewayServerOptions = {
|
||||
/**
|
||||
* Bind address policy for the Gateway WebSocket/HTTP server.
|
||||
|
|
@ -499,11 +502,8 @@ export type GatewayServerOptions = {
|
|||
runtime: import("../runtime.js").RuntimeEnv,
|
||||
prompter: import("../wizard/prompts.js").WizardPrompter,
|
||||
) => Promise<void>;
|
||||
/**
|
||||
* Let post-listen sidecars (channels, plugin services) finish in the background.
|
||||
* Defaults to false so gateway startup waits until sidecars are ready.
|
||||
*/
|
||||
deferStartupSidecars?: boolean;
|
||||
sidecarStartup?: GatewaySidecarStartupMode;
|
||||
channelAutostartSuppression?: ChannelAutostartSuppression;
|
||||
/**
|
||||
* Optional startup timestamp used for concise readiness logging.
|
||||
*/
|
||||
|
|
@ -865,8 +865,9 @@ export async function startGatewayServer(
|
|||
startupTrace,
|
||||
deferStartupAccountStartsUntil: startupAccountStartsReady,
|
||||
});
|
||||
const deferStartupSidecars = opts.deferStartupSidecars === true;
|
||||
const isGatewayStartupPending = () => !startupSidecarsReady && !deferStartupSidecars;
|
||||
channelManager.setAutostartSuppression(opts.channelAutostartSuppression ?? null);
|
||||
const sidecarStartup = opts.sidecarStartup ?? "start";
|
||||
const isGatewayStartupPending = () => !startupSidecarsReady && sidecarStartup === "start";
|
||||
const getReadiness = createReadinessChecker({
|
||||
channelManager,
|
||||
startedAt: serverStartedAt,
|
||||
|
|
@ -1217,6 +1218,7 @@ export async function startGatewayServer(
|
|||
clients,
|
||||
startChannel,
|
||||
stopChannel,
|
||||
getChannelAutostartSuppression: channelManager.getAutostartSuppression,
|
||||
logChannels,
|
||||
}),
|
||||
coreGatewayHandlers: coreGatewayHandlersLocal,
|
||||
|
|
@ -1714,8 +1716,7 @@ export async function startGatewayServer(
|
|||
},
|
||||
isClosing: () => closePreludeStarted,
|
||||
startupTrace,
|
||||
deferSidecars: deferStartupSidecars,
|
||||
logReadyOnSidecars: !deferStartupSidecars,
|
||||
sidecarStartup,
|
||||
providerAuthPrewarm: {
|
||||
getConfig: getRuntimeConfig,
|
||||
},
|
||||
|
|
@ -1724,7 +1725,7 @@ export async function startGatewayServer(
|
|||
));
|
||||
startupTrace.detail("memory.ready", collectGatewayProcessMemoryUsageMb());
|
||||
startupTrace.mark("ready");
|
||||
if (deferStartupSidecars) {
|
||||
if (sidecarStartup === "defer") {
|
||||
log.info("gateway ready");
|
||||
}
|
||||
finishGatewayRestartTrace("restart.ready", collectGatewayProcessMemoryUsageMb());
|
||||
|
|
@ -1764,6 +1765,7 @@ export async function startGatewayServer(
|
|||
},
|
||||
startChannel,
|
||||
stopChannel,
|
||||
getChannelAutostartSuppression: channelManager.getAutostartSuppression,
|
||||
stopPostReadySidecars: stopRegisteredPostReadySidecars,
|
||||
reloadPlugins: reloadAttachedGatewayPlugins,
|
||||
logHooks,
|
||||
|
|
|
|||
|
|
@ -32,6 +32,8 @@ function createManager(snapshot: ChannelRuntimeSnapshot): ChannelManager {
|
|||
startChannels: vi.fn(),
|
||||
startChannel: vi.fn(),
|
||||
stopChannel: vi.fn(),
|
||||
setAutostartSuppression: vi.fn(),
|
||||
getAutostartSuppression: vi.fn(() => null),
|
||||
markChannelLoggedOut: vi.fn(),
|
||||
isHealthMonitorEnabled: vi.fn(() => true),
|
||||
isManuallyStopped: vi.fn(() => false),
|
||||
|
|
@ -268,6 +270,25 @@ describe("createReadinessChecker", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("reports crash-loop suppressed stopped channels without failing readiness", () => {
|
||||
withReadinessClock(() => {
|
||||
const { manager, readiness } = createReadinessHarness({
|
||||
accounts: {
|
||||
discord: stoppedAccount({
|
||||
restartPending: false,
|
||||
lastError: "safe mode",
|
||||
}),
|
||||
},
|
||||
});
|
||||
vi.mocked(manager.getAutostartSuppression).mockReturnValue({
|
||||
reason: "crash-loop-breaker",
|
||||
message: "safe mode",
|
||||
});
|
||||
|
||||
expect(readiness()).toEqual(readySnapshot(FIVE_MIN_MS, { suppressed: ["discord"] }));
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps restart-pending channels ready during reconnect backoff", () => {
|
||||
withReadinessClock(() => {
|
||||
const startedAt = Date.now() - FIVE_MIN_MS;
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ import type { GatewayEventLoopHealth } from "./event-loop-health.js";
|
|||
export type ReadinessResult = {
|
||||
ready: boolean;
|
||||
failing: string[];
|
||||
suppressed?: string[];
|
||||
uptimeMs: number;
|
||||
eventLoop?: GatewayEventLoopHealth;
|
||||
};
|
||||
|
|
@ -26,10 +27,14 @@ const DEFAULT_READINESS_CACHE_TTL_MS = 1_000;
|
|||
function shouldIgnoreReadinessFailure(
|
||||
accountSnapshot: ChannelAccountSnapshot,
|
||||
health: ChannelHealthEvaluation,
|
||||
autostartSuppressed: boolean,
|
||||
): boolean {
|
||||
if (health.reason === "unmanaged" || health.reason === "stale-socket") {
|
||||
return true;
|
||||
}
|
||||
if (autostartSuppressed && health.reason === "not-running") {
|
||||
return true;
|
||||
}
|
||||
// Channel restarts spend time in backoff with running=false before the next
|
||||
// lifecycle re-enters startup grace. Keep readiness green during that handoff
|
||||
// window, but still surface hard failures once restart attempts are exhausted.
|
||||
|
|
@ -76,7 +81,9 @@ export function createReadinessChecker(deps: {
|
|||
}
|
||||
|
||||
const snapshot = channelManager.getRuntimeSnapshot();
|
||||
const autostartSuppressed = channelManager.getAutostartSuppression() !== null;
|
||||
const failing: string[] = [];
|
||||
const suppressed: string[] = [];
|
||||
|
||||
for (const [channelId, accounts] of Object.entries(snapshot.channelAccounts)) {
|
||||
if (!accounts) {
|
||||
|
|
@ -93,7 +100,16 @@ export function createReadinessChecker(deps: {
|
|||
channelId,
|
||||
};
|
||||
const health = evaluateChannelHealth(accountSnapshot, policy);
|
||||
if (!health.healthy && !shouldIgnoreReadinessFailure(accountSnapshot, health)) {
|
||||
if (!health.healthy && autostartSuppressed && health.reason === "not-running") {
|
||||
if (!suppressed.includes(channelId)) {
|
||||
suppressed.push(channelId);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (
|
||||
!health.healthy &&
|
||||
!shouldIgnoreReadinessFailure(accountSnapshot, health, autostartSuppressed)
|
||||
) {
|
||||
failing.push(channelId);
|
||||
break;
|
||||
}
|
||||
|
|
@ -101,7 +117,11 @@ export function createReadinessChecker(deps: {
|
|||
}
|
||||
|
||||
cachedAt = now;
|
||||
cachedState = { ready: failing.length === 0, failing };
|
||||
cachedState = {
|
||||
ready: failing.length === 0,
|
||||
failing,
|
||||
...(suppressed.length > 0 ? { suppressed } : {}),
|
||||
};
|
||||
return withEventLoopHealth({ ...cachedState, uptimeMs }, deps.getEventLoopHealth);
|
||||
};
|
||||
}
|
||||
|
|
|
|||
221
src/infra/gateway-boot-lifecycle.test.ts
Normal file
221
src/infra/gateway-boot-lifecycle.test.ts
Normal file
|
|
@ -0,0 +1,221 @@
|
|||
// Gateway boot lifecycle tests cover restart-loop breaker accounting.
|
||||
import fs from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js";
|
||||
import {
|
||||
closeOpenClawStateDatabaseForTest,
|
||||
openOpenClawStateDatabase,
|
||||
} from "../state/openclaw-state-db.js";
|
||||
import {
|
||||
GATEWAY_BOOT_LIFECYCLE_RETENTION_MS,
|
||||
GATEWAY_BOOT_LOOP_UNCLEAN_THRESHOLD,
|
||||
GATEWAY_BOOT_LOOP_WINDOW_MS,
|
||||
GATEWAY_CRASH_LOOP_BREAKER_REASON,
|
||||
GATEWAY_CRASH_LOOP_RECOVERED_REASON,
|
||||
completeGatewayBootLifecycle,
|
||||
inspectGatewayCrashLoopBreaker,
|
||||
recordGatewayBootStart,
|
||||
type GatewayBootLifecycleOutcome,
|
||||
} from "./gateway-boot-lifecycle.js";
|
||||
import { executeSqliteQuerySync, getNodeSqliteKysely } from "./kysely-sync.js";
|
||||
|
||||
type GatewayBootLifecycleTestDatabase = Pick<OpenClawStateKyselyDatabase, "gateway_boot_lifecycle">;
|
||||
|
||||
afterEach(() => {
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
});
|
||||
|
||||
function createLifecycleDb() {
|
||||
const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-gateway-boot-"));
|
||||
const env = { OPENCLAW_STATE_DIR: stateDir } as NodeJS.ProcessEnv;
|
||||
const { db } = openOpenClawStateDatabase({ env });
|
||||
const kysely = getNodeSqliteKysely<GatewayBootLifecycleTestDatabase>(db);
|
||||
return { env, db, kysely };
|
||||
}
|
||||
|
||||
function insertBootRows(
|
||||
params: ReturnType<typeof createLifecycleDb>,
|
||||
rows: ReadonlyArray<{
|
||||
bootId: string;
|
||||
startedAtMs: number;
|
||||
completedAtMs?: number | null;
|
||||
outcome?: GatewayBootLifecycleOutcome | null;
|
||||
startupReason?: string | null;
|
||||
reason?: string | null;
|
||||
}>,
|
||||
): void {
|
||||
executeSqliteQuerySync(
|
||||
params.db,
|
||||
params.kysely.insertInto("gateway_boot_lifecycle").values(
|
||||
rows.map((row) => ({
|
||||
boot_id: row.bootId,
|
||||
pid: 1,
|
||||
started_at_ms: row.startedAtMs,
|
||||
completed_at_ms: row.completedAtMs ?? null,
|
||||
outcome: row.outcome ?? null,
|
||||
startup_reason: row.startupReason ?? null,
|
||||
reason: row.reason ?? null,
|
||||
})),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
describe("gateway crash-loop breaker", () => {
|
||||
it("trips from the persisted unclean boot count", () => {
|
||||
const db = createLifecycleDb();
|
||||
const nowMs = 1_000_000;
|
||||
const windowStartMs = nowMs - GATEWAY_BOOT_LOOP_WINDOW_MS;
|
||||
|
||||
insertBootRows(db, [
|
||||
{ bootId: "a", startedAtMs: windowStartMs + 1 },
|
||||
{ bootId: "b", startedAtMs: windowStartMs + 2 },
|
||||
{
|
||||
bootId: "c",
|
||||
startedAtMs: windowStartMs - 60_000,
|
||||
completedAtMs: windowStartMs + 3,
|
||||
outcome: "startup_failed",
|
||||
},
|
||||
]);
|
||||
|
||||
const decision = inspectGatewayCrashLoopBreaker(db.env, nowMs);
|
||||
|
||||
expect(decision).toMatchObject({
|
||||
tripped: true,
|
||||
uncleanBoots: GATEWAY_BOOT_LOOP_UNCLEAN_THRESHOLD,
|
||||
shouldWriteStabilityBundle: true,
|
||||
});
|
||||
});
|
||||
|
||||
it("does not count clean, planned, or forced-stop outcomes as unclean", () => {
|
||||
const db = createLifecycleDb();
|
||||
const nowMs = 1_000_000;
|
||||
const windowStartMs = nowMs - GATEWAY_BOOT_LOOP_WINDOW_MS;
|
||||
|
||||
insertBootRows(db, [
|
||||
{ bootId: "open", startedAtMs: windowStartMs + 1 },
|
||||
{
|
||||
bootId: "planned",
|
||||
startedAtMs: windowStartMs + 2,
|
||||
completedAtMs: windowStartMs + 3,
|
||||
outcome: "planned_restart",
|
||||
},
|
||||
{
|
||||
bootId: "clean",
|
||||
startedAtMs: windowStartMs + 4,
|
||||
completedAtMs: windowStartMs + 5,
|
||||
outcome: "clean_stop",
|
||||
},
|
||||
{
|
||||
bootId: "forced",
|
||||
startedAtMs: windowStartMs + 6,
|
||||
completedAtMs: windowStartMs + 7,
|
||||
outcome: "forced_stop",
|
||||
},
|
||||
]);
|
||||
|
||||
const decision = inspectGatewayCrashLoopBreaker(db.env, nowMs);
|
||||
|
||||
expect(decision.tripped).toBe(false);
|
||||
expect(decision.uncleanBoots).toBe(1);
|
||||
});
|
||||
|
||||
it("writes the breaker bundle only on a persisted transition into tripped state", () => {
|
||||
const db = createLifecycleDb();
|
||||
const nowMs = 1_000_000;
|
||||
const windowStartMs = nowMs - GATEWAY_BOOT_LOOP_WINDOW_MS;
|
||||
|
||||
insertBootRows(db, [
|
||||
{
|
||||
bootId: "breaker-marker",
|
||||
startedAtMs: windowStartMs + 1,
|
||||
startupReason: GATEWAY_CRASH_LOOP_BREAKER_REASON,
|
||||
},
|
||||
{ bootId: "a", startedAtMs: windowStartMs + 2 },
|
||||
{ bootId: "b", startedAtMs: windowStartMs + 3 },
|
||||
{ bootId: "c", startedAtMs: windowStartMs + 4 },
|
||||
]);
|
||||
|
||||
const decision = inspectGatewayCrashLoopBreaker(db.env, nowMs);
|
||||
|
||||
expect(decision.tripped).toBe(true);
|
||||
expect(decision.shouldWriteStabilityBundle).toBe(false);
|
||||
});
|
||||
|
||||
it("logs recovery once after the breaker window drains", () => {
|
||||
const db = createLifecycleDb();
|
||||
const nowMs = 1_000_000;
|
||||
|
||||
insertBootRows(db, [
|
||||
{
|
||||
bootId: "breaker-marker",
|
||||
startedAtMs: nowMs - GATEWAY_BOOT_LOOP_WINDOW_MS - 1,
|
||||
startupReason: GATEWAY_CRASH_LOOP_BREAKER_REASON,
|
||||
},
|
||||
]);
|
||||
|
||||
const firstDecision = inspectGatewayCrashLoopBreaker(db.env, nowMs);
|
||||
insertBootRows(db, [
|
||||
{
|
||||
bootId: "recovery-marker",
|
||||
startedAtMs: nowMs,
|
||||
startupReason: GATEWAY_CRASH_LOOP_RECOVERED_REASON,
|
||||
},
|
||||
]);
|
||||
const secondDecision = inspectGatewayCrashLoopBreaker(db.env, nowMs + 1);
|
||||
|
||||
expect(firstDecision).toMatchObject({ tripped: false, recovered: true });
|
||||
expect(secondDecision).toMatchObject({ tripped: false, recovered: false });
|
||||
});
|
||||
|
||||
it("records forced stops without tripping the breaker", () => {
|
||||
const db = createLifecycleDb();
|
||||
const nowMs = 1_000_000;
|
||||
|
||||
for (let index = 0; index < GATEWAY_BOOT_LOOP_UNCLEAN_THRESHOLD; index += 1) {
|
||||
const bootId = recordGatewayBootStart(db.env, nowMs + index);
|
||||
completeGatewayBootLifecycle(
|
||||
bootId,
|
||||
{ outcome: "forced_stop", reason: "gateway.stop_shutdown_timeout" },
|
||||
db.env,
|
||||
nowMs + index + 1,
|
||||
);
|
||||
}
|
||||
|
||||
const decision = inspectGatewayCrashLoopBreaker(
|
||||
db.env,
|
||||
nowMs + GATEWAY_BOOT_LOOP_UNCLEAN_THRESHOLD + 1,
|
||||
);
|
||||
|
||||
expect(decision.tripped).toBe(false);
|
||||
expect(decision.uncleanBoots).toBe(0);
|
||||
});
|
||||
|
||||
it("prunes boot rows older than retention when recording a new boot", () => {
|
||||
const db = createLifecycleDb();
|
||||
const nowMs = 2 * GATEWAY_BOOT_LIFECYCLE_RETENTION_MS;
|
||||
|
||||
insertBootRows(db, [
|
||||
{
|
||||
bootId: "old",
|
||||
startedAtMs: nowMs - GATEWAY_BOOT_LIFECYCLE_RETENTION_MS - 1,
|
||||
},
|
||||
{
|
||||
bootId: "kept",
|
||||
startedAtMs: nowMs - GATEWAY_BOOT_LIFECYCLE_RETENTION_MS,
|
||||
},
|
||||
]);
|
||||
|
||||
recordGatewayBootStart(db.env, nowMs);
|
||||
|
||||
const rows = executeSqliteQuerySync(
|
||||
db.db,
|
||||
db.kysely.selectFrom("gateway_boot_lifecycle").select("boot_id").orderBy("boot_id"),
|
||||
).rows.map((row) => row.boot_id);
|
||||
|
||||
expect(rows).toHaveLength(2);
|
||||
expect(rows).toContain("kept");
|
||||
expect(rows).not.toContain("old");
|
||||
});
|
||||
});
|
||||
195
src/infra/gateway-boot-lifecycle.ts
Normal file
195
src/infra/gateway-boot-lifecycle.ts
Normal file
|
|
@ -0,0 +1,195 @@
|
|||
// Persists gateway boot outcomes for supervisor crash-loop decisions.
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createSubsystemLogger } from "../logging/subsystem.js";
|
||||
import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js";
|
||||
import {
|
||||
openOpenClawStateDatabase,
|
||||
runOpenClawStateWriteTransaction,
|
||||
} from "../state/openclaw-state-db.js";
|
||||
import {
|
||||
executeSqliteQuerySync,
|
||||
executeSqliteQueryTakeFirstSync,
|
||||
getNodeSqliteKysely,
|
||||
} from "./kysely-sync.js";
|
||||
|
||||
// Supervisors usually restart immediately. Three unclean boots in this window
|
||||
// means the gateway should come up without auto-start sidecars so operators
|
||||
// can inspect a stable process instead of a flap.
|
||||
export const GATEWAY_BOOT_LOOP_UNCLEAN_THRESHOLD = 3;
|
||||
export const GATEWAY_BOOT_LOOP_WINDOW_MS = 5 * 60_000;
|
||||
// Keep enough history for operator forensics while bounding one-row-per-boot
|
||||
// growth. Retention must comfortably exceed GATEWAY_BOOT_LOOP_WINDOW_MS.
|
||||
export const GATEWAY_BOOT_LIFECYCLE_RETENTION_MS = 24 * 60 * 60_000;
|
||||
export const GATEWAY_CRASH_LOOP_BREAKER_REASON = "gateway.crash_loop_breaker";
|
||||
export const GATEWAY_CRASH_LOOP_RECOVERED_REASON = "gateway.crash_loop_recovered";
|
||||
|
||||
const gatewayLifecycleLog = createSubsystemLogger("gateway/lifecycle");
|
||||
|
||||
type GatewayBootLifecycleDatabase = Pick<OpenClawStateKyselyDatabase, "gateway_boot_lifecycle">;
|
||||
|
||||
export type GatewayBootLifecycleOutcome =
|
||||
| "clean_stop"
|
||||
| "planned_restart"
|
||||
| "startup_failed"
|
||||
| "forced_stop";
|
||||
|
||||
export type GatewayBootLifecycleCompletion = {
|
||||
outcome: GatewayBootLifecycleOutcome;
|
||||
reason?: string;
|
||||
};
|
||||
|
||||
export type GatewayCrashLoopBreakerDecision = {
|
||||
tripped: boolean;
|
||||
uncleanBoots: number;
|
||||
windowMs: number;
|
||||
shouldWriteStabilityBundle: boolean;
|
||||
recovered: boolean;
|
||||
};
|
||||
|
||||
function buildGatewayCrashLoopBreakerDecision(params: {
|
||||
uncleanBoots: number;
|
||||
windowMs?: number;
|
||||
latestBreakerStartedAtMs?: number | null;
|
||||
latestRecoveryStartedAtMs?: number | null;
|
||||
}): GatewayCrashLoopBreakerDecision {
|
||||
const windowMs = params.windowMs ?? GATEWAY_BOOT_LOOP_WINDOW_MS;
|
||||
const tripped = params.uncleanBoots >= GATEWAY_BOOT_LOOP_UNCLEAN_THRESHOLD;
|
||||
const hasUnrecoveredBreakerMarker =
|
||||
typeof params.latestBreakerStartedAtMs === "number" &&
|
||||
(typeof params.latestRecoveryStartedAtMs !== "number" ||
|
||||
params.latestRecoveryStartedAtMs < params.latestBreakerStartedAtMs);
|
||||
// Recovery waits until the unclean window drains. A clean safe-mode boot
|
||||
// proves the control plane works, not that suppressed channel autostart is safe.
|
||||
return {
|
||||
tripped,
|
||||
uncleanBoots: params.uncleanBoots,
|
||||
windowMs,
|
||||
shouldWriteStabilityBundle: tripped && !hasUnrecoveredBreakerMarker,
|
||||
recovered: !tripped && hasUnrecoveredBreakerMarker,
|
||||
};
|
||||
}
|
||||
|
||||
export function inspectGatewayCrashLoopBreaker(
|
||||
env: NodeJS.ProcessEnv = process.env,
|
||||
nowMs = Date.now(),
|
||||
): GatewayCrashLoopBreakerDecision {
|
||||
try {
|
||||
const { db } = openOpenClawStateDatabase({ env });
|
||||
const kysely = getNodeSqliteKysely<GatewayBootLifecycleDatabase>(db);
|
||||
const windowStartMs = nowMs - GATEWAY_BOOT_LOOP_WINDOW_MS;
|
||||
// Unclean means startup_failed by completion time, or an open boot row
|
||||
// whose process disappeared. forced_stop is operator shutdown pressure,
|
||||
// not a startup crash-loop signal.
|
||||
const uncleanRow = executeSqliteQueryTakeFirstSync(
|
||||
db,
|
||||
kysely
|
||||
.selectFrom("gateway_boot_lifecycle")
|
||||
.select((eb) => eb.fn.countAll<number>().as("count"))
|
||||
.where((eb) =>
|
||||
eb.or([
|
||||
eb.and([eb("completed_at_ms", "is", null), eb("started_at_ms", ">=", windowStartMs)]),
|
||||
eb.and([
|
||||
eb("outcome", "=", "startup_failed"),
|
||||
eb("completed_at_ms", ">=", windowStartMs),
|
||||
]),
|
||||
]),
|
||||
),
|
||||
);
|
||||
const latestBreaker = executeSqliteQueryTakeFirstSync(
|
||||
db,
|
||||
kysely
|
||||
.selectFrom("gateway_boot_lifecycle")
|
||||
.select("started_at_ms as startedAtMs")
|
||||
.where("startup_reason", "=", GATEWAY_CRASH_LOOP_BREAKER_REASON)
|
||||
.orderBy("started_at_ms", "desc")
|
||||
.limit(1),
|
||||
);
|
||||
const latestRecovery = executeSqliteQueryTakeFirstSync(
|
||||
db,
|
||||
kysely
|
||||
.selectFrom("gateway_boot_lifecycle")
|
||||
.select("started_at_ms as startedAtMs")
|
||||
.where("startup_reason", "=", GATEWAY_CRASH_LOOP_RECOVERED_REASON)
|
||||
.orderBy("started_at_ms", "desc")
|
||||
.limit(1),
|
||||
);
|
||||
return buildGatewayCrashLoopBreakerDecision({
|
||||
uncleanBoots: Number(uncleanRow?.count ?? 0),
|
||||
latestBreakerStartedAtMs: latestBreaker?.startedAtMs,
|
||||
latestRecoveryStartedAtMs: latestRecovery?.startedAtMs,
|
||||
});
|
||||
} catch (err) {
|
||||
gatewayLifecycleLog.warn(`crash-loop breaker state unavailable; fail-open: ${String(err)}`);
|
||||
return buildGatewayCrashLoopBreakerDecision({ uncleanBoots: 0 });
|
||||
}
|
||||
}
|
||||
|
||||
export function recordGatewayBootStart(
|
||||
env: NodeJS.ProcessEnv = process.env,
|
||||
nowMs = Date.now(),
|
||||
reason?: string,
|
||||
): string | undefined {
|
||||
const bootId = randomUUID();
|
||||
try {
|
||||
runOpenClawStateWriteTransaction(
|
||||
({ db }) => {
|
||||
const kysely = getNodeSqliteKysely<GatewayBootLifecycleDatabase>(db);
|
||||
executeSqliteQuerySync(
|
||||
db,
|
||||
kysely
|
||||
.deleteFrom("gateway_boot_lifecycle")
|
||||
.where("started_at_ms", "<", nowMs - GATEWAY_BOOT_LIFECYCLE_RETENTION_MS),
|
||||
);
|
||||
executeSqliteQuerySync(
|
||||
db,
|
||||
kysely.insertInto("gateway_boot_lifecycle").values({
|
||||
boot_id: bootId,
|
||||
pid: process.pid,
|
||||
started_at_ms: nowMs,
|
||||
completed_at_ms: null,
|
||||
outcome: null,
|
||||
startup_reason: reason ?? null,
|
||||
reason: null,
|
||||
}),
|
||||
);
|
||||
},
|
||||
{ env },
|
||||
);
|
||||
return bootId;
|
||||
} catch (err) {
|
||||
gatewayLifecycleLog.warn(`failed to persist gateway boot start; fail-open: ${String(err)}`);
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
export function completeGatewayBootLifecycle(
|
||||
bootId: string | undefined,
|
||||
completion: GatewayBootLifecycleCompletion,
|
||||
env: NodeJS.ProcessEnv = process.env,
|
||||
nowMs = Date.now(),
|
||||
): void {
|
||||
if (!bootId) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
runOpenClawStateWriteTransaction(
|
||||
({ db }) => {
|
||||
const kysely = getNodeSqliteKysely<GatewayBootLifecycleDatabase>(db);
|
||||
executeSqliteQuerySync(
|
||||
db,
|
||||
kysely
|
||||
.updateTable("gateway_boot_lifecycle")
|
||||
.set({
|
||||
completed_at_ms: nowMs,
|
||||
outcome: completion.outcome,
|
||||
reason: completion.reason ?? null,
|
||||
})
|
||||
.where("boot_id", "=", bootId),
|
||||
);
|
||||
},
|
||||
{ env },
|
||||
);
|
||||
} catch (err) {
|
||||
gatewayLifecycleLog.warn(`failed to persist gateway boot outcome; fail-open: ${String(err)}`);
|
||||
}
|
||||
}
|
||||
|
|
@ -122,13 +122,18 @@ describe("installUnhandledRejectionHandler - fatal detection", () => {
|
|||
});
|
||||
|
||||
describe("configuration errors", () => {
|
||||
it("exits on configuration error codes", () => {
|
||||
const configurationCases = [
|
||||
{ code: "INVALID_CONFIG", message: "Invalid config" },
|
||||
{ code: "MISSING_API_KEY", message: "Missing API key" },
|
||||
] as const;
|
||||
it("uses exit 78 only for invalid configuration", () => {
|
||||
expectExitCodeFromUnhandled(
|
||||
Object.assign(new Error("Invalid config"), { code: "INVALID_CONFIG" }),
|
||||
[78],
|
||||
"configuration error",
|
||||
);
|
||||
|
||||
for (const { code, message } of configurationCases) {
|
||||
const transientCredentialCases = [
|
||||
{ code: "MISSING_API_KEY", message: "Missing API key" },
|
||||
{ code: "MISSING_CREDENTIALS", message: "Missing credentials" },
|
||||
] as const;
|
||||
for (const { code, message } of transientCredentialCases) {
|
||||
expectExitCodeFromUnhandled(
|
||||
Object.assign(new Error(message), { code }),
|
||||
[1],
|
||||
|
|
|
|||
|
|
@ -48,7 +48,13 @@ const FATAL_ERROR_CODES = new Set([
|
|||
"ERR_WORKER_INITIALIZATION_FAILED",
|
||||
]);
|
||||
|
||||
const CONFIG_ERROR_CODES = new Set(["INVALID_CONFIG", "MISSING_API_KEY", "MISSING_CREDENTIALS"]);
|
||||
const INVALID_CONFIG_ERROR_CODE = "INVALID_CONFIG";
|
||||
const CONFIG_ERROR_CODES = new Set([
|
||||
INVALID_CONFIG_ERROR_CODE,
|
||||
"MISSING_API_KEY",
|
||||
"MISSING_CREDENTIALS",
|
||||
]);
|
||||
const EXIT_CONFIG_ERROR = 78;
|
||||
|
||||
// Network error codes that indicate transient failures (shouldn't crash the gateway)
|
||||
const TRANSIENT_NETWORK_CODES = new Set([
|
||||
|
|
@ -507,12 +513,17 @@ export function isUncaughtExceptionHandled(error: unknown): boolean {
|
|||
}
|
||||
|
||||
export function installUnhandledRejectionHandler(): void {
|
||||
const exitWithTerminalRestore = (reason: string, error?: unknown, hookReason = reason) => {
|
||||
const exitWithTerminalRestore = (
|
||||
reason: string,
|
||||
error?: unknown,
|
||||
hookReason = reason,
|
||||
exitCode = 1,
|
||||
) => {
|
||||
for (const message of runFatalErrorHooks({ reason: hookReason, error })) {
|
||||
console.error("[openclaw]", message);
|
||||
}
|
||||
restoreTerminalState(reason, { resumeStdinIfPaused: false });
|
||||
process.exit(1);
|
||||
process.exit(exitCode);
|
||||
};
|
||||
|
||||
process.on("unhandledRejection", (reason, _promise) => {
|
||||
|
|
@ -535,7 +546,9 @@ export function installUnhandledRejectionHandler(): void {
|
|||
|
||||
if (isConfigError(reason)) {
|
||||
console.error("[openclaw] CONFIGURATION ERROR - requires fix:", formatUncaughtError(reason));
|
||||
exitWithTerminalRestore("configuration error", reason, "configuration_error");
|
||||
const exitCode =
|
||||
extractErrorCodeWithCause(reason) === INVALID_CONFIG_ERROR_CODE ? EXIT_CONFIG_ERROR : 1;
|
||||
exitWithTerminalRestore("configuration error", reason, "configuration_error", exitCode);
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
|
|||
11
src/state/openclaw-state-db.generated.d.ts
vendored
11
src/state/openclaw-state-db.generated.d.ts
vendored
|
|
@ -496,6 +496,16 @@ export interface FlowRuns {
|
|||
wait_json: string | null;
|
||||
}
|
||||
|
||||
export interface GatewayBootLifecycle {
|
||||
boot_id: string;
|
||||
completed_at_ms: number | null;
|
||||
outcome: "clean_stop" | "forced_stop" | "planned_restart" | "startup_failed" | null;
|
||||
pid: number;
|
||||
reason: string | null;
|
||||
started_at_ms: number;
|
||||
startup_reason: string | null;
|
||||
}
|
||||
|
||||
export interface GatewayRestartHandoff {
|
||||
created_at: number;
|
||||
expires_at: number;
|
||||
|
|
@ -1006,6 +1016,7 @@ export interface DB {
|
|||
diagnostic_stability_bundles: DiagnosticStabilityBundles;
|
||||
exec_approvals_config: ExecApprovalsConfig;
|
||||
flow_runs: FlowRuns;
|
||||
gateway_boot_lifecycle: GatewayBootLifecycle;
|
||||
gateway_restart_handoff: GatewayRestartHandoff;
|
||||
gateway_restart_intent: GatewayRestartIntent;
|
||||
gateway_restart_sentinel: GatewayRestartSentinel;
|
||||
|
|
|
|||
|
|
@ -98,6 +98,29 @@ describe("openclaw state database", () => {
|
|||
expect(listOpenFileDescriptorsForPath(databasePath)).toEqual([]);
|
||||
});
|
||||
|
||||
it("adds gateway boot lifecycle startup markers to existing state databases", () => {
|
||||
const stateDir = createTempStateDir();
|
||||
const database = openOpenClawStateDatabase({
|
||||
env: { OPENCLAW_STATE_DIR: stateDir },
|
||||
});
|
||||
const databasePath = database.path;
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
|
||||
const { DatabaseSync } = requireNodeSqlite();
|
||||
const legacyDb = new DatabaseSync(databasePath);
|
||||
legacyDb.exec("ALTER TABLE gateway_boot_lifecycle DROP COLUMN startup_reason");
|
||||
legacyDb.close();
|
||||
|
||||
const reopened = openOpenClawStateDatabase({
|
||||
env: { OPENCLAW_STATE_DIR: stateDir },
|
||||
});
|
||||
const columns = reopened.db
|
||||
.prepare("PRAGMA table_info(gateway_boot_lifecycle)")
|
||||
.all() as Array<{ name?: unknown }>;
|
||||
|
||||
expect(columns.map((column) => column.name)).toContain("startup_reason");
|
||||
});
|
||||
|
||||
it("migrates requester and executor attribution for existing cross-agent tasks", () => {
|
||||
const stateDir = createTempStateDir();
|
||||
const database = openOpenClawStateDatabase({
|
||||
|
|
|
|||
|
|
@ -861,6 +861,7 @@ function ensureAdditiveStateColumns(db: DatabaseSync): void {
|
|||
ensureColumn(db, "gateway_restart_sentinel", "continuation_json TEXT");
|
||||
ensureColumn(db, "gateway_restart_sentinel", "doctor_hint TEXT");
|
||||
ensureColumn(db, "gateway_restart_sentinel", "stats_json TEXT");
|
||||
ensureColumn(db, "gateway_boot_lifecycle", "startup_reason TEXT");
|
||||
runSqliteImmediateTransactionSync(db, () => {
|
||||
const addedTaskRequesterAgentId = ensureColumn(db, "task_runs", "requester_agent_id TEXT");
|
||||
if (addedTaskRequesterAgentId) {
|
||||
|
|
|
|||
|
|
@ -543,6 +543,19 @@ CREATE TABLE IF NOT EXISTS gateway_restart_handoff (
|
|||
CREATE INDEX IF NOT EXISTS idx_gateway_restart_handoff_expiry
|
||||
ON gateway_restart_handoff(expires_at, pid);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS gateway_boot_lifecycle (
|
||||
boot_id TEXT NOT NULL PRIMARY KEY,
|
||||
pid INTEGER NOT NULL,
|
||||
started_at_ms INTEGER NOT NULL,
|
||||
completed_at_ms INTEGER,
|
||||
outcome TEXT,
|
||||
startup_reason TEXT,
|
||||
reason TEXT
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_gateway_boot_lifecycle_started
|
||||
ON gateway_boot_lifecycle(started_at_ms);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS acp_sessions (
|
||||
session_key TEXT NOT NULL PRIMARY KEY,
|
||||
session_id TEXT,
|
||||
|
|
|
|||
|
|
@ -538,6 +538,19 @@ CREATE TABLE IF NOT EXISTS gateway_restart_handoff (
|
|||
CREATE INDEX IF NOT EXISTS idx_gateway_restart_handoff_expiry
|
||||
ON gateway_restart_handoff(expires_at, pid);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS gateway_boot_lifecycle (
|
||||
boot_id TEXT NOT NULL PRIMARY KEY,
|
||||
pid INTEGER NOT NULL,
|
||||
started_at_ms INTEGER NOT NULL,
|
||||
completed_at_ms INTEGER,
|
||||
outcome TEXT,
|
||||
startup_reason TEXT,
|
||||
reason TEXT
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_gateway_boot_lifecycle_started
|
||||
ON gateway_boot_lifecycle(started_at_ms);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS acp_sessions (
|
||||
session_key TEXT NOT NULL PRIMARY KEY,
|
||||
session_id TEXT,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue