File size: 5,607 Bytes
f778c12 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 | import { randomBytes } from "node:crypto";
import type {
GatewaySuspendPrepareResult,
GatewaySuspendResumeResult,
} from "../../../packages/gateway-protocol/src/index.js";
import { colorize, isRich, theme } from "../../../packages/terminal-core/src/theme.js";
import type { OutputRuntimeEnv } from "../../runtime.js";
import { formatCliCommand } from "../command-format.js";
import type { callGatewayFromCliWithTransport } from "../gateway-rpc.js";
type SuspendRpcOpts = Parameters<typeof callGatewayFromCliWithTransport>[1];
type SuspendRpcCall = (method: string, opts: SuspendRpcOpts, params?: unknown) => Promise<unknown>;
type SuspendCliDeps = {
callGateway: SuspendRpcCall;
runtime: OutputRuntimeEnv;
nowMs?: () => number;
sleep?: (delayMs: number) => Promise<void>;
};
const MIN_SUSPEND_POLL_DELAY_MS = 50;
function parseWaitMs(value: string | number | undefined): number | undefined {
if (value === undefined) {
return undefined;
}
const seconds = typeof value === "number" ? value : Number(value.trim() || Number.NaN);
if (!Number.isFinite(seconds) || seconds < 0) {
throw new Error("--wait must be a non-negative number of seconds");
}
const milliseconds = Math.floor(seconds * 1_000);
if (!Number.isSafeInteger(milliseconds)) {
throw new Error("--wait is too large");
}
return milliseconds;
}
function resolveRequestId(value: string | undefined): string {
if (value === undefined) {
return `cli-${randomBytes(4).toString("hex")}`;
}
const requestId = value.trim();
if (!requestId || requestId.length > 128) {
throw new Error("--request-id must contain 1 to 128 characters");
}
return requestId;
}
function formatBusyResult(
result: Extract<GatewaySuspendPrepareResult, { status: "busy" }>,
): string {
const blockers = result.blockers.map((blocker) => `- ${blocker.message}`);
return [
`Gateway suspension is busy (${result.reason}; ${result.activeCount} active).`,
...(blockers.length > 0 ? ["Blockers:", ...blockers] : []),
].join("\n");
}
export async function runGatewaySuspend(
options: {
rpcOpts: SuspendRpcOpts;
requestId?: string;
waitSeconds?: string | number;
json?: boolean;
},
deps: SuspendCliDeps,
): Promise<void> {
const nowMs = deps.nowMs ?? Date.now;
const sleep =
deps.sleep ??
(async (delayMs: number) =>
await new Promise<void>((resolve) => {
setTimeout(resolve, delayMs);
}));
const requestId = resolveRequestId(options.requestId);
const waitMs = parseWaitMs(options.waitSeconds);
const deadlineMs = waitMs === undefined ? undefined : nowMs() + waitMs;
const maxAttempts = waitMs === undefined ? 1 : Math.ceil(waitMs / MIN_SUSPEND_POLL_DELAY_MS) + 1;
let latest: GatewaySuspendPrepareResult | undefined;
for (let attempt = 0; attempt < maxAttempts; attempt += 1) {
// A sleep can overshoot the deadline; never issue a prepare that could
// suspend the Gateway after the operator's advertised --wait window.
if (attempt > 0 && deadlineMs !== undefined && nowMs() >= deadlineMs) {
break;
}
latest = (await deps.callGateway("gateway.suspend.prepare", options.rpcOpts, {
requestId,
})) as GatewaySuspendPrepareResult;
if (latest.status === "ready") {
if (options.json) {
deps.runtime.writeJson({ ...latest, requestId });
return;
}
const rich = isRich();
deps.runtime.log(colorize(rich, theme.success, "Gateway suspension prepared."));
deps.runtime.log(`${colorize(rich, theme.muted, "Suspension ID:")} ${latest.suspensionId}`);
deps.runtime.log(
`${colorize(rich, theme.muted, "Expires:")} ${new Date(latest.expiresAtMs).toISOString()} (${latest.expiresAtMs} ms)`,
);
const port = options.rpcOpts.localPortOverride;
const command = `openclaw gateway resume ${latest.suspensionId}`;
deps.runtime.log(
`Resume with: ${formatCliCommand(port === undefined ? command : `${command} --port ${port}`)}`,
);
return;
}
if (latest.status === "draining") {
throw new Error("Gateway suspension unexpectedly entered drain mode");
}
if (deadlineMs === undefined) {
if (options.json) {
deps.runtime.writeJson({ ...latest, requestId });
deps.runtime.exit(1);
return;
}
throw new Error(`${formatBusyResult(latest)}\nRetry later or use --wait <seconds>.`);
}
const remainingMs = deadlineMs - nowMs();
if (remainingMs <= 0) {
break;
}
const delayMs = Math.min(remainingMs, Math.max(MIN_SUSPEND_POLL_DELAY_MS, latest.retryAfterMs));
await sleep(delayMs);
}
if (!latest || latest.status !== "busy") {
throw new Error("Gateway suspension polling ended without a result");
}
if (options.json) {
deps.runtime.writeJson({ ...latest, requestId });
deps.runtime.exit(1);
return;
}
throw new Error(`${formatBusyResult(latest)}\nTimed out waiting for the Gateway to become idle.`);
}
export async function runGatewayResume(
options: { rpcOpts: SuspendRpcOpts; suspensionId: string; json?: boolean },
deps: Pick<SuspendCliDeps, "callGateway" | "runtime">,
): Promise<void> {
const result = (await deps.callGateway("gateway.suspend.resume", options.rpcOpts, {
suspensionId: options.suspensionId,
})) as GatewaySuspendResumeResult;
if (options.json) {
deps.runtime.writeJson(result);
return;
}
deps.runtime.log(
result.resumed
? "Gateway resumed."
: "No matching suspension was held (lease already expired or resumed); gateway is running.",
);
}
|