jason.woltje
4c8bf9ca4e
PR #1916 rule: a refusal names the field or file, never a config value or a host path. readJsonFile's three refusals and parseJsonFileText now name the file by its `what` label only. Callers whose label alone is ambiguous add a name that already passed a strict pattern (business id, role name, org id, turn id, project key) or a fixed path under the data root. Exit codes, refusal codes and check order are unchanged. organization: util, access, project, config-command, library-seats, org-service deleteOrg receipt, agent-levels regex; T1 files role.mjs and organization-config.mjs change only the readJsonFile label. cli: access lock, accounts owner, host state/lock, notifier journal, notify-config, run-evidence, dev-fixture, git-socket, org-bus, org-console. discord: turn record label names the turn id. Sentinel tests (sk-test-NOT-A-SECRET-1234 as a path component) cover each changed refusal and fail on BASE; old path pins updated.
1008 lines
54 KiB
JavaScript
1008 lines
54 KiB
JavaScript
import { test } from "node:test";
|
|
import assert from "node:assert/strict";
|
|
import { spawn, spawnSync } from "node:child_process";
|
|
import { subscribe, unsubscribe } from "node:diagnostics_channel";
|
|
import { createServer } from "node:http";
|
|
import { once } from "node:events";
|
|
import { chmodSync, existsSync, readFileSync, readdirSync, rmSync, statSync, symlinkSync, utimesSync, writeFileSync, mkdirSync, renameSync } from "node:fs";
|
|
import { dirname, join } from "node:path";
|
|
import { fileURLToPath } from "node:url";
|
|
import { Client } from "../../bus/src/client.mjs";
|
|
import { bootConfig, loadSystem, REPO } from "../src/config.mjs";
|
|
import { claimHostLock, hostDir, hostFile, hostLock, hostStatus, readHostState, startHost, startTimeOf, stopHost, watchChildren } from "../src/host.mjs";
|
|
import { journalPath } from "../src/notifier.mjs";
|
|
import { reserveOrg, activateOrg } from "../../organization/src/org-registry.mjs";
|
|
import { busStart, busStatus, busStop } from "../src/org-bus.mjs";
|
|
import { IDS, makeDeployment } from "../../discord/tests/helpers.mjs";
|
|
import { fixture, notifyConfig, OPTIONS, tmp } from "./helpers.mjs";
|
|
|
|
const CLI = fileURLToPath(new URL("../src/cli.mjs", import.meta.url));
|
|
const CHANNEL = "100000000000000900";
|
|
const TOKEN = "MTAw.abcdefghijklmnopqrstuvwxyz0123456789";
|
|
|
|
// A named scratch instance with an active organization. Keep the old fixture's
|
|
// business and role documents; only the instance/organization layout changes.
|
|
async function orgFixture(t, { notify = true } = {}) {
|
|
const root = tmp(t), f = fixture(root), instance = "hosttest";
|
|
const dir = join(root, ".config", `mosaic-${instance}`);
|
|
mkdirSync(dirname(dir), { recursive: true, mode: 0o700 });
|
|
renameSync(join(root, "config"), dir);
|
|
chmodSync(dir, 0o700); // S0 requires private instance configuration directories.
|
|
const env = { ...f.env, HOME: root, MOSAIC_CONFIG: join(dir, "config.json") };
|
|
writeFileSync(join(f.dataRoot, "instance.json"), JSON.stringify({ instanceVersion: 1, name: instance }), { mode: 0o600 });
|
|
const ctx = { configDir: dir, dataRoot: f.dataRoot, instance, env };
|
|
const org = await reserveOrg(ctx, "acme"); await activateOrg(ctx, "acme");
|
|
if (notify) notifyConfig(org.root, "acme", null);
|
|
return { ...f, env, ctx, instanceRoot: f.dataRoot, dataRoot: org.root };
|
|
}
|
|
|
|
// A fake Discord REST on 127.0.0.1: opens one DM channel, accepts messages.
|
|
async function fakeDiscord(t) {
|
|
const requests = [];
|
|
let n = 0;
|
|
const server = createServer((req, res) => {
|
|
let body = "";
|
|
req.on("data", (b) => (body += b));
|
|
req.on("end", () => {
|
|
requests.push({ method: req.method, url: req.url, body: body ? JSON.parse(body) : null, authorized: req.headers.authorization === `Bot ${TOKEN}` });
|
|
res.setHeader("content-type", "application/json");
|
|
if (req.url === "/users/@me/channels") return res.end(JSON.stringify({ id: CHANNEL, type: 1 }));
|
|
if (req.url === `/channels/${CHANNEL}/messages`) return res.end(JSON.stringify({ id: `30000000000000${String(++n).padStart(4, "0")}` }));
|
|
res.statusCode = 404;
|
|
res.end("{}");
|
|
});
|
|
});
|
|
server.listen(0, "127.0.0.1");
|
|
await once(server, "listening");
|
|
t.after(() => server.close());
|
|
return { base: `http://127.0.0.1:${server.address().port}`, requests, dms: () => requests.filter((r) => r.url.endsWith("/messages") && r.body.nonce.startsWith("dm")) };
|
|
}
|
|
|
|
async function until(fn, ms = 8000) {
|
|
const end = Date.now() + ms;
|
|
while (Date.now() < end) {
|
|
if (fn()) return;
|
|
await new Promise((r) => setTimeout(r, 50));
|
|
}
|
|
throw new Error("timed out waiting");
|
|
}
|
|
|
|
// Records every IPC message this process sends to a child it creates while
|
|
// the spy is on, by trapping the `send` that node installs on a new child.
|
|
// The host never exposes the notifier's reader capability; this is how a
|
|
// test sees it.
|
|
function spySends(t, onSend = () => {}) {
|
|
const sent = [];
|
|
const onChild = ({ process: child }) => {
|
|
let send;
|
|
Object.defineProperty(child, "send", {
|
|
configurable: true,
|
|
get: () => send,
|
|
set(fn) {
|
|
send = function (m, ...rest) {
|
|
sent.push(m);
|
|
onSend(m);
|
|
return fn.call(this, m, ...rest);
|
|
};
|
|
},
|
|
});
|
|
};
|
|
subscribe("child_process", onChild);
|
|
t.after(() => unsubscribe("child_process", onChild));
|
|
return sent;
|
|
}
|
|
|
|
// Fails this process's sends to new children the way node fails a write to a
|
|
// child that has just died: `fake(child, m)` returns an error to fail that
|
|
// send (to its callback if it has one, otherwise as an 'error' event on the
|
|
// next tick), or nothing to send it for real. Deterministic where the real
|
|
// window after a SIGKILL is a millisecond wide (Darkwing R1, lead decision 73).
|
|
// Every child it saw is killed after the test, so a host left half closed
|
|
// fails the test rather than hang it.
|
|
function failSends(t, fake) {
|
|
const children = [];
|
|
const onChild = ({ process: child }) => {
|
|
children.push(child);
|
|
let send;
|
|
Object.defineProperty(child, "send", {
|
|
configurable: true,
|
|
get: () => send,
|
|
set(fn) {
|
|
send = function (m, ...rest) {
|
|
const err = fake(this, m);
|
|
if (!err) return fn.call(this, m, ...rest);
|
|
const callback = rest.find((a) => typeof a === "function");
|
|
if (callback) process.nextTick(callback, err);
|
|
else process.nextTick(() => this.emit("error", err));
|
|
return false;
|
|
};
|
|
},
|
|
});
|
|
};
|
|
subscribe("child_process", onChild);
|
|
t.after(() => {
|
|
unsubscribe("child_process", onChild);
|
|
for (const child of children) child.kill("SIGKILL");
|
|
});
|
|
}
|
|
|
|
// 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. 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 }) => {
|
|
if (children.push(child) - 1 !== nth) return;
|
|
let send;
|
|
Object.defineProperty(child, "send", {
|
|
configurable: true,
|
|
get: () => send,
|
|
set() {
|
|
if (where === "fork") throw new Error("fork failed");
|
|
send = () => {
|
|
throw new Error("send failed");
|
|
};
|
|
},
|
|
});
|
|
};
|
|
subscribe("child_process", onChild);
|
|
t.after(() => {
|
|
unsubscribe("child_process", onChild);
|
|
for (const child of children) child.kill("SIGKILL");
|
|
});
|
|
}
|
|
|
|
// 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" });
|
|
};
|
|
|
|
const procText = (pid, what) => {
|
|
try {
|
|
return readFileSync(`/proc/${pid}/${what}`, "utf8");
|
|
} catch {
|
|
return "";
|
|
}
|
|
};
|
|
|
|
test("the host boots the broker, binds a launch in process, and the notifier DMs a blocking decision exactly once", 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 });
|
|
assert.equal("trackers" in boot, false);
|
|
const logs = [];
|
|
const sends = spySends(t);
|
|
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));
|
|
const start = sends.find((m) => m?.op === "start");
|
|
assert.equal(typeof start?.cap, "string", "the spy saw the notifier's start message");
|
|
assert.ok(start.cap.length >= 16);
|
|
|
|
const state = JSON.parse(readFileSync(hostFile(f.dataRoot), "utf8"));
|
|
assert.equal(statSync(hostFile(f.dataRoot)).mode & 0o777, 0o600);
|
|
assert.deepEqual(Object.keys(state).sort(), ["business", "hostVersion", "notifier", "pid", "startTime", "startedAt"]);
|
|
assert.equal(hostStatus(f.dataRoot).host.live, true);
|
|
|
|
const launch = await host.bindLaunch({ business: "acme", role: "coder", run: "coder-run", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) });
|
|
assert.equal(launch.run, "coder-run");
|
|
const coder = new Client({ path: host.path, cap: launch.cap });
|
|
await coder.call("role.claim");
|
|
const d = await coder.call("decision.raise", { action: "git.push.protected", target: "refactor", question: "Push?", options: OPTIONS, recommendation: "no", blocking: true, task_ref: "vikunja:1/7" });
|
|
|
|
await until(() => discord.dms().length === 1);
|
|
await new Promise((r) => setTimeout(r, 500));
|
|
assert.equal(discord.dms().length, 1, "five more polls send nothing new");
|
|
assert.ok(discord.requests.every((r) => r.authorized));
|
|
assert.match(discord.dms()[0].body.content, new RegExp(`mosaic decide ${d.id.slice(0, 8)}`));
|
|
|
|
// No capability, the launch's or the notifier's reader, in a child's argv
|
|
// or environment, or in the state file.
|
|
for (const cap of [launch.cap, start.cap]) {
|
|
for (const pid of Object.values(host.pids)) {
|
|
assert.notEqual(procText(pid, "cmdline"), "", `pid ${pid} is readable`);
|
|
assert.ok(!procText(pid, "cmdline").includes(cap));
|
|
assert.ok(!procText(pid, "environ").includes(cap));
|
|
}
|
|
assert.ok(!readFileSync(hostFile(f.dataRoot), "utf8").includes(cap));
|
|
}
|
|
|
|
assert.equal(await host.close(0), 0);
|
|
const journal = readFileSync(journalPath(f.dataRoot, "acme"), "utf8");
|
|
for (const id of [IDS.owner, CHANNEL, TOKEN]) assert.ok(!journal.includes(id));
|
|
assert.equal(journal.trim().split("\n").filter((l) => JSON.parse(l).kind === "dm").length, 1);
|
|
assert.equal(existsSync(hostFile(f.dataRoot)), false);
|
|
assert.equal(existsSync(join(f.dataRoot, "bus", "writer.lock")), false);
|
|
});
|
|
|
|
test("a notifier that dies takes the host down with exit 1, so the unit restarts the pair", 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 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));
|
|
process.kill(host.pids.notifier, "SIGKILL");
|
|
assert.equal(await host.done, 1);
|
|
assert.match(logs.join("\n"), /notifier exited \(SIGKILL\); stopping the host/);
|
|
assert.equal(existsSync(join(f.dataRoot, "bus", "writer.lock")), false);
|
|
assert.equal(existsSync(hostFile(f.dataRoot)), false);
|
|
assert.equal(existsSync(hostLock(f.dataRoot)), false);
|
|
});
|
|
|
|
test("a second host for the same data root refuses with exit 3 while the first runs", async (t) => {
|
|
const root = tmp(t);
|
|
const f = fixture(root);
|
|
makeDeployment(root);
|
|
const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
|
|
const host = await startHost({ boot, business: "acme", log: () => {} });
|
|
t.after(() => host.close(0));
|
|
const second = startHost({ boot, business: "acme", log: () => {} });
|
|
// If the refusal regresses and a second host starts, close it too.
|
|
t.after(async () => (await second.catch(() => null))?.close(0));
|
|
await assert.rejects(second, (e) => e.exitCode === 3 && e.message === `a bus host already runs for acme (pid ${process.pid})`);
|
|
assert.equal(hostStatus(f.dataRoot).host.live, true, "the first host still runs");
|
|
assert.equal(await host.close(0), 0);
|
|
});
|
|
|
|
test("two starts racing for one organization leave exactly one host", async t => {
|
|
const f = await orgFixture(t);
|
|
const results = await Promise.allSettled([busStart(f.ctx, "acme", { detach: true }), busStart(f.ctx, "acme", { detach: true })]);
|
|
t.after(async () => { try { await busStop(f.ctx, "acme"); } catch {} });
|
|
assert.equal(results.filter(r => r.status === "fulfilled").length, 1);
|
|
assert.equal(results.filter(r => r.status === "rejected").length, 1);
|
|
assert.equal((await busStatus(f.ctx, "acme")).healthy, true);
|
|
assert.equal((await busStop(f.ctx, "acme")).stopped, true);
|
|
});
|
|
|
|
test("a host takes over a host.lock its dead host left, and puts back a live claim that replaced it meanwhile", (t) => {
|
|
const dataRoot = tmp(t);
|
|
mkdirSync(hostDir(dataRoot), { recursive: true, mode: 0o700 });
|
|
const stale = { pid: 999999999, startTime: "1", business: "acme", startedAt: "2026-10-13T00:00:00.000Z" };
|
|
const mine = { hostVersion: 1, pid: process.pid, startTime: startTimeOf(process.pid), business: "acme", startedAt: "2026-10-13T00:00:01.000Z", notifier: null };
|
|
const left = () => readdirSync(hostDir(dataRoot)).sort();
|
|
const held = () => JSON.parse(readFileSync(hostLock(dataRoot), "utf8"));
|
|
|
|
writeFileSync(hostLock(dataRoot), JSON.stringify(stale), { mode: 0o600 });
|
|
claimHostLock(dataRoot, mine);
|
|
assert.equal(held().startedAt, mine.startedAt);
|
|
assert.equal(statSync(hostLock(dataRoot)).mode & 0o777, 0o600);
|
|
assert.deepEqual(left(), ["host.lock"]);
|
|
|
|
// Another host removes the stale lock and takes it between this host's
|
|
// read and its move: that claim goes back, and this host refuses.
|
|
const other = { ...mine, business: "beta", startedAt: "2026-10-13T00:00:02.000Z" };
|
|
writeFileSync(hostLock(dataRoot), JSON.stringify(stale), { mode: 0o600 });
|
|
const race = () => writeFileSync(hostLock(dataRoot), JSON.stringify(other), { mode: 0o600 });
|
|
assert.throws(() => claimHostLock(dataRoot, mine, { between: race }), (e) => e.exitCode === 3 && e.message === `a bus host already runs for beta (pid ${process.pid})`);
|
|
assert.equal(held().startedAt, other.startedAt);
|
|
assert.deepEqual(left(), ["host.lock"]);
|
|
});
|
|
|
|
// Row 106 (#1600, Dewey a): a lock removed after the check and before the
|
|
// read is tried again, within CLAIM_TRIES, and never refused as unreadable.
|
|
test("a host.lock that goes between the check and the read is tried again; one that can't be parsed still refuses as unreadable", (t) => {
|
|
const dataRoot = tmp(t);
|
|
mkdirSync(hostDir(dataRoot), { recursive: true, mode: 0o700 });
|
|
const stale = { pid: 999999999, startTime: "1", business: "acme", startedAt: "2026-10-13T00:00:00.000Z" };
|
|
const mine = { hostVersion: 1, pid: process.pid, startTime: startTimeOf(process.pid), business: "acme", startedAt: "2026-10-13T00:00:01.000Z", notifier: null };
|
|
writeFileSync(hostLock(dataRoot), JSON.stringify(stale), { mode: 0o600 });
|
|
let found = 0;
|
|
const gone = () => found++ === 0 && rmSync(hostLock(dataRoot));
|
|
claimHostLock(dataRoot, mine, { found: gone });
|
|
assert.equal(found, 1, "the lock was found once, and gone before the read");
|
|
assert.equal(JSON.parse(readFileSync(hostLock(dataRoot), "utf8")).startedAt, mine.startedAt);
|
|
assert.deepEqual(readdirSync(hostDir(dataRoot)), ["host.lock"]);
|
|
|
|
writeFileSync(hostLock(dataRoot), "{not json", { mode: 0o600 });
|
|
assert.throws(() => claimHostLock(dataRoot, mine), (e) => e.exitCode === 3 && e.message === "bus host state is unreadable");
|
|
assert.equal(readFileSync(hostLock(dataRoot), "utf8"), "{not json");
|
|
assert.deepEqual(readdirSync(hostDir(dataRoot)), ["host.lock"]);
|
|
});
|
|
|
|
test("a live host's lock stays, however old the file is", (t) => {
|
|
const dataRoot = tmp(t);
|
|
mkdirSync(hostDir(dataRoot), { recursive: true, mode: 0o700 });
|
|
const live = { hostVersion: 1, pid: process.pid, startTime: startTimeOf(process.pid), business: "beta", startedAt: "2000-01-01T00:00:00.000Z", notifier: null };
|
|
const text = JSON.stringify(live);
|
|
writeFileSync(hostLock(dataRoot), text, { mode: 0o600 });
|
|
const old = new Date("2000-01-01T00:00:00Z");
|
|
utimesSync(hostLock(dataRoot), old, old);
|
|
const mine = { ...live, business: "acme", startedAt: "2026-10-13T00:00:01.000Z" };
|
|
assert.throws(() => claimHostLock(dataRoot, mine), (e) => e.exitCode === 3 && e.message === `a bus host already runs for beta (pid ${process.pid})`);
|
|
assert.equal(readFileSync(hostLock(dataRoot), "utf8"), text);
|
|
assert.equal(statSync(hostLock(dataRoot)).mtimeMs, old.getTime());
|
|
assert.deepEqual(readdirSync(hostDir(dataRoot)), ["host.lock"]);
|
|
});
|
|
|
|
test("a stale organization host lock is preserved and refuses another start", async t => {
|
|
const f = await orgFixture(t), receipt = await busStart(f.ctx, "acme", { detach: true });
|
|
process.kill(receipt.pid, "SIGKILL");
|
|
await until(() => startTimeOf(receipt.pid) === null, 30000);
|
|
await until(() => !existsSync(join(f.dataRoot, "bus", "writer.lock")), 30000);
|
|
const before = readFileSync(hostLock(f.dataRoot), "utf8");
|
|
await assert.rejects(busStart(f.ctx, "acme", { detach: true }), /stale.*lock/i);
|
|
assert.equal(readFileSync(hostLock(f.dataRoot), "utf8"), before);
|
|
});
|
|
|
|
test("while an organization host boots, stop and status refuse its unmatched lock", async t => {
|
|
const f = await orgFixture(t);
|
|
const booting = { hostVersion: 1, pid: process.pid, startTime: startTimeOf(process.pid), business: "acme", startedAt: "2026-10-13T00:00:01.000Z", notifier: null };
|
|
claimHostLock(f.dataRoot, booting);
|
|
assert.equal(readHostState(f.dataRoot), null);
|
|
await assert.rejects(busStatus(f.ctx, "acme"), /lock/i);
|
|
await assert.rejects(busStop(f.ctx, "acme"), /lock/i);
|
|
const inbox = spawnSync(process.execPath, [CLI, "inbox"], { env: f.env, encoding: "utf8", timeout: 20000 });
|
|
assert.notEqual(inbox.status, 0);
|
|
assert.match(inbox.stderr, /name the organization/);
|
|
});
|
|
|
|
test("a host started before host.lock existed, holding only host.json, still refuses a second start, which leaves no lock", async (t) => {
|
|
const root = tmp(t);
|
|
const f = fixture(root);
|
|
makeDeployment(root);
|
|
mkdirSync(hostDir(f.dataRoot), { recursive: true, mode: 0o700 });
|
|
const legacy = { hostVersion: 1, pid: process.pid, startTime: startTimeOf(process.pid), business: "beta", startedAt: "2026-10-13T00:00:01.000Z", notifier: null };
|
|
writeFileSync(hostFile(f.dataRoot), JSON.stringify(legacy), { mode: 0o600 });
|
|
const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
|
|
const started = startHost({ boot, business: "acme", log: () => {} });
|
|
t.after(async () => (await started.catch(() => null))?.close(0));
|
|
await assert.rejects(started, (e) => e.exitCode === 3 && e.message === `a bus host already runs for beta (pid ${process.pid})`);
|
|
assert.deepEqual(readdirSync(hostDir(f.dataRoot)), ["host.json"]);
|
|
});
|
|
|
|
// Row 106 (#1600, Sage N1): a throw after the claim and before the boot
|
|
// reply gives host.lock back like a refusal does.
|
|
for (const where of ["fork", "send"])
|
|
test(`a broker ${where} that throws leaves no host.lock and no host.json`, async (t) => {
|
|
const root = tmp(t);
|
|
const f = fixture(root);
|
|
makeDeployment(root);
|
|
const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
|
|
throwOn(t, where);
|
|
const started = startHost({ boot, business: "acme", log: () => {} });
|
|
t.after(async () => (await started.catch(() => null))?.close(0));
|
|
await assert.rejects(started, (e) => e.message === `${where} failed`);
|
|
assert.equal(existsSync(hostLock(f.dataRoot)), false);
|
|
assert.equal(existsSync(hostFile(f.dataRoot)), false);
|
|
});
|
|
|
|
test("a broker that misses its boot wait leaves no host.lock and no host.json", async (t) => {
|
|
const root = tmp(t);
|
|
const f = fixture(root);
|
|
makeDeployment(root);
|
|
const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
|
|
const started = startHost({ boot, business: "acme", bootTimeoutMs: 1, log: () => {} });
|
|
t.after(async () => (await started.catch(() => null))?.close(0));
|
|
await assert.rejects(started, (e) => /broker/.test(e.message));
|
|
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 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);
|
|
makeDeployment(root);
|
|
const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
|
|
const logs = [];
|
|
const host = await startHost({ boot, business: "acme", requestTimeoutMs: 1000, log: (l) => logs.push(l) });
|
|
// Hooks run in order: the broker resumes before the close, which a
|
|
// stopped broker would hold.
|
|
t.after(() => {
|
|
try {
|
|
process.kill(host.pids.broker, "SIGCONT");
|
|
} catch {}
|
|
});
|
|
t.after(() => host.close(0));
|
|
const record = (role, run) => ({ business: "acme", role, run, harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) });
|
|
const coder = await host.bindLaunch(record("coder", "coder-run"));
|
|
assert.equal((await host.op({ op: "identity", cap: coder.cap })).role, "coder");
|
|
|
|
// Stall the broker with a request waiting and a bind queued behind it.
|
|
process.kill(host.pids.broker, "SIGSTOP");
|
|
const code = (p) => p.then((v) => ({ answered: v }), (e) => e.code);
|
|
const first = code(host.op({ op: "identity", cap: "bogus" }));
|
|
const second = code(host.bindLaunch(record("reviewer", "reviewer-run")));
|
|
assert.equal(await first, "broker-channel-broken");
|
|
assert.equal(await second, "broker-channel-broken", "a queued request never reaches the broker");
|
|
process.kill(host.pids.broker, "SIGCONT");
|
|
|
|
// The broker answers the stalled request late; nothing takes that reply.
|
|
assert.equal(await host.done, 1);
|
|
assert.ok(logs.some((l) => /^broker channel broken \(no reply within 1 s\); stopping the host$/.test(l)), logs.join("\n"));
|
|
assert.equal(await code(host.op({ op: "identity", cap: coder.cap })), "broker-channel-broken");
|
|
});
|
|
|
|
test("a broker reply with another request's id, or none, breaks the channel and the host exits 1", { timeout: 30000 }, async (t) => {
|
|
// The broker echoes whatever id it's sent, so changing the id on the way
|
|
// out gives the host the reply a confused broker would send.
|
|
let what, tamper;
|
|
spySends(t, (m) => m?.cap === "tamper" && tamper(m));
|
|
for ([what, tamper] of [
|
|
["another id", (m) => (m.id += 1000)],
|
|
["no id", (m) => delete m.id],
|
|
]) {
|
|
const root = tmp(t);
|
|
const f = fixture(root);
|
|
makeDeployment(root);
|
|
const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
|
|
const logs = [];
|
|
const host = await startHost({ boot, business: "acme", log: (l) => logs.push(l) });
|
|
t.after(() => host.close(0));
|
|
const record = (role, run) => ({ business: "acme", role, run, harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) });
|
|
const coder = await host.bindLaunch(record("coder", "coder-run"));
|
|
await assert.rejects(host.op({ op: "identity", cap: "tamper" }), (e) => e.code === "broker-channel-broken", what);
|
|
await assert.rejects(host.op({ op: "identity", cap: coder.cap }), (e) => e.code === "broker-channel-broken", what);
|
|
assert.equal(await host.done, 1, what);
|
|
assert.ok(logs.includes("broker channel broken (reply for another request); stopping the host"), `${what}: ${logs.join("\n")}`);
|
|
}
|
|
});
|
|
|
|
test("a broker reply with no request waiting breaks the channel and the host exits 1", { timeout: 30000 }, async (t) => {
|
|
// A request sent to the real broker past the host's queue: its reply
|
|
// arrives when nothing is waiting (Darkwing round 2, Mr).
|
|
const children = [];
|
|
const onChild = ({ process: child }) => children.push(child);
|
|
subscribe("child_process", onChild);
|
|
t.after(() => unsubscribe("child_process", onChild));
|
|
const root = tmp(t);
|
|
const f = fixture(root);
|
|
makeDeployment(root);
|
|
const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
|
|
const logs = [];
|
|
const host = await startHost({ boot, business: "acme", log: (l) => logs.push(l) });
|
|
t.after(() => host.close(0));
|
|
const record = { business: "acme", role: "coder", run: "coder-run", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) };
|
|
const coder = await host.bindLaunch(record);
|
|
children.find((c) => c.pid === host.pids.broker).send({ op: "identity", cap: coder.cap, id: 1_000_000 });
|
|
const late = new Promise((r) => setTimeout(r, 5000, "still running").unref());
|
|
assert.equal(await Promise.race([host.done, late]), 1);
|
|
assert.ok(logs.includes("broker channel broken (reply with no request waiting); stopping the host"), logs.join("\n"));
|
|
await assert.rejects(host.op({ op: "identity", cap: coder.cap }), (e) => e.code === "broker-channel-broken");
|
|
});
|
|
|
|
test("a notifier that refuses stops the broker and the host refuses with exit 3", async (t) => {
|
|
const root = tmp(t);
|
|
const f = fixture(root);
|
|
makeDeployment(root);
|
|
const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
|
|
const started = startHost({ boot, business: "acme", notifier: { binding: "test-seat" }, log: () => {} });
|
|
// If the refusal regresses, the host starts; close it so the file still ends.
|
|
t.after(async () => (await started.catch(() => null))?.close(0));
|
|
await assert.rejects(started, (e) => e.exitCode === 3 && /no dmRecipient/.test(e.message));
|
|
assert.equal(existsSync(join(f.dataRoot, "bus", "writer.lock")), false);
|
|
assert.equal(existsSync(hostFile(f.dataRoot)), false);
|
|
assert.equal(existsSync(hostLock(f.dataRoot)), false);
|
|
});
|
|
|
|
test("a notifier that refuses after the broker died still refuses with exit 3, without a send to the dead broker", async (t) => {
|
|
const root = tmp(t);
|
|
const f = fixture(root);
|
|
makeDeployment(root);
|
|
const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
|
|
const children = [];
|
|
const onChild = ({ process: child }) => children.push(child);
|
|
subscribe("child_process", onChild);
|
|
t.after(() => unsubscribe("child_process", onChild));
|
|
const errors = [];
|
|
// The broker dies as the host sends the notifier its start message.
|
|
const sends = spySends(t, (m) => {
|
|
if (m?.op !== "start") return;
|
|
children[0].on("error", (e) => errors.push(e.code));
|
|
children[0].kill("SIGKILL");
|
|
});
|
|
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 === 3 && /no dmRecipient/.test(e.message));
|
|
await new Promise((r) => setImmediate(r));
|
|
assert.deepEqual(errors, [], "no close was sent over the closed channel");
|
|
assert.equal(sends.filter((m) => m?.op === "close").length, 0);
|
|
});
|
|
|
|
test("a notifier that dies before it replies, after the broker died, still refuses, without a send to the dead broker", async (t) => {
|
|
const root = tmp(t);
|
|
const f = fixture(root);
|
|
makeDeployment(root);
|
|
const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
|
|
const children = [];
|
|
const onChild = ({ process: child }) => children.push(child);
|
|
subscribe("child_process", onChild);
|
|
t.after(() => unsubscribe("child_process", onChild));
|
|
const errors = [];
|
|
// The broker dies as the host sends the start message. The notifier is
|
|
// stopped so it cannot reply, and killed once the host has seen the broker
|
|
// disconnect: the host takes the no-reply path, not the refusal path.
|
|
const sends = spySends(t, (m) => {
|
|
if (m?.op !== "start") return;
|
|
children[0].on("error", (e) => errors.push(e.code));
|
|
children[1].kill("SIGSTOP");
|
|
children[0].once("disconnect", () => children[1].kill("SIGKILL"));
|
|
children[0].kill("SIGKILL");
|
|
});
|
|
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 && /notifier exited \(null\) before it replied/.test(e.message));
|
|
await new Promise((r) => setImmediate(r));
|
|
assert.deepEqual(errors, [], "no close was sent over the closed channel");
|
|
assert.equal(sends.filter((m) => m?.op === "close").length, 0);
|
|
assert.equal(existsSync(hostFile(f.dataRoot)), false);
|
|
assert.equal(existsSync(hostLock(f.dataRoot)), false);
|
|
});
|
|
|
|
test("a close send that fails with EPIPE after the notifier refuses still gives the notifier's refusal, exit 3", async (t) => {
|
|
const root = tmp(t);
|
|
const f = fixture(root);
|
|
makeDeployment(root);
|
|
const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
|
|
failSends(t, (child, m) => (m?.op === "close" ? epipe(child) : 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 === 3 && /^notifier refused to start: .*no dmRecipient/.test(e.message));
|
|
});
|
|
|
|
test("a close send that fails with EPIPE after the notifier dies unanswered still gives the notifier's error", async (t) => {
|
|
const root = tmp(t);
|
|
const f = fixture(root);
|
|
makeDeployment(root);
|
|
const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
|
|
failSends(t, (child, m) => {
|
|
if (m?.op === "close") return epipe(child);
|
|
if (m?.op === "start") {
|
|
// Stopped, it cannot reply; then it dies.
|
|
child.kill("SIGSTOP");
|
|
setImmediate(() => child.kill("SIGKILL"));
|
|
}
|
|
return 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 && /^notifier exited \(null\) before it replied/.test(e.message));
|
|
});
|
|
|
|
test("close() whose stop and close sends fail with EPIPE still finishes, with exit 1", 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 });
|
|
failSends(t, (child, m) => (m?.op === "stop" || m?.op === "close" ? epipe(child) : null));
|
|
const host = await startHost({ boot, business: "acme", notifier: { binding: "test-seat", base: discord.base, pollMs: 100 }, log: () => {} });
|
|
// No t.after close: a close() that rejected leaves `done` pending forever.
|
|
assert.equal(await host.close(0), 1, "both children died by signal");
|
|
assert.equal(existsSync(hostFile(f.dataRoot)), false);
|
|
});
|
|
|
|
test("watchChildren reports a child that died before it was called, and one that dies later", async (t) => {
|
|
const early = spawn(process.execPath, ["-e", "process.exit(7)"], { stdio: "ignore" });
|
|
await once(early, "exit");
|
|
const killed = spawn(process.execPath, ["-e", "setTimeout(() => {}, 60000)"], { stdio: "ignore" });
|
|
await once(killed, "spawn");
|
|
killed.kill("SIGKILL");
|
|
await once(killed, "exit");
|
|
const signalled = [];
|
|
watchChildren({ broker: killed }, (...d) => signalled.push(d));
|
|
assert.deepEqual(signalled, [["broker", null, "SIGKILL"]], "a death by signal before the watch is not lost either");
|
|
const late = spawn(process.execPath, ["-e", "setTimeout(() => {}, 60000)"], { stdio: "ignore" });
|
|
t.after(() => late.kill("SIGKILL"));
|
|
await once(late, "spawn");
|
|
const deaths = [];
|
|
watchChildren({ broker: early, notifier: late, none: null }, (...d) => deaths.push(d));
|
|
assert.deepEqual(deaths, [["broker", 7, null]], "the exit before the watch is not lost");
|
|
late.kill("SIGTERM");
|
|
await once(late, "exit");
|
|
assert.deepEqual(deaths, [["broker", 7, null], ["notifier", null, "SIGTERM"]]);
|
|
});
|
|
|
|
test("bus stop refuses to signal a live pid that is not a bus host", async (t) => {
|
|
const dataRoot = tmp(t);
|
|
const child = spawn(process.execPath, ["-e", "setTimeout(() => {}, 60000)"], { stdio: "ignore" });
|
|
t.after(() => child.kill("SIGKILL"));
|
|
await once(child, "spawn");
|
|
mkdirSync(hostDir(dataRoot), { recursive: true, mode: 0o700 });
|
|
writeFileSync(hostFile(dataRoot), JSON.stringify({ pid: child.pid, startTime: startTimeOf(child.pid), business: "acme" }), { mode: 0o600 });
|
|
await assert.rejects(stopHost(dataRoot, { timeoutMs: 1000 }), (e) => e.exitCode === 3 && /is not a bus host; refusing to signal it/.test(e.message));
|
|
await new Promise((r) => setTimeout(r, 200));
|
|
assert.equal(child.exitCode, null);
|
|
assert.equal(child.signalCode, null, "the child was not signalled");
|
|
});
|
|
|
|
test("a stale organization writer lock refuses start and status without removing it", async t => {
|
|
const f = await orgFixture(t);
|
|
mkdirSync(join(f.dataRoot, "bus"), { recursive: true, mode: 0o700 });
|
|
const file = join(f.dataRoot, "bus", "writer.lock"), text = JSON.stringify({ pid: 999999999, at: "2026-10-08T00:00:00Z" });
|
|
writeFileSync(file, text, { mode: 0o600 });
|
|
await assert.rejects(busStart(f.ctx, "acme", { detach: true }), /stale.*lock/i);
|
|
await assert.rejects(busStatus(f.ctx, "acme"), /stale.*lock/i);
|
|
assert.equal(readFileSync(file, "utf8"), text);
|
|
});
|
|
|
|
test("organization bus start refuses without a notifier config", async t => {
|
|
// PR #1916 (org-redact5): the refusal names the log inside the organization's
|
|
// runtime directory, never by host path. (A sentinel directory in the root would
|
|
// push the bus socket path past the socket length limit, so the check is on the
|
|
// fixture's own roots.)
|
|
const f = await orgFixture(t, { notify: false });
|
|
await assert.rejects(busStart(f.ctx, "acme", { detach: true }), (e) => e.exitCode === 3
|
|
&& e.message === "bus failed to start; see bus-host/host.log in the runtime directory of organization acme"
|
|
&& !e.message.includes(f.dataRoot) && !e.message.includes(f.instanceRoot));
|
|
assert.match(readFileSync(join(f.dataRoot, "bus-host", "host.log"), "utf8"), /no notifier config/);
|
|
assert.equal(existsSync(join(f.dataRoot, "bus", "writer.lock")), false);
|
|
});
|
|
|
|
test("organization bus start, status and stop report the same live host", async t => {
|
|
const f = await orgFixture(t);
|
|
const started = await busStart(f.ctx, "acme", { detach: true });
|
|
t.after(async () => { try { await busStop(f.ctx, "acme"); } catch {} });
|
|
const status = await busStatus(f.ctx, "acme");
|
|
assert.equal(status.healthy, true); assert.equal(status.pid, started.pid);
|
|
assert.equal((await busStop(f.ctx, "acme")).stopped, true);
|
|
assert.equal((await busStatus(f.ctx, "acme")).healthy, false);
|
|
assert.equal(existsSync(join(f.dataRoot, "bus", "writer.lock")), false);
|
|
});
|
|
|
|
test("organization bus --no-launch preserves the launch fence; --pm conflicts", async t => {
|
|
const f = await orgFixture(t);
|
|
const both = spawnSync(process.execPath, [CLI, "bus", "start", "acme", "--pm", "--no-launch"], { env: f.env, encoding: "utf8" });
|
|
assert.equal(both.status, 4); assert.match(both.stderr, /--pm or --no-launch/);
|
|
const receipt = await busStart(f.ctx, "acme", { detach: true, noLaunch: true });
|
|
t.after(async () => { try { await busStop(f.ctx, "acme"); } catch {} });
|
|
assert.equal(readHostState(f.dataRoot).options.noLaunch, true);
|
|
assert.equal(existsSync(join(f.dataRoot, "bus-host", "sessions.json")), false);
|
|
assert.equal((await busStatus(f.ctx, "acme")).pid, receipt.pid);
|
|
assert.equal((await busStop(f.ctx, "acme")).stopped, true);
|
|
});
|
|
|
|
test("bus-service.sh renders the unit and installs it into a given directory", (t) => {
|
|
const dir = tmp(t);
|
|
const script = join(REPO, "scripts", "bus-service.sh");
|
|
// Row 190: the script selects an instance from the system config first; a
|
|
// scratch path keeps it off ~/.config/mosaic-dev. A missing config selects
|
|
// the default names, as before.
|
|
const cfgEnv = { ...process.env, MOSAIC_CONFIG: join(dir, "absent", "config.json") };
|
|
const render = spawnSync(script, ["render"], { encoding: "utf8", env: cfgEnv });
|
|
assert.equal(render.status, 0, render.stderr);
|
|
assert.match(render.stdout, new RegExp(`ExecStart=${REPO.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}/scripts/mosaic bus start %i`));
|
|
assert.match(render.stdout, /RestartPreventExitStatus=2 3 4/);
|
|
// Row 163: --no-launch is for the D2 rehearsal's own host only; no unit starts a bus with it.
|
|
assert.doesNotMatch(render.stdout, /--no-launch/);
|
|
// One session credential file, optional so Pi sessions start without it.
|
|
// The README's paragraph names the same file; nothing else does.
|
|
const files = render.stdout.match(/^EnvironmentFile=.*$/gm);
|
|
assert.equal(files.length, 1, files.join("\n"));
|
|
assert.match(files[0], /^EnvironmentFile=-%h\/\S+$/);
|
|
assert.ok(readFileSync(join(REPO, "packages", "cli", "README.md"), "utf8").includes(`\`${files[0]}\``), "the README names the unit's file");
|
|
assert.doesNotMatch(render.stdout, /@REPO@|@PATH@|@CLAUDE_ENV@/);
|
|
// A user unit cannot order on a system target (Darkwing F3).
|
|
assert.doesNotMatch(render.stdout, /network-online/);
|
|
const fakeBin=join(dir,"bin");mkdirSync(fakeBin);
|
|
writeFileSync(join(fakeBin,"systemctl"),"#!/bin/sh\ncase \"$*\" in *list-units*) exit 0;; *)exit 99;; esac\n");chmodSync(join(fakeBin,"systemctl"),0o755);
|
|
writeFileSync(join(fakeBin,"systemd-analyze"),`#!/bin/sh\nprintf '%s\\n' '${dir}/unit-search'\n`);chmodSync(join(fakeBin,"systemd-analyze"),0o755);
|
|
const unitEnv={...cfgEnv,PATH:`${fakeBin}:${process.env.PATH}`};
|
|
const first = spawnSync(script, ["install", "--dir", dir, "--no-reload"], { encoding: "utf8",env:unitEnv });
|
|
assert.equal(first.status, 0, first.stderr);
|
|
assert.match(first.stdout, /written: /);
|
|
assert.match(first.stdout, /^ {2}mkdir -m 0700 -p <dataRoot>\/notify\/<business> +the notifier refuses a looser directory$/m);
|
|
assert.equal(readFileSync(join(dir, "[email protected]"), "utf8"), render.stdout);
|
|
assert.match(spawnSync(script, ["install", "--dir", dir, "--no-reload"], { encoding: "utf8",env:unitEnv }).stdout, /unchanged: /);
|
|
assert.match(spawnSync(script, ["uninstall", "--dir", dir, "--no-reload"], { encoding: "utf8",env:unitEnv }).stdout, /removed: /);
|
|
assert.equal(spawnSync(script, ["bogus"], { encoding: "utf8", env: cfgEnv }).status, 4);
|
|
});
|
|
|
|
// Row 68: the unit pins the Claude Code binary as MOSAIC_CLAUDE_BIN, from
|
|
// --claude or else `command -v claude`, kept as given (not resolved).
|
|
test("bus-service.sh pins the claude binary, refuses one the unit can't carry, and renders without one", (t) => {
|
|
const dir = tmp(t);
|
|
const script = join(REPO, "scripts", "bus-service.sh");
|
|
const stand = (p, mode = 0o755) => {
|
|
writeFileSync(p, '#!/bin/sh\necho "9.9.9 (Claude Code)"\n');
|
|
chmodSync(p, mode);
|
|
return p;
|
|
};
|
|
mkdirSync(join(dir, "bin"));
|
|
writeFileSync(join(dir,"bin","systemctl"),'#!/bin/sh\ncase "$*" in *list-units*) exit 0;; *)exit 99;; esac\n');chmodSync(join(dir,"bin","systemctl"),0o755);
|
|
writeFileSync(join(dir,"bin","systemd-analyze"),`#!/bin/sh\nprintf '%s\\n' '${dir}/unit-search'\n`);chmodSync(join(dir,"bin","systemd-analyze"),0o755);
|
|
const onPath = stand(join(dir, "bin", "claude"));
|
|
const named = stand(join(dir, "claude-named"));
|
|
symlinkSync(named, join(dir, "claude-link"));
|
|
// Row 190: a scratch MOSAIC_CONFIG keeps the instance selection off ~/.config/mosaic-dev.
|
|
const env = { ...process.env, MOSAIC_CONFIG: join(dir, "absent", "config.json"), PATH: `${join(dir, "bin")}:${dirname(process.execPath)}:/usr/bin:/bin` };
|
|
const run = (args, e = env) => spawnSync(script, args, { encoding: "utf8", env: e });
|
|
const pinned = (out) => out.match(/^Environment=MOSAIC_CLAUDE_BIN=(.*)$/m)?.[1];
|
|
|
|
assert.equal(pinned(run(["render"]).stdout), onPath);
|
|
assert.equal(pinned(run(["render", "--claude", named]).stdout), named);
|
|
assert.equal(pinned(run(["render", "--claude", join(dir, "claude-link")]).stdout), join(dir, "claude-link"));
|
|
|
|
const refusals = [
|
|
["claude", /must be absolute/],
|
|
[stand(join(dir, "claude-noexec"), 0o644), /not an executable file/],
|
|
[join(dir, "missing"), /not an executable file/],
|
|
[stand(join(dir, "claude%h")), /cannot carry/],
|
|
[stand(join(dir, "claude x")), /cannot carry/],
|
|
];
|
|
for (const [path, why] of refusals) {
|
|
const r = run(["render", "--claude", path]);
|
|
assert.equal(r.status, 1, path);
|
|
assert.match(r.stderr, why, path);
|
|
assert.equal(r.stdout, "", path);
|
|
}
|
|
|
|
// install prints the pin and its version; a refused pin writes nothing.
|
|
const units = join(dir, "units");
|
|
const bad = run(["install", "--dir", units, "--no-reload", "--claude", "claude"]);
|
|
assert.equal(bad.status, 1);
|
|
assert.deepEqual(readdirSync(units), []);
|
|
const ok = run(["install", "--dir", units, "--no-reload", "--claude", named]);
|
|
assert.equal(ok.status, 0, ok.stderr);
|
|
assert.match(ok.stdout, new RegExp(`^claude: ${named.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")} \\(9\\.9\\.9 \\(Claude Code\\)\\)$`, "m"));
|
|
assert.equal(pinned(readFileSync(join(units, "[email protected]"), "utf8")), named);
|
|
assert.equal(run(["uninstall", "--dir", units, "--no-reload", "--claude", named]).status, 4);
|
|
assert.equal(run(["render", "--dir", units]).status, 4);
|
|
|
|
// No claude anywhere on PATH: the unit carries a comment, not the variable.
|
|
const tools = join(dir, "tools");
|
|
mkdirSync(tools);
|
|
symlinkSync(process.execPath, join(tools, "node"));
|
|
for (const tool of ["dirname", "sed"]) {
|
|
const at = ["/usr/bin", "/bin"].map((d) => join(d, tool)).find((p) => existsSync(p));
|
|
symlinkSync(at, join(tools, tool));
|
|
}
|
|
const bash = ["/usr/bin/bash", "/bin/bash"].find((p) => existsSync(p));
|
|
const none = spawnSync(bash, [script, "render"], { encoding: "utf8", env: { PATH: tools, HOME: dir } });
|
|
assert.equal(none.status, 0, none.stderr);
|
|
assert.match(none.stderr, /no claude on PATH; MOSAIC_CLAUDE_BIN left unset/);
|
|
assert.equal(pinned(none.stdout), undefined);
|
|
assert.match(none.stdout, /^# MOSAIC_CLAUDE_BIN unset: no claude on PATH at install time$/m);
|
|
assert.doesNotMatch(none.stdout, /@CLAUDE_ENV@/);
|
|
});
|
|
|
|
// Row 72 (#1566): mutants row 68's tests left alive. A directory is -x, so
|
|
// only the -f check keeps a directory pin out of the unit; the default pin
|
|
// keeps a symlink as given, so a claude that updates through it is followed.
|
|
test("bus-service.sh refuses a directory pin and keeps a symlink default unresolved", (t) => {
|
|
const dir = tmp(t);
|
|
const script = join(REPO, "scripts", "bus-service.sh");
|
|
const pinned = (out) => out.match(/^Environment=MOSAIC_CLAUDE_BIN=(.*)$/m)?.[1];
|
|
const tools = join(dir, "tools");
|
|
mkdirSync(tools);
|
|
symlinkSync(process.execPath, join(tools, "node"));
|
|
for (const tool of ["dirname", "sed", "readlink"]) {
|
|
const at = ["/usr/bin", "/bin"].map((d) => join(d, tool)).find((p) => existsSync(p));
|
|
symlinkSync(at, join(tools, tool));
|
|
}
|
|
const bash = ["/usr/bin/bash", "/bin/bash"].find((p) => existsSync(p));
|
|
const run = (args, bin) => spawnSync(bash, [script, ...args], { encoding: "utf8", env: { PATH: `${bin}:${tools}`, HOME: dir } });
|
|
|
|
// --claude DIR: a directory passes -x and fails -f.
|
|
const pinDir = join(dir, "claude-dir");
|
|
mkdirSync(pinDir);
|
|
chmodSync(pinDir, 0o755);
|
|
let r = run(["render", "--claude", pinDir], join(dir, "empty"));
|
|
assert.equal(r.status, 1);
|
|
assert.match(r.stderr, /not an executable file/);
|
|
assert.equal(r.stdout, "");
|
|
|
|
// A directory named claude on PATH: bash's `command -v` never returns a
|
|
// directory, so the default finds no claude and pins nothing. It never
|
|
// pins the directory.
|
|
const dirBin = join(dir, "dir-bin");
|
|
mkdirSync(join(dirBin, "claude"), { recursive: true });
|
|
chmodSync(join(dirBin, "claude"), 0o755);
|
|
r = run(["render"], dirBin);
|
|
assert.equal(r.status, 0, r.stderr);
|
|
assert.match(r.stderr, /no claude on PATH; MOSAIC_CLAUDE_BIN left unset/);
|
|
assert.equal(pinned(r.stdout), undefined);
|
|
assert.ok(!r.stdout.includes(join(dirBin, "claude")));
|
|
|
|
// A symlink named claude on PATH: the pin is the link, not its target.
|
|
const real = join(dir, "versions", "claude-9.9.9");
|
|
mkdirSync(dirname(real));
|
|
writeFileSync(real, '#!/bin/sh\necho "9.9.9 (Claude Code)"\n');
|
|
chmodSync(real, 0o755);
|
|
const linkBin = join(dir, "link-bin");
|
|
mkdirSync(linkBin);
|
|
symlinkSync(real, join(linkBin, "claude"));
|
|
r = run(["render"], linkBin);
|
|
assert.equal(r.status, 0, r.stderr);
|
|
assert.equal(pinned(r.stdout), join(linkBin, "claude"));
|
|
assert.ok(!r.stdout.includes(real), "the target never reaches the unit");
|
|
});
|
|
|
|
// Row 78 (#1572): the default from `command -v claude` takes the same checks
|
|
// as --claude. PATH holds only the test's dirs, so a real claude elsewhere on
|
|
// the host can't stand in for the one under test.
|
|
test("bus-service.sh refuses a default claude without the execute bit, or in a PATH dir with a space", (t) => {
|
|
const dir = tmp(t);
|
|
const script = join(REPO, "scripts", "bus-service.sh");
|
|
const tools = join(dir, "tools");
|
|
mkdirSync(tools);
|
|
symlinkSync(process.execPath, join(tools, "node"));
|
|
for (const tool of ["dirname", "sed", "readlink"]) {
|
|
const at = ["/usr/bin", "/bin"].map((d) => join(d, tool)).find((p) => existsSync(p));
|
|
symlinkSync(at, join(tools, tool));
|
|
}
|
|
const bash = ["/usr/bin/bash", "/bin/bash"].find((p) => existsSync(p));
|
|
const run = (bin) => spawnSync(bash, [script, "render"], { encoding: "utf8", env: { PATH: `${bin}:${tools}`, HOME: dir } });
|
|
const stand = (bin, mode) => {
|
|
mkdirSync(bin);
|
|
writeFileSync(join(bin, "claude"), '#!/bin/sh\necho "9.9.9 (Claude Code)"\n');
|
|
chmodSync(join(bin, "claude"), mode);
|
|
return bin;
|
|
};
|
|
|
|
// The only claude on PATH has no execute bit: `command -v` still returns
|
|
// it, and the -x check refuses it.
|
|
let r = run(stand(join(dir, "noexec-bin"), 0o644));
|
|
assert.equal(r.status, 1);
|
|
assert.match(r.stderr, /not an executable file/);
|
|
assert.equal(r.stdout, "");
|
|
|
|
// claude in a PATH dir with a space: systemd would split the pin there.
|
|
const spaced = stand(join(dir, "spaced bin"), 0o755);
|
|
r = run(spaced);
|
|
assert.equal(r.status, 1);
|
|
assert.match(r.stderr, /cannot carry/);
|
|
assert.ok(r.stderr.includes(join(spaced, "claude")), r.stderr);
|
|
assert.equal(r.stdout, "");
|
|
});
|
|
|
|
// Row 191: a restarted halted instance must keep its recovery bus reachable.
|
|
test('organization bus --pm while halted keeps the control host up', { timeout: 30000 }, async t => {
|
|
const { initializeControl, transition, readControl } = await import('../../processing-control/src/state.mjs');
|
|
const f = await orgFixture(t);
|
|
initializeControl(f.dataRoot);
|
|
transition(f.dataRoot, { business: 'acme', human: 'jason', halted: true });
|
|
const receipt = await busStart(f.ctx, 'acme', { detach: true, pm: true });
|
|
t.after(async () => { try { await busStop(f.ctx, 'acme'); } catch {} });
|
|
assert.match(readFileSync(join(f.dataRoot, 'bus-host/host.log'), 'utf8'), /processing halted: chief launch skipped/);
|
|
assert.equal((await busStatus(f.ctx, 'acme')).pid, receipt.pid);
|
|
assert.equal(readControl(f.dataRoot).halted, true);
|
|
assert.equal((await busStop(f.ctx, 'acme')).stopped, true);
|
|
});
|
|
|
|
test("organization notifier uses instance bindings and its own journal", async t => {
|
|
const f = await orgFixture(t), root = dirname(f.instanceRoot);
|
|
makeDeployment(root, { dmRecipient: IDS.owner });
|
|
const discord = await fakeDiscord(t);
|
|
const system = loadSystem({ env: f.env, instance: f.ctx.instance });
|
|
const boot = bootConfig({ system, businessId: "acme", orgRoot: f.dataRoot, env: f.env });
|
|
const host = await startHost({ boot, business: "acme", notifier: { binding: "test-seat", base: discord.base }, log: () => {} });
|
|
t.after(() => host.close(0));
|
|
assert.equal(host.dataRoot, f.dataRoot);
|
|
assert.ok(existsSync(journalPath(f.dataRoot, "acme")));
|
|
assert.equal(existsSync(journalPath(f.instanceRoot, "acme")), false);
|
|
await host.close(0);
|
|
});
|