jason.woltjeandClaude Opus 5.5
9774d70db5
Dewey approve, comment 28331, manifest 5e888f93. Sage gate on be6f2ed8c
plus the patch: every package suite green (cli 178); test-queue failed
review.test.mjs:123 once under load (16s vs 3.7s), passed 28/28 rerun alone;
release 14, task 98. Dewey note a (ended() rejects on a child error and skips
release()) is recorded for the next row that touches host.mjs.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
409 lines
18 KiB
JavaScript
409 lines
18 KiB
JavaScript
// The trusted bus host (lead decision 70). It forks
|
|
// packages/bus/src/process.mjs, boots it over IPC, keeps the reader
|
|
// capability it gets back, and hands it to the notifier child over IPC.
|
|
// Capabilities never appear in argv, stdout or the environment. While the
|
|
// host runs, `<dataRoot>/bus-host/host.json` (0600) names it for `mosaic bus
|
|
// stop|status` and for the human commands' default business. From the start
|
|
// of its boot until it stops, a host holds `bus-host/host.lock`
|
|
// (claimHostLock()), so one host runs per dataRoot.
|
|
//
|
|
// startHost() is also the in-process API the S6 launcher (src/launcher.mjs)
|
|
// uses: bindLaunch(record) binds a launched run's process identity and
|
|
// returns its capability; op(message) runs one of the broker's launch ops
|
|
// (identity, authorizeLaunch, refuse, endLaunch, credentialStatus); and
|
|
// beforeClose(fn) runs fn before the host stops its children, while those
|
|
// ops still work.
|
|
|
|
import { fork } from "node:child_process";
|
|
import { randomBytes } from "node:crypto";
|
|
import { once } from "node:events";
|
|
import { existsSync, linkSync, mkdirSync, readFileSync, renameSync, rmSync, writeFileSync } from "node:fs";
|
|
import { join } from "node:path";
|
|
import { fileURLToPath } from "node:url";
|
|
import { CliError } from "./errors.mjs";
|
|
import { cmdlineOf, startTimeOf } from "../../seat/src/proc.mjs";
|
|
|
|
const BROKER = fileURLToPath(new URL("../../bus/src/process.mjs", import.meta.url));
|
|
const NOTIFIER = fileURLToPath(new URL("./notifier-process.mjs", import.meta.url));
|
|
export const BOOT_TIMEOUT_MS = 30000;
|
|
const START_TIMEOUT_MS = 15000;
|
|
const CLOSE_TIMEOUT_MS = 20000;
|
|
const REQUEST_TIMEOUT_MS = 10000;
|
|
// How many times startHost tries to take host.lock while it keeps changing.
|
|
const CLAIM_TRIES = 3;
|
|
|
|
export const hostDir = (dataRoot) => join(dataRoot, "bus-host");
|
|
export const hostFile = (dataRoot) => join(hostDir(dataRoot), "host.json");
|
|
export const hostLock = (dataRoot) => join(hostDir(dataRoot), "host.lock");
|
|
|
|
export { cmdlineOf, startTimeOf };
|
|
|
|
// The state file, or null. `live` is true only when the pid still runs with
|
|
// the recorded start time, so a recycled pid never counts as the host.
|
|
export function readHostState(dataRoot) {
|
|
return readState(hostFile(dataRoot));
|
|
}
|
|
|
|
// A file that goes between the check and the read counts as no file.
|
|
// `found` runs between the two; it is for the tests.
|
|
function readState(file, found = () => {}) {
|
|
if (!existsSync(file)) return null;
|
|
found();
|
|
let state;
|
|
try {
|
|
state = JSON.parse(readFileSync(file, "utf8"));
|
|
} catch (e) {
|
|
if (e.code === "ENOENT") return null;
|
|
throw new CliError(`bus host state is unreadable: ${file}`, 3);
|
|
}
|
|
if (!state || !Number.isSafeInteger(state.pid) || typeof state.startTime !== "string" || typeof state.business !== "string") {
|
|
throw new CliError(`bus host state is malformed: ${file}`, 3);
|
|
}
|
|
return { ...state, live: startTimeOf(state.pid) === state.startTime };
|
|
}
|
|
|
|
// One host's state, the same one: pid, start time and start stamp.
|
|
const same = (a, b) => a?.pid === b?.pid && a?.startTime === b?.startTime && a?.startedAt === b?.startedAt;
|
|
|
|
// Takes host.lock for `state`, or refuses while a live host holds it. The
|
|
// state goes to a file of its own and is then linked in. A link fails if
|
|
// host.lock exists, so two hosts can't both take it, and the lock is never
|
|
// half written. A lock whose holder no longer runs, whatever its age, is
|
|
// moved aside, checked to be the file that was read, and removed; another
|
|
// host's claim moved aside by that race goes back. A third host racing the
|
|
// same stale lock can still lose a claim there, and then the broker's writer
|
|
// lock refuses the second broker. A lock that goes before it is read is
|
|
// tried again, like one that goes before the check. `found` runs once the
|
|
// lock is found and before it is read, `between` after the stale lock is
|
|
// read; both are for the tests. Exported for the tests.
|
|
export function claimHostLock(dataRoot, state, { found = () => {}, between = () => {} } = {}) {
|
|
mkdirSync(hostDir(dataRoot), { recursive: true, mode: 0o700 });
|
|
const file = hostLock(dataRoot);
|
|
const tag = `${process.pid}.${randomBytes(4).toString("hex")}`;
|
|
const tmp = `${file}.${tag}.tmp`;
|
|
const aside = `${file}.${tag}.stale`;
|
|
writeFileSync(tmp, `${JSON.stringify(state, null, 2)}\n`, { mode: 0o600, flag: "wx" });
|
|
try {
|
|
for (let i = 0; i < CLAIM_TRIES; i++) {
|
|
try {
|
|
linkSync(tmp, file);
|
|
return;
|
|
} catch (e) {
|
|
if (e.code !== "EEXIST") throw e;
|
|
}
|
|
const prior = readState(file, found);
|
|
if (!prior) continue;
|
|
if (prior.live) throw new CliError(`a bus host already runs for ${prior.business} (pid ${prior.pid})`, 3);
|
|
between();
|
|
try {
|
|
renameSync(file, aside);
|
|
} catch (e) {
|
|
if (e.code === "ENOENT") continue;
|
|
throw e;
|
|
}
|
|
if (!same(readState(aside), prior)) {
|
|
try {
|
|
linkSync(aside, file);
|
|
} catch (e) {
|
|
if (e.code !== "EEXIST") throw e;
|
|
}
|
|
}
|
|
rmSync(aside, { force: true });
|
|
}
|
|
throw new CliError(`${file} kept changing; no bus host started`, 3);
|
|
} finally {
|
|
rmSync(tmp, { force: true });
|
|
}
|
|
}
|
|
|
|
function writeHostState(dataRoot, state) {
|
|
mkdirSync(hostDir(dataRoot), { recursive: true, mode: 0o700 });
|
|
const file = hostFile(dataRoot);
|
|
const tmp = `${file}.${process.pid}.tmp`;
|
|
writeFileSync(tmp, `${JSON.stringify(state, null, 2)}\n`, { mode: 0o600, flag: "wx" });
|
|
renameSync(tmp, file);
|
|
}
|
|
|
|
// The first IPC message from child, or a refusal when it exits, emits
|
|
// `error` or the wait runs out.
|
|
function firstReply(child, timeoutMs, what) {
|
|
return new Promise((resolve, reject) => {
|
|
const done = (fn, v) => {
|
|
clearTimeout(timer);
|
|
child.off("message", onMessage);
|
|
child.off("exit", onExit);
|
|
child.off("error", onError);
|
|
fn(v);
|
|
};
|
|
const onMessage = (m) => done(resolve, m);
|
|
const onExit = (code) => done(reject, new CliError(`${what} exited (${code}) before it replied`, 1));
|
|
const onError = (e) => done(reject, new CliError(`${what} failed before it replied: ${e.message}`, 1));
|
|
const timer = setTimeout(() => done(reject, new CliError(`${what} did not reply within ${Math.round(timeoutMs / 1000)} s`, 1)), timeoutMs);
|
|
child.on("message", onMessage);
|
|
child.on("exit", onExit);
|
|
child.on("error", onError);
|
|
});
|
|
}
|
|
|
|
async function ended(child, timeoutMs) {
|
|
if (child.exitCode !== null || child.signalCode !== null) return child.exitCode;
|
|
const timer = setTimeout(() => child.kill("SIGKILL"), timeoutMs);
|
|
const [code] = await once(child, "exit");
|
|
clearTimeout(timer);
|
|
return code;
|
|
}
|
|
|
|
// Calls onDeath(name, code, signal) once for each child that exits. A child
|
|
// that exited before this runs (the broker while the notifier starts, say)
|
|
// has already emitted its exit event, so its exit state is checked here.
|
|
export function watchChildren(children, onDeath) {
|
|
for (const [name, child] of Object.entries(children)) {
|
|
if (!child) continue;
|
|
if (child.exitCode !== null || child.signalCode !== null) onDeath(name, child.exitCode, child.signalCode);
|
|
else child.once("exit", (code, signal) => onDeath(name, code, signal));
|
|
}
|
|
}
|
|
|
|
// boot: the {op:'boot'} config from bootConfig(). notifier: null, or
|
|
// {binding, base?, pollMs?}; base and pollMs exist for the tests, and the
|
|
// command line never sets them, nor requestTimeoutMs. Resolves once both
|
|
// children are up.
|
|
export async function startHost({ boot, business, notifier = null, bootTimeoutMs = BOOT_TIMEOUT_MS, requestTimeoutMs = REQUEST_TIMEOUT_MS, log = (l) => process.stderr.write(`mosaic-bus: ${l}\n`) }) {
|
|
const dataRoot = boot.dataRoot;
|
|
// A host started before host.lock existed holds only host.json.
|
|
const prior = readHostState(dataRoot);
|
|
if (prior?.live) throw new CliError(`a bus host already runs for ${prior.business} (pid ${prior.pid})`, 3);
|
|
// host.lock is this host's from here until it stops, and a start that
|
|
// fails gives it back. host.json comes once the children are up, so a
|
|
// reader never finds a booting host with no broker socket.
|
|
const state = { hostVersion: 1, pid: process.pid, startTime: startTimeOf(process.pid), business, startedAt: new Date().toISOString(), notifier: notifier ? notifier.binding : null };
|
|
claimHostLock(dataRoot, state);
|
|
// A lock this host can't read isn't its own, so it stays.
|
|
const release = () => {
|
|
let held = null;
|
|
try {
|
|
held = readState(hostLock(dataRoot));
|
|
} catch {}
|
|
if (same(held, state)) rmSync(hostLock(dataRoot), { force: true });
|
|
};
|
|
|
|
// Node emits `error` on a child for a send with no callback that fails, or
|
|
// a kill that fails, and with no listener that ends this process with
|
|
// host.lock held. Until the host is up, an `error` fails the start: the
|
|
// wait for a reply rejects on it, and one that comes at any other time is
|
|
// kept and thrown before host.json is written. Once the host is up, an
|
|
// `error` stops the host with exit 1, as a child's death does.
|
|
let up = false;
|
|
let childError = null;
|
|
const onChildError = (name) => (e) => {
|
|
if (!up) {
|
|
childError ??= new CliError(`${name} failed: ${e.message}`, 1);
|
|
return;
|
|
}
|
|
if (name === "broker") return breakChannel(`broker error: ${e.message}`);
|
|
if (stopping) return;
|
|
log(`notifier error (${e.message}); stopping the host`);
|
|
close(1);
|
|
};
|
|
|
|
// A fork or a send that throws gives the lock back too. The wait starts
|
|
// after the send, so a send that throws leaves no wait behind. The reply
|
|
// and the exit event come on a later tick, so the wait misses neither.
|
|
let broker = null;
|
|
let ready;
|
|
try {
|
|
broker = fork(BROKER, [], { stdio: ["ignore", "inherit", "inherit", "ipc"] });
|
|
broker.on("error", onChildError("broker"));
|
|
broker.send({ op: "boot", config: boot });
|
|
ready = await firstReply(broker, bootTimeoutMs, "broker");
|
|
} catch (e) {
|
|
if (broker) {
|
|
broker.kill("SIGTERM");
|
|
await ended(broker, 5000);
|
|
}
|
|
release();
|
|
throw e;
|
|
}
|
|
if (ready?.ok !== true) {
|
|
await ended(broker, 5000);
|
|
release();
|
|
throw new CliError(`broker refused to start: ${typeof ready?.error === "string" ? ready.error : "startup-refused"}`, 3);
|
|
}
|
|
|
|
// Every step from here to host.json shares one failure path: it stops the
|
|
// notifier if one started, closes the broker, waits for both to end and
|
|
// gives the lock back. The notifier's wait starts after its send, as the
|
|
// broker's does.
|
|
let notify = null;
|
|
try {
|
|
const reader = ready.readers?.find((r) => r.business === business);
|
|
if (!reader) throw new CliError(`broker has no reader for business ${business}`, 3);
|
|
if (notifier) {
|
|
notify = fork(NOTIFIER, [], { stdio: ["ignore", "inherit", "inherit", "ipc"] });
|
|
notify.on("error", onChildError("notifier"));
|
|
notify.send({ op: "start", path: ready.path, cap: reader.cap, business, dataRoot, binding: notifier.binding, base: notifier.base, pollMs: notifier.pollMs });
|
|
const ok = await firstReply(notify, START_TIMEOUT_MS, "notifier");
|
|
if (ok?.ok !== true) throw new CliError(`notifier refused to start: ${typeof ok?.error === "string" ? ok.error : "notifier-refused"}`, 3);
|
|
}
|
|
if (childError) throw childError;
|
|
rmSync(hostFile(dataRoot), { force: true });
|
|
writeHostState(dataRoot, state);
|
|
} catch (e) {
|
|
if (notify) {
|
|
notify.kill("SIGTERM");
|
|
await ended(notify, 5000);
|
|
}
|
|
if (broker.connected) broker.send({ op: "close" }, () => {});
|
|
await ended(broker, CLOSE_TIMEOUT_MS);
|
|
release();
|
|
throw e;
|
|
}
|
|
|
|
// stopping: close() has started (no new close, child deaths are expected).
|
|
// closing: the before-close hooks have run, so no more broker requests.
|
|
let stopping = false;
|
|
let closing = false;
|
|
let finish;
|
|
const done = new Promise((r) => (finish = r));
|
|
const hooks = [];
|
|
|
|
// Requests go one at a time, each with an id that process.mjs echoes. A
|
|
// reply that misses the wait, or carries an id no request is waiting for,
|
|
// breaks the channel: the waiting request and every later one refuse, and
|
|
// the host stops with exit 1 so the unit restarts it. A late reply can
|
|
// never answer a later request.
|
|
let queue = Promise.resolve();
|
|
let seq = 0;
|
|
let pending = null;
|
|
let broken = null;
|
|
const fail = (code, text) => Object.assign(new CliError(text, 1), { code });
|
|
function breakChannel(why) {
|
|
if (broken) return;
|
|
broken = why;
|
|
if (pending) {
|
|
clearTimeout(pending.timer);
|
|
pending.reject(fail("broker-channel-broken", `broker channel broken: ${why}`));
|
|
pending = null;
|
|
}
|
|
if (stopping) return;
|
|
log(`broker channel broken (${why}); stopping the host`);
|
|
close(1);
|
|
}
|
|
broker.on("message", (m) => {
|
|
if (pending && m?.id === pending.id) {
|
|
clearTimeout(pending.timer);
|
|
const { resolve } = pending;
|
|
pending = null;
|
|
resolve(m);
|
|
} else breakChannel(pending ? "reply for another request" : "reply with no request waiting");
|
|
});
|
|
broker.once("exit", () => {
|
|
if (!pending) return;
|
|
clearTimeout(pending.timer);
|
|
pending.reject(fail("broker-exited", "broker exited before it replied"));
|
|
pending = null;
|
|
});
|
|
function request(message, { what, pick }) {
|
|
const run = queue.then(async () => {
|
|
if (broken) throw fail("broker-channel-broken", `broker channel broken: ${broken}`);
|
|
if (closing || !broker.connected) throw fail("host-closing", "bus host is closing");
|
|
const id = ++seq;
|
|
const m = await new Promise((resolve, reject) => {
|
|
const timer = setTimeout(() => breakChannel(`no reply within ${Math.round(requestTimeoutMs / 1000)} s`), requestTimeoutMs);
|
|
pending = { id, resolve, reject, timer };
|
|
broker.send({ ...message, id });
|
|
});
|
|
if (m?.ok !== true) {
|
|
const code = typeof m?.error === "string" ? m.error : `${what}-refused`;
|
|
throw Object.assign(new CliError(`${what} refused: ${code}`, 3), { code });
|
|
}
|
|
return pick(m);
|
|
});
|
|
queue = run.catch(() => {});
|
|
return run;
|
|
}
|
|
const bindLaunch = (record) => request({ op: "bindLaunch", record }, { what: "launch bind", pick: (m) => m.launch });
|
|
const op = (message) => request(message, { what: message.op, pick: (m) => m.result });
|
|
const beforeClose = (fn) => void hooks.push(fn);
|
|
|
|
async function close(code = 0) {
|
|
if (stopping) return done;
|
|
stopping = true;
|
|
for (const fn of hooks) {
|
|
try {
|
|
await fn();
|
|
} catch (e) {
|
|
log(`before close: ${e.message}`);
|
|
}
|
|
}
|
|
closing = true;
|
|
let result = code;
|
|
if (notify) {
|
|
if (notify.connected) notify.send({ op: "stop" }, () => {});
|
|
if ((await ended(notify, CLOSE_TIMEOUT_MS)) !== 0 && result === 0) result = 1;
|
|
}
|
|
await queue;
|
|
if (broker.connected) broker.send({ op: "close" }, () => {});
|
|
if ((await ended(broker, CLOSE_TIMEOUT_MS)) !== 0 && result === 0) result = 1;
|
|
const now = readHostState(dataRoot);
|
|
if (now && now.pid === state.pid && now.startTime === state.startTime) rmSync(hostFile(dataRoot), { force: true });
|
|
release();
|
|
finish(result);
|
|
return done;
|
|
}
|
|
|
|
// A child that dies on its own takes the host down with exit 1: the unit
|
|
// restarts the pair rather than run a broker without its notifier.
|
|
// Runs after `queue` exists, since close() awaits it. No event comes
|
|
// between host.json and here, so an `error` meets breakChannel and close.
|
|
up = true;
|
|
watchChildren({ broker, notifier: notify }, (name, code, signal) => {
|
|
if (stopping) return;
|
|
log(`${name} exited (${signal ?? code}); stopping the host`);
|
|
close(1);
|
|
});
|
|
|
|
return Object.freeze({ path: ready.path, dataRoot, business, bindLaunch, op, beforeClose, close, done, pids: Object.freeze({ broker: broker.pid, notifier: notify?.pid ?? null }) });
|
|
}
|
|
|
|
// `mosaic bus stop`: SIGTERM to the recorded host, after checking that the
|
|
// pid still runs with the recorded start time and is a `bus start` process.
|
|
export async function stopHost(dataRoot, { timeoutMs = 60000 } = {}) {
|
|
const state = readHostState(dataRoot);
|
|
if (!state || !state.live) return { stopped: false, reason: state ? "stale state file; no host runs" : "no host runs" };
|
|
const argv = cmdlineOf(state.pid) ?? [];
|
|
const i = argv.findIndex((a) => a.endsWith("/packages/cli/src/cli.mjs"));
|
|
if (i < 0 || argv[i + 1] !== "bus" || argv[i + 2] !== "start") {
|
|
throw new CliError(`pid ${state.pid} is not a bus host; refusing to signal it`, 3);
|
|
}
|
|
process.kill(state.pid, "SIGTERM");
|
|
const deadline = Date.now() + timeoutMs;
|
|
while (Date.now() < deadline) {
|
|
if (startTimeOf(state.pid) !== state.startTime) return { stopped: true, business: state.business, pid: state.pid };
|
|
await new Promise((r) => setTimeout(r, 200));
|
|
}
|
|
throw new CliError(`bus host pid ${state.pid} did not stop within ${Math.round(timeoutMs / 1000)} s`, 1);
|
|
}
|
|
|
|
// `mosaic bus status`: what the files and /proc say. Never removes a lock.
|
|
export function hostStatus(dataRoot) {
|
|
const state = readHostState(dataRoot);
|
|
const lock = join(dataRoot, "bus", "writer.lock");
|
|
// The bus store writes {pid, at}. A dead pid means a broker was killed
|
|
// hard; the bus README leaves removing the lock to the operator.
|
|
let owner = null;
|
|
if (existsSync(lock)) {
|
|
try {
|
|
owner = JSON.parse(readFileSync(lock, "utf8"));
|
|
} catch {
|
|
owner = {};
|
|
}
|
|
}
|
|
const lockPid = Number.isSafeInteger(owner?.pid) ? owner.pid : null;
|
|
return {
|
|
host: state ? { business: state.business, pid: state.pid, startedAt: state.startedAt, notifier: state.notifier ?? null, live: state.live } : null,
|
|
socket: existsSync(join(dataRoot, "bus", "broker.sock")),
|
|
writerLock: owner ? { pid: lockPid, at: typeof owner.at === "string" ? owner.at : null, live: lockPid !== null && startTimeOf(lockPid) !== null } : null,
|
|
};
|
|
}
|