cli: release host.lock on a late fork or send throw; retry a vanishing lock (row 106, #1600)
Darkwing round 1, approved by Dewey in #1600 comment 28319 on manifest
ae38cbe280aa5885aec7648d38143920753db86ea1ca3934d0a8885fb61e5105.
Sage gate rerun on f63c965f9 plus build.patch: every package suite and
every scripts/test-*.sh green (cli 168, harness 73, test-release 14).
readHostState now treats a host.json that vanishes mid-read as no host
for stopHost and hostStatus too (Dewey note b); Sage accepts it as
correct. Nothing goes live until a re-pin under Mos's go.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
@@ -181,8 +181,10 @@ mosaic bus status [--json] host state file, socket, writer.lock
|
||||
<business> (pid <pid>)`. A lock whose host no longer runs is taken over
|
||||
whatever its age: it is moved aside, checked to be the file that was
|
||||
read, and removed; if another host's claim was moved aside in that race,
|
||||
it goes back. Each refusal below removes `host.lock` again, so a host
|
||||
that didn't start leaves no file.
|
||||
it goes back. Of three starts racing one stale lock, two can both claim
|
||||
it, and then the broker's `writer.lock` refuses the second broker. Each
|
||||
refusal below removes `host.lock` again, so a host that didn't start
|
||||
leaves no file.
|
||||
4. It forks `packages/bus/src/process.mjs` and waits up to 30 s for the boot
|
||||
reply. If the broker answers `{ok: false, error}`, the host exits 3 and
|
||||
prints `broker refused to start: <code>`. A timeout or an early exit
|
||||
|
||||
+23
-12
@@ -44,12 +44,16 @@ export function readHostState(dataRoot) {
|
||||
return readState(hostFile(dataRoot));
|
||||
}
|
||||
|
||||
function readState(file) {
|
||||
// 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 {
|
||||
} catch (e) {
|
||||
if (e.code === "ENOENT") return null;
|
||||
throw new CliError(`bus host state is unreadable: ${file}`, 3);
|
||||
}
|
||||
if (!state || !Number.isSafeInteger(state.pid) || typeof state.startTime !== "string" || typeof state.business !== "string") {
|
||||
@@ -68,9 +72,11 @@ const same = (a, b) => a?.pid === b?.pid && a?.startTime === b?.startTime && a?.
|
||||
// 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. `between` runs after the stale lock is
|
||||
// read; it is for the tests. Exported for the tests.
|
||||
export function claimHostLock(dataRoot, state, { between = () => {} } = {}) {
|
||||
// lock refuses the second broker. A lock that goes before it is read is
|
||||
// tried again, like one that goes before the check. `found` runs once the
|
||||
// lock is found and before it is read, `between` after the stale lock is
|
||||
// read; both are for the tests. Exported for the tests.
|
||||
export function claimHostLock(dataRoot, state, { found = () => {}, between = () => {} } = {}) {
|
||||
mkdirSync(hostDir(dataRoot), { recursive: true, mode: 0o700 });
|
||||
const file = hostLock(dataRoot);
|
||||
const tag = `${process.pid}.${randomBytes(4).toString("hex")}`;
|
||||
@@ -85,7 +91,7 @@ export function claimHostLock(dataRoot, state, { between = () => {} } = {}) {
|
||||
} catch (e) {
|
||||
if (e.code !== "EEXIST") throw e;
|
||||
}
|
||||
const prior = readState(file);
|
||||
const prior = readState(file, found);
|
||||
if (!prior) continue;
|
||||
if (prior.live) throw new CliError(`a bus host already runs for ${prior.business} (pid ${prior.pid})`, 3);
|
||||
between();
|
||||
@@ -177,15 +183,20 @@ export async function startHost({ boot, business, notifier = null, bootTimeoutMs
|
||||
if (same(held, state)) rmSync(hostLock(dataRoot), { force: true });
|
||||
};
|
||||
|
||||
const broker = fork(BROKER, [], { stdio: ["ignore", "inherit", "inherit", "ipc"] });
|
||||
const reply = firstReply(broker, bootTimeoutMs, "broker");
|
||||
broker.send({ op: "boot", config: boot });
|
||||
// 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 {
|
||||
ready = await reply;
|
||||
broker = fork(BROKER, [], { stdio: ["ignore", "inherit", "inherit", "ipc"] });
|
||||
broker.send({ op: "boot", config: boot });
|
||||
ready = await firstReply(broker, bootTimeoutMs, "broker");
|
||||
} catch (e) {
|
||||
broker.kill("SIGTERM");
|
||||
await ended(broker, 5000);
|
||||
if (broker) {
|
||||
broker.kill("SIGTERM");
|
||||
await ended(broker, 5000);
|
||||
}
|
||||
release();
|
||||
throw e;
|
||||
}
|
||||
|
||||
@@ -4,7 +4,7 @@ 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, statSync, symlinkSync, utimesSync, writeFileSync, mkdirSync } from "node:fs";
|
||||
import { chmodSync, existsSync, readFileSync, readdirSync, rmSync, statSync, symlinkSync, utimesSync, writeFileSync, mkdirSync } from "node:fs";
|
||||
import { dirname, join } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { Client } from "../../bus/src/client.mjs";
|
||||
@@ -108,6 +108,33 @@ function failSends(t, fake) {
|
||||
});
|
||||
}
|
||||
|
||||
// Makes the next fork throw (`where` "fork") or the new child's 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) {
|
||||
const children = [];
|
||||
const onChild = ({ process: child }) => {
|
||||
children.push(child);
|
||||
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");
|
||||
});
|
||||
}
|
||||
|
||||
const epipe = (child) => {
|
||||
child.kill("SIGKILL");
|
||||
return Object.assign(new Error("write EPIPE"), { code: "EPIPE", errno: -32, syscall: "write" });
|
||||
@@ -259,6 +286,27 @@ test("a host takes over a host.lock its dead host left, and puts back a live cla
|
||||
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: ${hostLock(dataRoot)}`);
|
||||
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 });
|
||||
@@ -330,6 +378,22 @@ test("a host started before host.lock existed, holding only host.json, still ref
|
||||
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);
|
||||
|
||||
Reference in New Issue
Block a user