mosaicstack/stack
25
0
Fork 1
Code Issues 367 Pull Requests 13 Actions Packages Projects Releases 15 Wiki Activity

cli: bus host start releases host.lock and the broker on every failure after the reply, and handles the child error event, row 111 (#1605)

Browse Source 
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]>
This commit is contained in:
jason.woltje
2026-10-13 18:59:06 -05:00
co-authored by Claude Opus 5.5
parent e60410566c
commit 9774d70db5
2 changed files with 183 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/cli/src/host.mjs
+51 -25
View File
@@ -124,20 +124,24 @@ function writeHostState(dataRoot, state) {
renameSync(tmp, file);
}
// The first IPC message from child, or a refusal when it exits or the wait runs out.
// 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);
});
}
@@ -183,6 +187,25 @@ export async function startHost({ boot, business, notifier = null, bootTimeoutMs
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.
@@ -190,6 +213,7 @@ export async function startHost({ boot, business, notifier = null, bootTimeoutMs
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) {
@@ -205,36 +229,36 @@ export async function startHost({ boot, business, notifier = null, bootTimeoutMs
release();
throw new CliError(`broker refused to start: ${typeof ready?.error === "string" ? ready.error : "startup-refused"}`, 3);
}
const reader = ready.readers.find((r) => r.business === business);
// 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;
if (notifier) {
notify = fork(NOTIFIER, [], { stdio: ["ignore", "inherit", "inherit", "ipc"] });
const started = firstReply(notify, START_TIMEOUT_MS, "notifier");
notify.send({ op: "start", path: ready.path, cap: reader.cap, business, dataRoot, binding: notifier.binding, base: notifier.base, pollMs: notifier.pollMs });
let ok;
try {
ok = await started;
} catch (e) {
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;
}
if (ok?.ok !== true) {
await ended(notify, 5000);
if (broker.connected) broker.send({ op: "close" }, () => {});
await ended(broker, CLOSE_TIMEOUT_MS);
release();
throw new CliError(`notifier refused to start: ${typeof ok?.error === "string" ? ok.error : "notifier-refused"}`, 3);
}
if (broker.connected) broker.send({ op: "close" }, () => {});
await ended(broker, CLOSE_TIMEOUT_MS);
release();
throw e;
}
rmSync(hostFile(dataRoot), { force: true });
writeHostState(dataRoot, state);
// 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;
@@ -330,7 +354,9 @@ export async function startHost({ boot, business, notifier = null, bootTimeoutMs
// 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.
// 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`);