jason.woltje
8fec494694
T1 follow-up. A definition's text becomes an agent's contract, but a load ran the held-secret check with an empty store. Bus start now checks every definitions entry (the boot copy's org: entries and the file's own) with the live store, and the launch check asks the bus the same question for the org: definition it just read, before a contract exists. A hit refuses as definition-holds-credential, a new code in the refusal table; the refusal names the definition and key, never the text or the credential. bus start records the refusal in the boot status.
413 lines
18 KiB
JavaScript
413 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", 3);
|
|
}
|
|
if (!state || !Number.isSafeInteger(state.pid) || typeof state.startTime !== "string" || typeof state.business !== "string") {
|
|
throw new CliError("bus host state is malformed", 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 = () => {}, refuseStale = false } = {}) {
|
|
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);
|
|
if (refuseStale) throw new CliError("stale bus host lock", 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("bus host lock 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, options = null, refuseStale = false, 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, ...(options ? { options } : {}) };
|
|
claimHostLock(dataRoot, state, { refuseStale });
|
|
// 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();
|
|
const code = typeof ready?.error === "string" ? ready.error : "startup-refused";
|
|
// org-defscan: the broker's reason names the definition and key that hold a credential, never the text.
|
|
const why = code === "definition-holds-credential" && typeof ready?.reason === "string" ? `: ${ready.reason.slice(0, 300)}` : "";
|
|
throw Object.assign(new CliError(`broker refused to start: ${code}${why}`, 3), code === "definition-holds-credential" ? { code } : {});
|
|
}
|
|
|
|
// 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, instanceRoot: boot.instanceRoot ?? 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, expectedBusiness = null } = {}) {
|
|
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" || (expectedBusiness !== null && (state.business !== expectedBusiness || argv[i + 3] !== expectedBusiness))) {
|
|
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,
|
|
};
|
|
}
|