diff --git a/packages/cli/src/host.mjs b/packages/cli/src/host.mjs index 02b15477d..69640f23c 100644 --- a/packages/cli/src/host.mjs +++ b/packages/cli/src/host.mjs @@ -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) { - notify.kill("SIGTERM"); - await ended(notify, 5000); - if (broker.connected) broker.send({ op: "close" }, () => {}); - await ended(broker, CLOSE_TIMEOUT_MS); - release(); - throw 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 (ok?.ok !== true) { + 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 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`); diff --git a/packages/cli/tests/host.test.mjs b/packages/cli/tests/host.test.mjs index 1b55505c6..6a1170357 100644 --- a/packages/cli/tests/host.test.mjs +++ b/packages/cli/tests/host.test.mjs @@ -108,14 +108,14 @@ function failSends(t, fake) { }); } -// Makes the next fork throw (`where` "fork") or the new child's sends throw +// Makes the nth new child's fork throw (`where` "fork") or its sends throw // ("send"), through the `send` that node installs on a new child: a fork -// throws when the install does. Every child it saw is killed after the test, -// the one a throwing fork started included. -function throwOn(t, where) { +// throws when the install does. Other children are left alone. Every child +// it saw is killed after the test, the one a throwing fork started included. +function throwOn(t, where, nth = 0) { const children = []; const onChild = ({ process: child }) => { - children.push(child); + if (children.push(child) - 1 !== nth) return; let send; Object.defineProperty(child, "send", { configurable: true, @@ -135,6 +135,22 @@ function throwOn(t, where) { }); } +// Every child this process creates while the test runs, in fork order. Each +// is killed after the test, so a start that leaves a child running fails the +// test rather than hang the file. +function forksSeen(t) { + const children = []; + const onChild = ({ process: child }) => children.push(child); + subscribe("child_process", onChild); + t.after(() => { + unsubscribe("child_process", onChild); + for (const child of children) child.kill("SIGKILL"); + }); + return children; +} + +const gone = (child) => child.exitCode !== null || child.signalCode !== null; + const epipe = (child) => { child.kill("SIGKILL"); return Object.assign(new Error("write EPIPE"), { code: "EPIPE", errno: -32, syscall: "write" }); @@ -407,6 +423,117 @@ test("a broker that misses its boot wait leaves no host.lock and no host.json", assert.equal(existsSync(join(f.dataRoot, "bus", "writer.lock")), false); }); +for (const where of ["fork", "send"]) + test(`a notifier ${where} that throws stops the broker and leaves no host.lock and no host.json`, async (t) => { + const root = tmp(t); + const f = fixture(root); + makeDeployment(root, { dmRecipient: IDS.owner }); + const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env }); + const children = forksSeen(t); + throwOn(t, where, 1); + const started = startHost({ boot, business: "acme", notifier: { binding: "test-seat" }, log: () => {} }); + t.after(async () => (await started.catch(() => null))?.close(0)); + await assert.rejects(started, (e) => e.message === `${where} failed`); + assert.equal(children.length, 2); + assert.ok(gone(children[0]), "the broker has ended"); + // A fork that throws returns no child to stop; throwOn kills it. + if (where === "send") assert.ok(gone(children[1]), "the notifier has ended"); + assert.equal(existsSync(hostLock(f.dataRoot)), false); + assert.equal(existsSync(hostFile(f.dataRoot)), false); + assert.equal(existsSync(join(f.dataRoot, "bus", "writer.lock")), false); + }); + +for (const withNotifier of [true, false]) + test(`a boot reply with no reader for the business refuses with exit 3 naming it, ${withNotifier ? "before the notifier starts" : "with no notifier"}, and leaves no host.lock and no live broker`, async (t) => { + const root = tmp(t); + const f = fixture(root); + makeDeployment(root, { dmRecipient: IDS.owner }); + const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env }); + const children = forksSeen(t); + const started = startHost({ boot, business: "other", notifier: withNotifier ? { binding: "test-seat" } : null, log: () => {} }); + t.after(async () => (await started.catch(() => null))?.close(0)); + await assert.rejects(started, (e) => e.exitCode === 3 && e.message === "broker has no reader for business other"); + assert.equal(children.length, 1, "no notifier was started"); + assert.ok(gone(children[0]), "the broker has ended"); + assert.equal(existsSync(hostLock(f.dataRoot)), false); + assert.equal(existsSync(hostFile(f.dataRoot)), false); + assert.equal(existsSync(join(f.dataRoot, "bus", "writer.lock")), false); + }); + +test("a host.json that can't be written stops both children and leaves no host.lock", async (t) => { + const root = tmp(t); + const f = fixture(root); + makeDeployment(root, { dmRecipient: IDS.owner }); + const discord = await fakeDiscord(t); + const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env }); + const children = forksSeen(t); + // The host writes host.json through this name with the `wx` flag. + mkdirSync(hostDir(f.dataRoot), { recursive: true, mode: 0o700 }); + writeFileSync(`${hostFile(f.dataRoot)}.${process.pid}.tmp`, ""); + const started = startHost({ boot, business: "acme", notifier: { binding: "test-seat", base: discord.base, pollMs: 100 }, log: () => {} }); + t.after(async () => (await started.catch(() => null))?.close(0)); + await assert.rejects(started, (e) => e.code === "EEXIST"); + assert.equal(children.length, 2); + assert.ok(children.every(gone), "the broker and the notifier have ended"); + assert.equal(existsSync(hostLock(f.dataRoot)), false); + assert.equal(existsSync(hostFile(f.dataRoot)), false); + assert.equal(existsSync(join(f.dataRoot, "bus", "writer.lock")), false); +}); + +for (const [name, op] of [["broker", "boot"], ["notifier", "start"]]) + test(`an error event on the ${name} before it replies fails the start at once, and leaves no host.lock and no live child`, { timeout: 8000 }, async (t) => { + const root = tmp(t); + const f = fixture(root); + makeDeployment(root, { dmRecipient: IDS.owner }); + const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env }); + const children = forksSeen(t); + // The send fails, and the child, still running, never gets the message. + failSends(t, (child, m) => (m?.op === op ? Object.assign(new Error("write EPIPE"), { code: "EPIPE" }) : null)); + const started = startHost({ boot, business: "acme", notifier: { binding: "test-seat" }, log: () => {} }); + t.after(async () => (await started.catch(() => null))?.close(0)); + await assert.rejects(started, (e) => e.exitCode === 1 && e.message === `${name} failed before it replied: write EPIPE`); + assert.ok(children.every(gone), "every child has ended"); + assert.equal(existsSync(hostLock(f.dataRoot)), false); + assert.equal(existsSync(hostFile(f.dataRoot)), false); + }); + +test("an error event on the broker while the notifier starts fails the start once the notifier replies", async (t) => { + const root = tmp(t); + const f = fixture(root); + makeDeployment(root, { dmRecipient: IDS.owner }); + const discord = await fakeDiscord(t); + const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env }); + const children = forksSeen(t); + spySends(t, (m) => { + if (m?.op === "start") process.nextTick(() => children[0].emit("error", new Error("write EPIPE"))); + }); + const started = startHost({ boot, business: "acme", notifier: { binding: "test-seat", base: discord.base, pollMs: 100 }, log: () => {} }); + t.after(async () => (await started.catch(() => null))?.close(0)); + await assert.rejects(started, (e) => e.exitCode === 1 && e.message === "broker failed: write EPIPE"); + assert.ok(children.every(gone), "the broker and the notifier have ended"); + assert.equal(existsSync(hostLock(f.dataRoot)), false); + assert.equal(existsSync(hostFile(f.dataRoot)), false); +}); + +for (const [n, name, line] of [[0, "broker", "broker channel broken (broker error: write EPIPE); stopping the host"], [1, "notifier", "notifier error (write EPIPE); stopping the host"]]) + test(`an error event on the ${name} once the host is up stops the host with exit 1, and leaves no host.lock and no host.json`, { timeout: 15000 }, async (t) => { + const root = tmp(t); + const f = fixture(root); + makeDeployment(root, { dmRecipient: IDS.owner }); + const discord = await fakeDiscord(t); + const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env }); + const children = forksSeen(t); + const logs = []; + const host = await startHost({ boot, business: "acme", notifier: { binding: "test-seat", base: discord.base, pollMs: 100 }, log: (l) => logs.push(l) }); + t.after(() => host.close(0)); + children[n].emit("error", new Error("write EPIPE")); + assert.equal(await host.done, 1); + assert.ok(logs.includes(line), logs.join("\n")); + assert.ok(children.every(gone), "the broker and the notifier have ended"); + assert.equal(existsSync(hostLock(f.dataRoot)), false); + assert.equal(existsSync(hostFile(f.dataRoot)), false); + }); + test("a broker reply that misses the wait breaks the channel: the late reply answers nothing, later requests refuse, the host exits 1", { timeout: 30000 }, async (t) => { const root = tmp(t); const f = fixture(root);