mirror of
https://github.com/mvschwarz/openrig.git
synced 2026-11-07 05:56:21 -10:00
fix(cli): bound doctor cmux status and topology reads (#2564)
rig doctor bounds the cmux status read and the live rig and node reads behind --spec (2 s each, headers and body), through the existing fetchWithTimeout, so a listener that accepts connections but never answers can no longer hold it. A timeout keeps today's outcome: a cmux warning, or conformance unknown, never a matching topology. This finishes the waits #2549 left. Refs #2531.
This commit is contained in:
1 parent
e0a285968d
commit
ec926c2e2d
2 files changed
+150
-11
No files matched your search
@@ -6,6 +6,7 @@ import { execSync } from "node:child_process";
|
||||
import { resolveDaemonPath } from "../daemon-lifecycle.js";
|
||||
import { DaemonClient, ForeignLocalDaemonError, formatDaemonHostForUrl } from "../client.js";
|
||||
import { ConfigStore } from "../config-store.js";
|
||||
import { fetchWithTimeout } from "../fetch-with-timeout.js";
|
||||
import { buildWritableHomeCheck } from "../system-preflight.js";
|
||||
import {
|
||||
CMUX_SETTINGS_DISCLOSURE_PATH,
|
||||
@@ -45,7 +46,7 @@ export interface DoctorDeps {
|
||||
/** The caller's effective user id, compared with the one the port's daemon reports. Absent: this process's own;
|
||||
* null: this platform has none. */
|
||||
callerUid?: number | null;
|
||||
/** How long each part of the port's /healthz response may take (default 2000 ms). */
|
||||
/** How long each part of /healthz and the cmux status response may take (default 2000 ms). */
|
||||
ownerReportTimeoutMs?: number;
|
||||
/**
|
||||
* Build B — path to a rig spec to check against the RUNNING rig of the same name (`--spec`).
|
||||
@@ -61,6 +62,7 @@ export interface DoctorDeps {
|
||||
}
|
||||
|
||||
const DEFAULT_PORT = 7433;
|
||||
const DOCTOR_HTTP_TIMEOUT_MS = 2000;
|
||||
const CMUX_SOCKET_CONTROL_FIX = 'In cmux Settings > Automation, select "Automation mode" for Socket Control Mode, or set automation.socketControlMode to "automation" in ~/.config/cmux/cmux.json (older versions use ~/.config/cmux/settings.json; current cmux reads that legacy file only for keys absent from cmux.json). cmux reloads saved settings automatically; rerun `rig doctor` to re-probe without restarting the daemon.';
|
||||
|
||||
/** The port's OpenRig daemon, judged by the OS user it reports. /healthz's pid and euid are the daemon's own report,
|
||||
@@ -351,9 +353,16 @@ export function runDoctorChecks(deps: DoctorDeps, providerChecks: DoctorCheck[]
|
||||
} finally { health.abort(); }
|
||||
|
||||
try {
|
||||
const cmuxRes = await fetchFn(`${daemonBase}/api/adapters/cmux/status`);
|
||||
if (cmuxRes.ok && cmuxRes.json) {
|
||||
const data = (await cmuxRes.json()) as { available?: boolean };
|
||||
let data: { available?: boolean } = {};
|
||||
const cmuxRes = await fetchWithTimeout(fetchFn as typeof fetch, `${daemonBase}/api/adapters/cmux/status`, {}, {
|
||||
timeoutMs: deps.ownerReportTimeoutMs ?? DOCTOR_HTTP_TIMEOUT_MS,
|
||||
timeoutMessage: "Doctor's cmux status read timed out.",
|
||||
consumeResponse: async res => {
|
||||
if (res.ok && res.json) data = (await res.json()) as { available?: boolean };
|
||||
else await res.body?.cancel();
|
||||
},
|
||||
});
|
||||
if (cmuxRes.ok) {
|
||||
if (data.available) {
|
||||
return { name: "cmux_daemon", status: "pass", message: "Daemon cmux control available." };
|
||||
}
|
||||
@@ -512,18 +521,32 @@ class AmbiguousLiveRigError extends Error {
|
||||
export async function readLiveLogicalIds(base: string, rigName: string, ownerClient = new DaemonClient(base)): Promise<string[] | null> {
|
||||
try {
|
||||
await ownerClient.assertLocalOwner();
|
||||
const rigsRes = await fetch(`${base}/api/rigs`);
|
||||
let rigs: Array<{ id?: string; name?: string }> = [];
|
||||
const rigsRes = await fetchWithTimeout(fetch, `${base}/api/rigs`, {}, {
|
||||
timeoutMs: DOCTOR_HTTP_TIMEOUT_MS,
|
||||
timeoutMessage: "Doctor's live rig read timed out.",
|
||||
consumeResponse: async res => {
|
||||
if (res.ok) rigs = (await res.json()) as typeof rigs;
|
||||
else await res.body?.cancel();
|
||||
},
|
||||
});
|
||||
if (!rigsRes.ok) return null;
|
||||
const rigs = (await rigsRes.json()) as Array<{ id?: string; name?: string }>;
|
||||
const matches = rigs.filter((r) => r.name === rigName);
|
||||
if (matches.length > 1) throw new AmbiguousLiveRigError(matches.map(r => r.id ?? "(missing id)"));
|
||||
if (matches.length === 0) return null;
|
||||
const rig = matches[0];
|
||||
if (!rig?.id) return null;
|
||||
await ownerClient.assertLocalOwner();
|
||||
const nodesRes = await fetch(`${base}/api/rigs/${rig.id}/nodes`);
|
||||
let nodes: Array<{ logicalId?: string | null }> = [];
|
||||
const nodesRes = await fetchWithTimeout(fetch, `${base}/api/rigs/${rig.id}/nodes`, {}, {
|
||||
timeoutMs: DOCTOR_HTTP_TIMEOUT_MS,
|
||||
timeoutMessage: "Doctor's live node read timed out.",
|
||||
consumeResponse: async res => {
|
||||
if (res.ok) nodes = (await res.json()) as typeof nodes;
|
||||
else await res.body?.cancel();
|
||||
},
|
||||
});
|
||||
if (!nodesRes.ok) return null;
|
||||
const nodes = (await nodesRes.json()) as Array<{ logicalId?: string | null }>;
|
||||
return nodes.map((n) => n.logicalId ?? "");
|
||||
} catch (error) {
|
||||
if (error instanceof ForeignLocalDaemonError || error instanceof AmbiguousLiveRigError) throw error;
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
import { describe, it, expect, vi } from "vitest";
|
||||
import { createProgram } from "../src/index.js";
|
||||
import { Command } from "commander";
|
||||
import { doctorCommand, runDoctorChecks, type DoctorDeps } from "../src/commands/doctor.js";
|
||||
import { doctorCommand, runDoctorChecks, readLiveLogicalIds, type DoctorDeps } from "../src/commands/doctor.js";
|
||||
import { DaemonClient } from "../src/client.js";
|
||||
import { resolveCmuxSettingsPath } from "../src/cmux-config.js";
|
||||
|
||||
const CMUX_SETTINGS_PATH = resolveCmuxSettingsPath();
|
||||
@@ -567,8 +568,9 @@ describe("runDoctorChecks", () => {
|
||||
it("settles a /healthz body that never finishes as owner unknown within its bound, and releases the request", async () => {
|
||||
const signals: AbortSignal[] = [];
|
||||
const deps = makeDeps({ checkPort: async () => false, callerUid: 1002, ownerReportTimeoutMs: 50,
|
||||
fetch: async (_url: string, init?: { signal?: AbortSignal }) => {
|
||||
if (init?.signal) signals.push(init.signal);
|
||||
fetch: async (url: string, init?: { signal?: AbortSignal }) => {
|
||||
// This owner-body cell measures the two health probes; status has its own deadline cells below.
|
||||
if (init?.signal && url.endsWith("/healthz")) signals.push(init.signal);
|
||||
return { ok: true, json: () => new Promise<unknown>(() => {}) };
|
||||
} });
|
||||
const started = Date.now();
|
||||
@@ -796,3 +798,117 @@ describe("rig doctor", () => {
|
||||
expect(portArgs).toContain(7433);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
describe("doctor JSON HTTP deadlines", () => {
|
||||
it.each(["cmux headers", "cmux body", "healthy"] as const)("returns an honest diagnostic when a real listener holds %s", async (held) => {
|
||||
const { createServer } = await import("node:http");
|
||||
const requests: string[] = [];
|
||||
const server = createServer((req, res) => {
|
||||
requests.push(req.url!);
|
||||
if (req.url === "/healthz") {
|
||||
res.writeHead(200, { "content-type": "application/json" });
|
||||
res.end(JSON.stringify({ ok: true, pid: 4242, euid: 1002 }));
|
||||
} else if (held === "healthy") {
|
||||
res.writeHead(200, { "content-type": "application/json" });
|
||||
res.end(JSON.stringify({ available: true }));
|
||||
} else if (held === "cmux body") {
|
||||
res.writeHead(200, { "content-type": "application/json" });
|
||||
res.write('{"available":');
|
||||
}
|
||||
});
|
||||
await new Promise<void>(resolve => server.listen(0, "127.0.0.1", resolve));
|
||||
const port = (server.address() as { port: number }).port;
|
||||
const log = vi.spyOn(console, "log").mockImplementation(() => {});
|
||||
const priorExitCode = process.exitCode;
|
||||
const program = new Command();
|
||||
program.addCommand(doctorCommand(makeDeps({ checkPort: async () => false, callerUid: 1002,
|
||||
ownerReportTimeoutMs: 100, fetch: (url, init) => globalThis.fetch(url, init),
|
||||
configStore: { resolve: () => ({ ...defaultConfig, daemon: { host: "127.0.0.1", port } }) },
|
||||
})));
|
||||
const task = program.parseAsync(["node", "rig", "doctor", "--json"]);
|
||||
let observationTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
const outcome = await Promise.race([task.then(() => "settled"), new Promise<string>(resolve => {
|
||||
observationTimer = setTimeout(() => resolve("still-waiting"), 1500);
|
||||
})]);
|
||||
expect(["darwin", "linux"]).toContain(process.platform);
|
||||
expect(requests).toContain("/healthz");
|
||||
expect(requests).toContain("/api/adapters/cmux/status");
|
||||
console.info("[doctor HTTP deadline]", JSON.stringify({ platform: process.platform, held, requests, outcome }));
|
||||
expect(outcome).toBe("settled");
|
||||
const result = JSON.parse(String(log.mock.calls.at(-1)![0]));
|
||||
const portCheck = result.checks.find((c: { name: string }) => c.name === "port");
|
||||
const cmux = result.checks.find((c: { name: string }) => c.name === "cmux_daemon");
|
||||
expect(portCheck.status).toBe("pass");
|
||||
expect(cmux.status).toBe(held === "healthy" ? "pass" : "warn");
|
||||
} finally {
|
||||
clearTimeout(observationTimer);
|
||||
server.closeAllConnections();
|
||||
await new Promise<void>(resolve => server.close(() => resolve()));
|
||||
await Promise.allSettled([task]);
|
||||
log.mockRestore(); process.exitCode = priorExitCode;
|
||||
expect(server.listening).toBe(false);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
describe("doctor --spec topology HTTP deadlines", () => {
|
||||
it.each(["rig headers", "rig body", "nodes headers", "nodes body", "healthy"] as const)(
|
||||
"returns unknown or measured conformance when the topology listener is %s", async held => {
|
||||
const { createServer } = await import("node:http");
|
||||
const requests: string[] = [];
|
||||
const heldPath = held.startsWith("rig") ? "/api/rigs" : "/api/rigs/rig-id/nodes";
|
||||
const server = createServer((req, res) => {
|
||||
requests.push(req.url!);
|
||||
if (held !== "healthy" && req.url === heldPath) {
|
||||
if (held.endsWith("body")) { res.writeHead(200, { "content-type": "application/json" }); res.write("["); }
|
||||
return;
|
||||
}
|
||||
res.writeHead(200, { "content-type": "application/json" });
|
||||
res.end(JSON.stringify(req.url === "/healthz" ? { ok: true, pid: 4242, euid: 1002 }
|
||||
: req.url === "/api/rigs" ? [{ id: "rig-id", name: "fixture" }]
|
||||
: req.url === "/api/rigs/rig-id/nodes" ? [{ logicalId: "dev.worker" }] : { available: true }));
|
||||
});
|
||||
await new Promise<void>(resolve => server.listen(0, "127.0.0.1", resolve));
|
||||
const port = (server.address() as { port: number }).port;
|
||||
const base = `http://127.0.0.1:${port}`;
|
||||
vi.stubEnv("OPENRIG_TRANSPORT_LOCAL_USER_GUARD", "true");
|
||||
vi.stubEnv("OPENRIG_URL", ""); vi.stubEnv("RIGGED_URL", "");
|
||||
const ownerProbe = vi.fn(async () => ({ state: "unknown" as const, reason: "fixture-unconfirmed" }));
|
||||
const ownerClient = new DaemonClient(base, { ownerProbe });
|
||||
const specPath = "/fixture/rig.yaml";
|
||||
const spec = 'version: "0.2"\nname: fixture\npods:\n - id: dev\n label: Development\n members:\n - id: worker\n';
|
||||
const log = vi.spyOn(console, "log").mockImplementation(() => {});
|
||||
const priorExitCode = process.exitCode;
|
||||
const program = new Command().addCommand(doctorCommand(makeDeps({ callerUid: 1002,
|
||||
checkPort: async () => false, specPath, readFile: path => path === specPath ? spec : null,
|
||||
configStore: { resolve: () => ({ ...defaultConfig, daemon: { host: "127.0.0.1", port } }) },
|
||||
fetch: (url, init) => globalThis.fetch(url, init),
|
||||
fetchLiveLogicalIds: name => readLiveLogicalIds(base, name, ownerClient),
|
||||
})));
|
||||
const task = program.parseAsync(["node", "rig", "doctor", "--json", "--spec", specPath]);
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
const outcome = await Promise.race([task.then(() => "settled"), new Promise<string>(resolve => {
|
||||
timer = setTimeout(() => resolve("still-waiting"), 4500);
|
||||
})]);
|
||||
expect(["darwin", "linux"]).toContain(process.platform);
|
||||
expect(requests).toContain(held === "healthy" ? "/api/rigs/rig-id/nodes" : heldPath);
|
||||
expect(ownerProbe).toHaveBeenCalled();
|
||||
console.info("[doctor spec HTTP deadline]", JSON.stringify({ platform: process.platform, held, requests, outcome }));
|
||||
expect(outcome).toBe("settled");
|
||||
const result = JSON.parse(String(log.mock.calls.at(-1)![0]));
|
||||
const check = result.checks.find((c: { name: string }) => c.name === "spec_live_conformance");
|
||||
expect(check.status).toBe(held === "healthy" ? "pass" : "skipped");
|
||||
if (held === "healthy") expect(ownerProbe).toHaveBeenCalledTimes(2);
|
||||
} finally {
|
||||
clearTimeout(timer); server.closeAllConnections();
|
||||
await new Promise<void>(resolve => server.close(() => resolve()));
|
||||
await Promise.allSettled([task]);
|
||||
log.mockRestore(); process.exitCode = priorExitCode; vi.unstubAllEnvs();
|
||||
expect(server.listening).toBe(false);
|
||||
}
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user