mosaicstack/stack
25
0
Fork 1
Code Issues 377 Pull Requests 13 Actions Packages Projects Releases 15 Wiki Activity
Files
67e4160742e9a82d2222c79f42a11bf559e7a1af
stack/agents/darkwing/work/queue-103/r1/build.patch
T

1259 lines
69 KiB
Diff
Raw Blame History

diff --git a/packages/cli/README.md b/packages/cli/README.md
index fb964dd44..8a3af3fde 100644
--- a/packages/cli/README.md
+++ b/packages/cli/README.md
@@ -172,17 +172,31 @@ mosaic bus status [--json] host state file, socket, writer.lock
and `readers: [<business>]`. It gets `config.trackers` (below) when a
tracker is set.
2. It reads the notifier config (below).
-3. It forks `packages/bus/src/process.mjs` and waits up to 30 s for the boot
+3. It takes `<dataRoot>/bus-host/host.lock` (0600), which holds the same
+ state as `host.json` (step 6). It writes the state to a file of its own
+ and hard-links it in as `host.lock`. The link fails if the file exists,
+ so of two hosts started at once, one gets it. If a live host holds the
+ lock, or a live host started before `host.lock` existed holds
+ `host.json`, the start refuses with exit 3: `a bus host already runs for
+ <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.
+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
gives exit 1.
-4. It forks `src/notifier-process.mjs` and hands it the reader capability
+5. It forks `src/notifier-process.mjs` and hands it the reader capability
over IPC in `{op: 'start'}`. A notifier refusal closes the broker, and
the host exits 3.
-5. It writes `<dataRoot>/bus-host/host.json` (0600): the pid, the process
+6. It writes `<dataRoot>/bus-host/host.json` (0600): the pid, the process
start time, the business, the start time as text and the notifier
- binding. The file holds no capability.
-6. It starts the session launcher (`src/launcher.mjs`, "Sessions" below)
+ binding. The file holds no capability. `bus stop`, `bus status` and the
+ human commands' default business read it. It appears only once the
+ broker and notifier are up, so a reader never finds a booting host with
+ no broker socket.
+7. It starts the session launcher (`src/launcher.mjs`, "Sessions" below)
and, with `--pm`, launches the business's `launch.by` instance. A PM
launch that refuses closes the host with the refusal's exit code.
@@ -202,11 +216,11 @@ one, and the host stops with exit 1 for the unit to restart. A late reply
never answers a later request, and a launch waiting on one refuses.
SIGTERM or SIGINT stops the notifier after its poll in flight, then closes
-the broker and removes `host.json`. If either child dies on its own, the
-host goes down with exit 1. The unit then restarts both. The host watches
-the children once both have started, and it checks each child's exit state
-at that point too, so a broker that dies while the notifier starts still
-takes the host down.
+the broker and removes `host.json`, then `host.lock`. If either child
+dies on its own, the host goes down with exit 1. The unit then restarts
+both. The host watches the children once both have started, and it
+checks each child's exit state at that point too, so a broker that dies
+while the notifier starts still takes the host down.
`bus stop` signals only a pid that still runs with the recorded start time
and whose command line is `…/packages/cli/src/cli.mjs bus start`.
@@ -360,7 +374,9 @@ A bundle file lives in `bus-host/` (0700) under a random name and is
deleted on every path. On close the launcher answers `host-closing` to a
connection still sending its request line, since nothing of it ran yet,
and drops every other connection. It waits for any push in flight to finish
-under its lock before the broker stops.
+under its lock before the broker stops. It waits at most 2 s
+(`CLOSE_WAIT_MS`) for its `host-closing` answers to leave, then drops those
+connections too, so a peer that never reads can't hold the close.
### Trackers
diff --git a/packages/cli/src/git-socket.mjs b/packages/cli/src/git-socket.mjs
index f3ea43117..6664bd024 100644
--- a/packages/cli/src/git-socket.mjs
+++ b/packages/cli/src/git-socket.mjs
@@ -36,7 +36,8 @@
// and go to the log as well.
//
// close() answers host-closing to a connection still waiting for its request
-// line, since nothing of it ran, and drops every other connection.
+// line, since nothing of it ran, and drops every other connection. It waits
+// at most CLOSE_WAIT_MS for those answers to leave, then drops the rest.
//
// A bundle file lives in bus-host/ (0700) under a random name, and the host
// deletes it on every path.
@@ -62,6 +63,8 @@ const BODY_MIN_RATE = 256;
export const bodyDeadline = (size) => BODY_IDLE_MS + Math.ceil(size / BODY_MIN_RATE);
// Connections served at once; more refuse git-busy.
const MAX_ACTIVE = 8;
+// The longest close() waits for its host-closing answers to leave.
+export const CLOSE_WAIT_MS = 2000;
const CAP = /^[0-9a-f]{64}$/;
const OID = /^[0-9a-f]{40}([0-9a-f]{24})?$/;
@@ -119,8 +122,8 @@ function readHeader(socket, waiting) {
}
// Exactly `size` bytes from the socket into a new file (0600), within
-// `deadlineMs` and with no gap over BODY_IDLE_MS.
-function readBody(socket, file, size, deadlineMs) {
+// `deadlineMs` and with no gap over `idleMs`.
+function readBody(socket, file, size, deadlineMs, idleMs) {
return new Promise((resolve, reject) => {
const fd = openSync(file, "wx", 0o600);
let got = 0;
@@ -139,7 +142,7 @@ function readBody(socket, file, size, deadlineMs) {
err ? reject(err) : resolve();
};
const onShort = () => done(new Refusal("invalid-request", `the bundle ended after ${got} of ${size} bytes`));
- const onIdle = () => done(new Refusal("body-timeout", `no bundle bytes for ${BODY_IDLE_MS} ms, after ${got} of ${size}`));
+ const onIdle = () => done(new Refusal("body-timeout", `no bundle bytes for ${idleMs} ms, after ${got} of ${size}`));
const deadline = setTimeout(() => done(new Refusal("body-timeout", `${got} of ${size} bundle bytes in ${deadlineMs} ms`)), deadlineMs);
const onData = (chunk) => {
if (got + chunk.length > size) return done(new Refusal("invalid-request", "more bytes than the request's size"));
@@ -154,7 +157,7 @@ function readBody(socket, file, size, deadlineMs) {
socket.on("data", onData);
socket.once("end", onShort);
socket.once("close", onShort);
- socket.setTimeout(BODY_IDLE_MS);
+ socket.setTimeout(idleMs);
socket.once("timeout", onIdle);
socket.resume();
});
@@ -164,8 +167,13 @@ function readBody(socket, file, size, deadlineMs) {
// own file throws as it is. The reviewer reads exactly the bundle's size and
// hangs up, which can come before the loop sees the file's end or before
// 'finish'. So a close is fetch-aborted while chunks are still to go, or when
-// writing one failed (EPIPE: the reader left with bytes still queued here).
-// Exported for the tests.
+// writing one failed, and served otherwise.
+//
+// That rests on one assumption: a reader that leaves while bytes are still
+// queued in this process fails the write that holds them (EPIPE, or
+// ECONNRESET), which sets socket.errored. Bytes the kernel already took count
+// as sent, even if the reviewer never reads them; its own size check catches
+// a short bundle on its side. Exported for the tests.
export async function send(socket, file) {
const aborted = () => new Refusal("fetch-aborted", "the reviewer left before the bundle was sent");
const served = () => !socket.errored;
@@ -197,9 +205,21 @@ export async function send(socket, file) {
if (!served()) throw aborted();
}
+// Waits until each socket has flushed what it was sent or closed, for at most
+// `ms`, then destroys every one still open: a peer that never reads can't hold
+// close() up. Exported for the tests.
+export async function letGo(sockets, ms = CLOSE_WAIT_MS) {
+ let timer = null;
+ const capped = new Promise((r) => (timer = setTimeout(r, ms)));
+ const flushed = sockets.map((s) => (s.destroyed || s.writableFinished ? null : new Promise((r) => (s.once("finish", r), s.once("close", r)))));
+ await Promise.race([capped, Promise.all(flushed)]);
+ clearTimeout(timer);
+ for (const s of sockets) s.destroy();
+}
+
// host: startHost()'s handle; business: loadBusiness() output. bundleMax,
-// bodyDeadline, maxActive and fsck are for the tests.
-export function createGitSocket({ host, business: b, dataRoot, env = process.env, log = () => {}, bundleMax = BUNDLE_MAX, bodyDeadline: deadlineOf = bodyDeadline, maxActive = MAX_ACTIVE, fsck = FSCK }) {
+// bodyDeadline, bodyIdle, maxActive and fsck are for the tests.
+export function createGitSocket({ host, business: b, dataRoot, env = process.env, log = () => {}, bundleMax = BUNDLE_MAX, bodyDeadline: deadlineOf = bodyDeadline, bodyIdle = BODY_IDLE_MS, maxActive = MAX_ACTIVE, fsck = FSCK }) {
const path = gitSocketPath(dataRoot);
const dir = join(dataRoot, "bus-host");
let server = null;
@@ -228,7 +248,7 @@ export function createGitSocket({ host, business: b, dataRoot, env = process.env
if (rest.length) throw new Refusal("invalid-request", "bundle bytes before send");
await op({ op: "checkPush", cap, task_ref });
socket.write(`${JSON.stringify({ ok: true, send: size })}\n`);
- await readBody(socket, file, size, deadlineOf(size));
+ await readBody(socket, file, size, deadlineOf(size), bodyIdle);
const result = await push({
dataRoot, business: b.id, project: req.project, root: p.root, taskRef: task_ref, branch, bundle: file, env, fsck,
record: (r) => op({ op: "recordPush", cap, push: { task_ref, project: req.project, branch, ...r } }),
@@ -319,8 +339,8 @@ export function createGitSocket({ host, business: b, dataRoot, env = process.env
for (const socket of sockets) if (!told.has(socket)) socket.destroy();
await Promise.allSettled([...active]);
// The refusal is a few bytes; once it is flushed, or the peer is gone,
- // nothing more is owed.
- await Promise.all([...told].map((s) => (s.destroyed || s.writableFinished ? null : new Promise((r) => (s.once("finish", r), s.once("close", r))))));
+ // nothing more is owed, and past CLOSE_WAIT_MS it is dropped unsent.
+ await letGo([...told]);
for (const socket of sockets) socket.destroy();
await stopped;
rmSync(path, { force: true });
diff --git a/packages/cli/src/host.mjs b/packages/cli/src/host.mjs
index 0073091b2..23150d4c1 100644
--- a/packages/cli/src/host.mjs
+++ b/packages/cli/src/host.mjs
@@ -3,7 +3,9 @@
// 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.
+// 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
@@ -13,8 +15,9 @@
// ops still work.
import { fork } from "node:child_process";
+import { randomBytes } from "node:crypto";
import { once } from "node:events";
-import { existsSync, mkdirSync, readFileSync, renameSync, rmSync, writeFileSync } from "node:fs";
+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";
@@ -26,16 +29,22 @@ 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) {
- const file = hostFile(dataRoot);
+ return readState(hostFile(dataRoot));
+}
+
+function readState(file) {
if (!existsSync(file)) return null;
let state;
try {
@@ -49,6 +58,58 @@ export function readHostState(dataRoot) {
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. `between` runs after the stale lock is
+// read; it is for the tests. Exported for the tests.
+export function claimHostLock(dataRoot, state, { between = () => {} } = {}) {
+ 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);
+ if (!prior) continue;
+ if (prior.live) throw new CliError(`a bus host already runs for ${prior.business} (pid ${prior.pid})`, 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(`${file} 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);
@@ -99,8 +160,22 @@ export function watchChildren(children, onDeath) {
// children are up.
export async function startHost({ boot, business, notifier = null, 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 };
+ claimHostLock(dataRoot, state);
+ // 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 });
+ };
const broker = fork(BROKER, [], { stdio: ["ignore", "inherit", "inherit", "ipc"] });
const reply = firstReply(broker, bootTimeoutMs, "broker");
@@ -111,10 +186,12 @@ export async function startHost({ boot, business, notifier = null, bootTimeoutMs
} catch (e) {
broker.kill("SIGTERM");
await ended(broker, 5000);
+ release();
throw e;
}
if (ready?.ok !== true) {
await ended(broker, 5000);
+ 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);
@@ -132,17 +209,18 @@ export async function startHost({ boot, business, notifier = null, bootTimeoutMs
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);
}
}
- const state = { hostVersion: 1, pid: process.pid, startTime: startTimeOf(process.pid), business, startedAt: new Date().toISOString(), notifier: notifier ? notifier.binding : null };
rmSync(hostFile(dataRoot), { force: true });
writeHostState(dataRoot, state);
@@ -234,6 +312,7 @@ export async function startHost({ boot, business, notifier = null, bootTimeoutMs
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;
}
diff --git a/packages/cli/tests/git-socket.test.mjs b/packages/cli/tests/git-socket.test.mjs
index 998a0d795..83978e097 100644
--- a/packages/cli/tests/git-socket.test.mjs
+++ b/packages/cli/tests/git-socket.test.mjs
@@ -11,6 +11,7 @@ import { spawnSync } from "node:child_process";
import { createHash, randomBytes } from "node:crypto";
import { connect } from "node:net";
import { chmodSync, existsSync, lstatSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, readlinkSync, renameSync, rmSync, statSync, symlinkSync, writeFileSync } from "node:fs";
+import { open } from "node:fs/promises";
import { tmpdir } from "node:os";
import { Duplex } from "node:stream";
import { join } from "node:path";
@@ -21,7 +22,7 @@ import { gitSocketPath } from "../../seat/src/session.mjs";
import { workingRepo } from "../../seat/src/working.mjs";
import { fetchReview, pushWorking } from "../../harness/src/git-tools.mjs";
import { bootConfig, loadSystem, rolesDir } from "../src/config.mjs";
-import { createGitSocket, send } from "../src/git-socket.mjs";
+import { CLOSE_WAIT_MS, createGitSocket, letGo, send } from "../src/git-socket.mjs";
import { startHost, startTimeOf } from "../src/host.mjs";
import { fixture } from "./helpers.mjs";
@@ -526,6 +527,83 @@ test("send: a close after every byte left is served; one that fails a queued wri
await rejects(gone, "fetch-aborted");
});
+test("a bundle with no bytes for bodyIdle refuses body-timeout long before its deadline", async (t) => {
+ // The deadline is a minute away, so only the idle limit can end it.
+ const w = await setup(t, { bodyDeadline: () => 60000, bodyIdle: 300 });
+ const { coder } = w.roles;
+ const started = Date.now();
+ // Sends one byte of 1000 after "send", then nothing.
+ const stalled = new Promise((resolve, reject) => {
+ const s = connect(w.sock.path);
+ let buf = "";
+ s.on("connect", () => s.write(pushReq(coder.cap, 1000)));
+ s.on("data", (b) => {
+ const sent = buf === `${JSON.stringify({ ok: true, send: 1000 })}\n`;
+ buf += b;
+ if (!sent && buf === `${JSON.stringify({ ok: true, send: 1000 })}\n`) s.write("x");
+ });
+ s.on("close", () => resolve({ last: buf.trim().split("\n").pop(), ms: Date.now() - started }));
+ s.on("error", () => {});
+ setTimeout(() => (s.destroy(), reject(new Error("no refusal in 5 s"))), 5000).unref();
+ });
+ const { last, ms } = await stalled;
+ assert.deepEqual(JSON.parse(last), { ok: false, error: "body-timeout" });
+ assert.ok(ms >= 300 && ms < 5000, `${ms} ms`);
+ assert.ok((await refusedOn(w)).includes("coder body-timeout"));
+ assert.equal(w.refs(), "");
+});
+
+// A FIFO as the bundle holds send()'s loop open on a read, so a test can end
+// the connection between the last chunk and the loop's end: the arm after the
+// loop. A plain close there was served; a failed write was fetch-aborted.
+test("send: a close once the loop is past its last chunk is served, and a failed write there is fetch-aborted", async (t) => {
+ const dir = mkdtempSync(join(tmpdir(), "mosaic-send-"));
+ t.after(() => rmSync(dir, { recursive: true, force: true }));
+ const fifo = join(dir, "b.fifo");
+ assert.equal(spawnSync("mkfifo", [fifo]).status, 0);
+
+ // `fail` is how the stand-in socket ends: destroyed as it is, or by
+ // failing its held write as a socket whose reader left does.
+ const run = async (fail) => {
+ let pending = null;
+ const sock = new Duplex({ read() {}, write: (c, e, cb) => (pending = cb), final: (cb) => cb() });
+ sock.on("error", () => {});
+ const sent = send(sock, fifo);
+ // Settled before the FIFO ends, so neither outcome is unhandled.
+ const outcome = sent.then(() => "served", (e) => e.code);
+ const writer = await open(fifo, "w");
+ await writer.write("ten bytes!");
+ await until(() => pending, "the write");
+ if (fail) pending(Object.assign(new Error("write EPIPE"), { code: "EPIPE" }));
+ else sock.destroy();
+ await until(() => sock.destroyed, "the socket's end");
+ await writer.close();
+ return outcome;
+ };
+ assert.equal(await run(false), "served");
+ assert.equal(await run(true), "fetch-aborted");
+});
+
+// A peer that never reads leaves the host's last write unfinished; this
+// stand-in never finishes one.
+test("letGo waits for an answer to leave for at most its cap, then destroys the socket", async (t) => {
+ const stuck = new Duplex({ read() {}, write() {}, final() {} });
+ stuck.on("error", () => {});
+ stuck.end("host-closing\n");
+ const quick = new Duplex({ read() {}, write: (c, e, cb) => cb(), final: (cb) => cb() });
+ quick.on("error", () => {});
+ quick.end("host-closing\n");
+ const started = Date.now();
+ let timer = null;
+ const hung = new Promise((r) => (timer = setTimeout(() => r("hung"), 3000)));
+ t.after(() => clearTimeout(timer));
+ assert.equal(await Promise.race([letGo([stuck, quick], 300).then(() => "let go"), hung]), "let go");
+ const ms = Date.now() - started;
+ assert.ok(ms >= 290 && ms < 3000, `${ms} ms`);
+ assert.ok(stuck.destroyed && quick.destroyed);
+ assert.ok(CLOSE_WAIT_MS > 0 && CLOSE_WAIT_MS <= 10000, `CLOSE_WAIT_MS is ${CLOSE_WAIT_MS}`);
+});
+
test("a record that fails and can't be undone refuses rollback-failed, which the host logs", async (t) => {
let lock = null;
const w = await setup(t, {
diff --git a/packages/cli/tests/host.test.mjs b/packages/cli/tests/host.test.mjs
index b6644cf3c..0202646a0 100644
--- a/packages/cli/tests/host.test.mjs
+++ b/packages/cli/tests/host.test.mjs
@@ -4,12 +4,12 @@ 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, writeFileSync, mkdirSync } from "node:fs";
+import { chmodSync, existsSync, readFileSync, readdirSync, 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";
import { bootConfig, loadSystem, REPO } from "../src/config.mjs";
-import { hostDir, hostFile, hostStatus, startHost, startTimeOf, stopHost, watchChildren } from "../src/host.mjs";
+import { claimHostLock, hostDir, hostFile, hostLock, hostStatus, readHostState, startHost, startTimeOf, stopHost, watchChildren } from "../src/host.mjs";
import { journalPath } from "../src/notifier.mjs";
import { IDS, makeDeployment } from "../../discord/tests/helpers.mjs";
import { fixture, notifyConfig, OPTIONS, tmp } from "./helpers.mjs";
@@ -186,6 +186,7 @@ test("a notifier that dies takes the host down with exit 1, so the unit restarts
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) => {
@@ -203,6 +204,145 @@ test("a second host for the same data root refuses with exit 3 while the first r
assert.equal(await host.close(0), 0);
});
+test("two bus starts racing for one data root leave one host running; the other refuses that a host already runs", async (t) => {
+ const f = fixture(tmp(t));
+ notifyConfig(f.dataRoot, "acme", null);
+ const starts = [0, 1].map(() => {
+ const child = spawn(process.execPath, [CLI, "bus", "start", "acme"], { env: f.env, stdio: ["ignore", "pipe", "pipe"] });
+ t.after(() => child.exitCode === null && child.signalCode === null && child.kill("SIGKILL"));
+ const run = { child, out: "", exit: once(child, "exit") };
+ child.stdout.on("data", (b) => (run.out += b));
+ child.stderr.on("data", (b) => (run.out += b));
+ return run;
+ });
+ // The loser exits; the winner keeps running.
+ const [first] = await Promise.race(starts.map((r, i) => r.exit.then((e) => [i, e])));
+ const loser = starts[first];
+ const winner = starts[1 - first];
+ assert.equal(loser.child.exitCode, 3, loser.out);
+ assert.match(loser.out, new RegExp(`a bus host already runs for acme \\(pid ${winner.child.pid}\\)`));
+ await until(() => /bus host up: business acme/.test(winner.out), 30000);
+ assert.equal(winner.child.exitCode, null);
+ const state = readHostState(f.dataRoot);
+ assert.equal(state.pid, winner.child.pid);
+ assert.equal(state.live, true);
+ assert.equal(JSON.parse(readFileSync(hostLock(f.dataRoot), "utf8")).pid, winner.child.pid);
+ const claims = () => readdirSync(hostDir(f.dataRoot)).filter((n) => n.startsWith("host.")).sort();
+ assert.deepEqual(claims(), ["host.json", "host.lock"], "no claim file left over");
+ const stop = spawnSync(process.execPath, [CLI, "bus", "stop"], { env: f.env, encoding: "utf8", timeout: 70000 });
+ assert.equal(stop.status, 0, stop.stderr);
+ assert.equal((await winner.exit)[0], 0, winner.out);
+ assert.deepEqual(claims(), [], "the stop gives back host.lock and host.json");
+});
+
+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"]);
+});
+
+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 host.lock left by a host killed with SIGKILL doesn't stop the next bus start", async (t) => {
+ const f = fixture(tmp(t));
+ notifyConfig(f.dataRoot, "acme", null);
+ const start = () => {
+ const child = spawn(process.execPath, [CLI, "bus", "start", "acme"], { env: f.env, stdio: ["ignore", "pipe", "pipe"] });
+ t.after(() => child.exitCode === null && child.signalCode === null && child.kill("SIGKILL"));
+ const run = { child, out: "", exit: once(child, "exit") };
+ child.stdout.on("data", (b) => (run.out += b));
+ child.stderr.on("data", (b) => (run.out += b));
+ return run;
+ };
+ const first = start();
+ await until(() => /bus host up: business acme/.test(first.out), 30000);
+ first.child.kill("SIGKILL");
+ await first.exit;
+ assert.equal(JSON.parse(readFileSync(hostLock(f.dataRoot), "utf8")).pid, first.child.pid, "the killed host never gave back its lock");
+ // The orphaned broker sees its channel close and lets go of the writer lock.
+ await until(() => !existsSync(join(f.dataRoot, "bus", "writer.lock")), 30000);
+ const second = start();
+ await until(() => /bus host up: business acme/.test(second.out) || second.child.exitCode !== null, 30000);
+ assert.equal(second.child.exitCode, null, second.out);
+ assert.equal(readHostState(f.dataRoot).pid, second.child.pid);
+ assert.equal(JSON.parse(readFileSync(hostLock(f.dataRoot), "utf8")).pid, second.child.pid);
+ const stop = spawnSync(process.execPath, [CLI, "bus", "stop"], { env: f.env, encoding: "utf8", timeout: 70000 });
+ assert.equal(stop.status, 0, stop.stderr);
+ assert.equal((await second.exit)[0], 0, second.out);
+});
+
+test("while a host boots, a reader sees no host: the human commands still ask for --business, stop and status find none", (t) => {
+ const f = fixture(tmp(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);
+ assert.equal(hostStatus(f.dataRoot).host, null);
+ const inbox = spawnSync(process.execPath, [CLI, "inbox"], { env: f.env, encoding: "utf8", timeout: 20000 });
+ assert.notEqual(inbox.status, 0);
+ assert.match(inbox.stderr, /no bus host runs here, so name the business with --business/);
+ const stop = spawnSync(process.execPath, [CLI, "bus", "stop"], { env: f.env, encoding: "utf8", timeout: 20000 });
+ assert.equal(stop.status, 0, stop.stderr);
+ assert.match(stop.stdout, /no host runs/);
+});
+
+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"]);
+});
+
+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);
+});
+
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);
@@ -296,6 +436,7 @@ test("a notifier that refuses stops the broker and the host refuses with exit 3"
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) => {
@@ -348,6 +489,8 @@ test("a notifier that dies before it replies, after the broker died, still refus
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) => {
@@ -435,6 +578,8 @@ test("bus start refuses with exit 3 and the code when the broker refuses to boot
const r = spawnSync(process.execPath, [CLI, "bus", "start", "acme"], { env: f.env, encoding: "utf8", timeout: 40000 });
assert.equal(r.status, 3, r.stderr);
assert.match(r.stderr, /broker refused to start: startup-refused/);
+ assert.equal(existsSync(hostFile(f.dataRoot)), false);
+ assert.equal(existsSync(hostLock(f.dataRoot)), false);
const s = spawnSync(process.execPath, [CLI, "bus", "status"], { env: f.env, encoding: "utf8" });
assert.equal(s.status, 0, s.stderr);
assert.match(s.stdout, /host: none/);
diff --git a/packages/harness/src/git-tools.mjs b/packages/harness/src/git-tools.mjs
index 50fd5336e..4409dcfc9 100644
--- a/packages/harness/src/git-tools.mjs
+++ b/packages/harness/src/git-tools.mjs
@@ -28,6 +28,9 @@ const PROJECT = /^[a-z0-9][a-z0-9._-]{0,63}$/;
const WORK_REF = /^work\/vikunja-[1-9][0-9]*-[1-9][0-9]*$/;
const OID = /^[0-9a-f]{40}([0-9a-f]{24})?$/;
const TIMEOUT_MS = 300_000;
+// How long a git has after SIGTERM before SIGKILL, as on the host.
+const GRACE_MS = 5000;
+const LIMITS = { timeoutMs: TIMEOUT_MS, graceMs: GRACE_MS };
const SAFE = [
"-c", "core.hooksPath=/dev/null",
"-c", "core.fsmonitor=false",
@@ -42,20 +45,44 @@ const failure = (code, message = code) => Object.assign(new Error(message), { co
export const validBranch = (branch) =>
typeof branch === "string" && /^[A-Za-z0-9][A-Za-z0-9._\/-]{0,199}$/.test(branch) && !/\.\.|\/\/|\/\.|@\{|\.lock(\/|$)|[./]$/.test(branch);
-function git(args, env = process.env) {
+// git with SAFE first, in its own process group (detached). Past
+// limits.timeoutMs the group gets SIGTERM, on which git removes its
+// lockfiles, then SIGKILL limits.graceMs later; once a signalled git has
+// closed, the group gets SIGKILL again. The same sequence as the host's
+// runGit in packages/seat/src/checkout.mjs, kept here so the runner doesn't
+// load the host's module. Exported for the tests, which shorten `limits`.
+export function git(args, env = process.env, limits = LIMITS) {
return new Promise((resolve) => {
const child = spawn("git", [...SAFE, ...args], {
env: { PATH: env.PATH ?? "/usr/bin:/bin", HOME: "/nonexistent", GIT_CONFIG_NOSYSTEM: "1", GIT_CONFIG_GLOBAL: "/dev/null", GIT_TERMINAL_PROMPT: "0", LC_ALL: "C" },
- timeout: TIMEOUT_MS,
- killSignal: "SIGKILL",
+ detached: true,
stdio: ["ignore", "pipe", "pipe"],
});
let out = "";
let err = "";
+ let signalled = false;
+ let kill = null;
+ // A git that never started has no pid; an empty group is ESRCH.
+ const group = (signal) => {
+ try {
+ process.kill(-child.pid, signal);
+ } catch {}
+ };
+ const term = setTimeout(() => {
+ signalled = true;
+ group("SIGTERM");
+ kill = setTimeout(() => group("SIGKILL"), limits.graceMs);
+ }, limits.timeoutMs);
+ const settle = (r) => {
+ clearTimeout(term);
+ clearTimeout(kill);
+ if (signalled) group("SIGKILL");
+ resolve(r);
+ };
child.stdout.setEncoding("utf8").on("data", (d) => (out += d));
child.stderr.setEncoding("utf8").on("data", (d) => (err += d));
- child.on("error", (e) => resolve({ ok: false, out, err, why: e.code ?? e.message }));
- child.on("close", (code, signal) => resolve({ ok: code === 0, out, err, why: signal ? `killed by ${signal}` : err.trim().split("\n").pop() || `exit ${code}` }));
+ child.on("error", (e) => settle({ ok: false, out, err, why: e.code ?? e.message }));
+ child.on("close", (code, signal) => settle({ ok: code === 0, out, err, why: signal ? `killed by ${signal}` : err.trim().split("\n").pop() || `exit ${code}` }));
});
}
@@ -169,8 +196,11 @@ export async function pushWorking({ path, cap, workspace, task_ref, project, bra
// Resolves { project, ref, commit, clone }; ref is the clone's remote ref.
export async function fetchReview({ path, cap, workspace, task_ref, request, tmp = "/tmp", env = process.env }) {
// Commits the host may leave out: what the clones' remotes already hold.
- // The working refs come first, newest first, since a review is most often
- // of a commit built on one; origin's fill what is left of HAVE_MAX.
+ // The working refs come first, since a review is most often of a commit
+ // built on one; origin's fill what is left of HAVE_MAX. The order is per
+ // clone, not across clones: each clone's refs newest first, one clone
+ // after another in the workspace's directory order. So past HAVE_MAX, a
+ // later clone's newer refs can lose to an earlier clone's older ones.
const have = new Set();
const projects = projectsIn(workspace);
for (const remote of ["refs/remotes/working", "refs/remotes/origin"])
diff --git a/packages/harness/tests/git-tools.test.mjs b/packages/harness/tests/git-tools.test.mjs
index b66fc2874..2880b76f0 100644
--- a/packages/harness/tests/git-tools.test.mjs
+++ b/packages/harness/tests/git-tools.test.mjs
@@ -5,10 +5,10 @@
import { test } from "node:test";
import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
-import { mkdirSync } from "node:fs";
+import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
import { createServer } from "node:net";
import { join } from "node:path";
-import { fetchReview, HAVE_MAX } from "../src/git-tools.mjs";
+import { fetchReview, git as runnerGit, HAVE_MAX } from "../src/git-tools.mjs";
import { scratch } from "./helpers.mjs";
const ENV = { PATH: process.env.PATH, HOME: "/nonexistent", GIT_CONFIG_NOSYSTEM: "1", GIT_CONFIG_GLOBAL: "/dev/null", LC_ALL: "C" };
@@ -78,3 +78,88 @@ test("fetch_review's have lists the working refs before origin's, and no more th
const origin = new Set(Object.entries(a).filter(([n]) => n.startsWith("refs/remotes/origin/")).map(([, oid]) => oid));
for (const oid of req.have.slice(13)) assert.ok(origin.has(oid), oid);
});
+
+// Past HAVE_MAX the order is per clone: each clone's working refs newest
+// first, one clone after another, so an earlier clone's older refs come
+// ahead of a later clone's newer ones. The clones' order is the workspace's
+// directory order, so the test takes it from the request.
+test("fetch_review's have orders the working refs newest first within each clone, one clone after another", async (t) => {
+ const dir = scratch(t);
+ const workspace = join(dir, "ws");
+ // 20 working refs in each clone; beta's are all newer than alpha's.
+ // Created in a shuffled order, so the name order isn't the date order.
+ const order = [7, 2, 15, 0, 11, 19, 4, 9, 13, 1, 17, 6, 3, 18, 10, 5, 14, 8, 16, 12];
+ const refs = (task, base) => Object.fromEntries(order.map((i) => [`refs/remotes/working/work/vikunja-${task}-${i + 1}`, base + i * 60]));
+ const alpha = clone(join(workspace, "alpha"), refs(1, 1_700_000_000));
+ const beta = clone(join(workspace, "beta"), refs(2, 1_750_000_000));
+ const host = await fakeHost(t, dir);
+ await assert.rejects(fetchReview({ path: host.path, cap: "c".repeat(64), workspace, task_ref: "vikunja:1/2", request: "r1", tmp: dir, env: ENV }), (e) => e.code === "request-not-found");
+ const [req] = host.lines;
+ // Newest first: the ref for i = 19 first, i = 0 last.
+ const newest = (oids, task) => order.slice().sort((x, y) => y - x).map((i) => oids[`refs/remotes/working/work/vikunja-${task}-${i + 1}`]);
+ const [first, second] = new Set(Object.values(alpha)).has(req.have[0]) ? [newest(alpha, 1), newest(beta, 2)] : [newest(beta, 2), newest(alpha, 1)];
+ assert.deepEqual(req.have, [...first, ...second].slice(0, HAVE_MAX));
+});
+
+// A git on PATH that runs `body` in place of the real one.
+function fakeGit(dir, body) {
+ const bin = join(dir, "bin");
+ mkdirSync(bin);
+ writeFileSync(join(bin, "git"), `#!/bin/bash\n${body}\n`, { mode: 0o755 });
+ return { ...ENV, PATH: `${bin}:${process.env.PATH}` };
+}
+
+// True once `pid` has gone, or is a zombie waiting to be reaped.
+async function gone(pid, ms) {
+ const end = Date.now() + ms;
+ for (;;) {
+ let stat = null;
+ try {
+ stat = readFileSync(`/proc/${pid}/stat`, "utf8");
+ } catch {}
+ if (stat === null || stat.slice(stat.lastIndexOf(")") + 2)[0] === "Z") return true;
+ if (Date.now() > end) return false;
+ await new Promise((r) => setTimeout(r, 50));
+ }
+}
+
+async function pidIn(file) {
+ for (let i = 0; i < 100 && !existsSync(file); i++) await new Promise((r) => setTimeout(r, 20));
+ return Number(readFileSync(file, "utf8").trim());
+}
+
+const kill = (pid) => {
+ try {
+ process.kill(pid, "SIGKILL");
+ } catch {}
+};
+
+// git ignores SIGTERM and waits on a sleep it started. SIGTERM goes to the
+// group, so the sleep dies at the timeout; git itself lasts until SIGKILL.
+test("a runner git past its time: SIGTERM reaches its process group, and SIGKILL follows after the grace period", { timeout: 20000 }, async (t) => {
+ const dir = scratch(t);
+ const file = join(dir, "child.pid");
+ const env = fakeGit(dir, `sleep 30 </dev/null >/dev/null 2>&1 &\necho $! > '${file}'\ntrap '' TERM\nwait $!\nexec sleep 30`);
+ const started = Date.now();
+ const run = runnerGit(["fetch"], env, { timeoutMs: 300, graceMs: 1500 });
+ const pid = await pidIn(file);
+ t.after(() => kill(pid));
+ assert.equal(await gone(pid, 800), true, "the sleep git started died at SIGTERM");
+ assert.ok(Date.now() - started < 1500, "before the grace period was over");
+ const r = await run;
+ assert.equal(r.why, "killed by SIGKILL");
+ assert.ok(Date.now() - started >= 1800, "SIGKILL waited out the grace period");
+});
+
+// git dies at SIGTERM; a process it started ignores SIGTERM and holds none
+// of git's output. Once git has closed, its group gets SIGKILL.
+test("once a signalled runner git has closed, its group gets SIGKILL, so a process that ignored SIGTERM dies too", { timeout: 20000 }, async (t) => {
+ const dir = scratch(t);
+ const file = join(dir, "child.pid");
+ const env = fakeGit(dir, `(trap '' TERM; exec sleep 30) </dev/null >/dev/null 2>&1 &\necho $! > '${file}'\nexec sleep 30`);
+ const r = await runnerGit(["fetch"], env, { timeoutMs: 300, graceMs: 20000 });
+ assert.equal(r.why, "killed by SIGTERM");
+ const pid = await pidIn(file);
+ t.after(() => kill(pid));
+ assert.equal(await gone(pid, 1500), true, "the sleep that ignored SIGTERM is gone well inside the grace period");
+});
diff --git a/packages/seat/README.md b/packages/seat/README.md
index 8b830a956..2d01c9d0f 100644
--- a/packages/seat/README.md
+++ b/packages/seat/README.md
@@ -279,7 +279,8 @@ so the host never recurses into one: a clone with a submodule is kept, and
`status` runs with `--ignore-submodules=all` and every call with
`submodule.recurse=false`. The root is read with `GIT_CEILING_DIRECTORIES`
at its parent, so a root that isn't a repository is never taken for the
-one above it. Each git call is killed after 100 s.
+one above it. Each git call gets SIGTERM after 100 s and SIGKILL 5 s
+later, as below.
The checks hold only while nothing else writes the clone. Each workspace
is one instance's, and the launcher refuses `instance-running` before it
@@ -332,15 +333,23 @@ The temp ref is deleted on every path, and so is the bundle file in
`bus-host/`. No other ref moves, and the root is only read.
The repository's lock is a promise chain inside the host's process, so it
-assumes one launcher per dataRoot. `startHost` refuses a second bus host
-while `bus-host/host.json` names a live one, but that check isn't atomic:
-two launchers started at the same moment could both pass it, and their
-pushes to one repository would no longer be serialized.
+assumes one launcher per dataRoot. `startHost` takes `bus-host/host.lock`
+with a hard link, which fails if the file exists, and refuses while a live
+host holds it, so two launchers started at the same moment can't both run
+(`packages/cli/README.md`, `bus start`).
A host git call that runs past 100 s (`GIT_TIMEOUT_MS`) gets SIGTERM. git
removes its lockfiles on SIGTERM. Only a git still running 5 s later
(`GIT_GRACE_MS`) gets SIGKILL, which can leave a ref's `.lock` behind; that
-lock refuses every later push to the ref until someone removes it.
+lock refuses every later push to the ref until someone removes it. The
+checkouts (`checkout.mjs`) and the working repositories (`working.mjs`) run
+git the same way, through `runGit`. Each git is the leader of its own
+process group, and both signals go to the group, so a hook or helper git
+started can't outlive it. Once a signalled git has ended, the group gets
+SIGKILL once more, for a process that ignored SIGTERM. A git that ends on
+its own isn't signalled. Since git has its own group, a Ctrl-C at the
+terminal of a host started by hand doesn't reach it; a systemd stop still
+does, since it kills the unit's cgroup.
For a review, under the same lock, the host fetches `main` again and
writes a bundle of the requested commit with one ref,
diff --git a/packages/seat/src/checkout.mjs b/packages/seat/src/checkout.mjs
index 35172b7f6..f91fe1590 100644
--- a/packages/seat/src/checkout.mjs
+++ b/packages/seat/src/checkout.mjs
@@ -45,6 +45,9 @@ import { lstatSync, readdirSync, renameSync, rmSync } from "node:fs";
import { dirname, join } from "node:path";
export const GIT_TIMEOUT_MS = 100_000;
+// How long a git has after SIGTERM before SIGKILL.
+export const GIT_GRACE_MS = 5000;
+export const GIT_LIMITS = Object.freeze({ timeoutMs: GIT_TIMEOUT_MS, graceMs: GIT_GRACE_MS });
// A clone's config is a few hundred bytes, a little more per branch.
export const CONFIG_MAX_BYTES = 64 * 1024;
@@ -78,20 +81,54 @@ export function gitEnv(env = process.env) {
};
}
+// runGit(argv, env, limits) runs git with exactly `argv` and `env`, and
+// resolves { ok, out, err, why }; working.mjs runs its git through it too.
// Async, so a long clone doesn't hold up the rest of the bus host.
-function git(args, env, extra = {}) {
+//
+// git runs as the leader of its own process group (detached), so a timeout
+// reaches what git started as well: SIGTERM to the group at
+// limits.timeoutMs, on which git removes its lockfiles, then SIGKILL to the
+// group limits.graceMs later. Once a git that was signalled has closed, the
+// group gets SIGKILL again, for a process that ignored SIGTERM and let go of
+// git's output. A git that ends on its own isn't signalled at all. One
+// consequence: a Ctrl-C at the terminal of a host run by hand no longer
+// reaches its git, though a systemd stop still does (it kills the cgroup).
+export function runGit(argv, env, limits = GIT_LIMITS) {
return new Promise((resolve) => {
- const child = spawn("git", [...SAFE, ...args], { env: { ...gitEnv(env), ...extra }, timeout: GIT_TIMEOUT_MS, killSignal: "SIGKILL", stdio: ["ignore", "pipe", "pipe"] });
+ const child = spawn("git", argv, { env, detached: true, stdio: ["ignore", "pipe", "pipe"] });
let out = "";
let err = "";
+ let signalled = false;
+ let kill = null;
+ // A git that never started has no pid; an empty group is ESRCH. Neither
+ // leaves anything to signal.
+ const group = (signal) => {
+ try {
+ process.kill(-child.pid, signal);
+ } catch {}
+ };
+ const term = setTimeout(() => {
+ signalled = true;
+ group("SIGTERM");
+ kill = setTimeout(() => group("SIGKILL"), limits.graceMs);
+ }, limits.timeoutMs);
+ const settle = (r) => {
+ clearTimeout(term);
+ clearTimeout(kill);
+ if (signalled) group("SIGKILL");
+ resolve(r);
+ };
child.stdout.setEncoding("utf8").on("data", (d) => (out += d));
child.stderr.setEncoding("utf8").on("data", (d) => (err += d));
// A git that can't start may not emit close, so error settles it too.
- child.on("error", (e) => resolve({ ok: false, out, why: e.code ?? e.message }));
- child.on("close", (code, signal) => resolve({ ok: code === 0, out, why: signal ? `killed by ${signal}` : err.trim().split("\n").pop() || `exit ${code}` }));
+ child.on("error", (e) => settle({ ok: false, out, err, why: e.code ?? e.message }));
+ child.on("close", (code, signal) => settle({ ok: code === 0, out, err, why: signal ? `killed by ${signal}` : err.trim().split("\n").pop() || `exit ${code}` }));
});
}
+// git in a clone or a root, with SAFE first. `limits` is for the tests.
+const git = (args, env, extra = {}, limits = GIT_LIMITS) => runGit([...SAFE, ...args], { ...gitEnv(env), ...extra }, limits);
+
const inClone = (dest) => [`--git-dir=${join(dest, ".git")}`, `--work-tree=${dest}`];
// The keys `git clone` writes, and what each may hold. branch.<name>.remote
@@ -113,7 +150,7 @@ function allowed(key, value, root) {
}
// Why the clone at `dest` isn't safe for the host to run git in, or null.
-async function unsafe(dest, root, env) {
+async function unsafe(dest, root, g) {
const isDir = (p) => {
try {
return lstatSync(p).isDirectory();
@@ -159,7 +196,7 @@ async function unsafe(dest, root, env) {
} catch {
return "config-changed";
}
- const list = await git(["config", "--file", join(dest, ".git", "config"), "--list", "--null"], env);
+ const list = await g(["config", "--file", join(dest, ".git", "config"), "--list", "--null"]);
if (!list.ok) return "config-changed";
const seen = new Set();
for (const entry of list.out.split("\0").filter(Boolean)) {
@@ -174,23 +211,23 @@ async function unsafe(dest, root, env) {
// The branch the root has checked out, null when it has none, or an error.
// The ceiling stops git finding a repository above a root that isn't one.
-async function rootBranch(root, env) {
+async function rootBranch(root, g) {
const ceiling = { GIT_CEILING_DIRECTORIES: dirname(root) };
- const r = await git(["-C", root, "symbolic-ref", "-q", "--short", "HEAD"], env, ceiling);
+ const r = await g(["-C", root, "symbolic-ref", "-q", "--short", "HEAD"], ceiling);
if (r.ok) return { branch: r.out.trim() };
// symbolic-ref -q exits 1 with no output on a detached HEAD.
- const head = await git(["-C", root, "rev-parse", "--verify", "-q", "HEAD"], env, ceiling);
+ const head = await g(["-C", root, "rev-parse", "--verify", "-q", "HEAD"], ceiling);
return head.ok ? { branch: null } : { error: r.why };
}
-async function clone(project, root, workspace, dest, env) {
- const at = await rootBranch(root, env);
+async function clone(project, root, workspace, dest, g) {
+ const at = await rootBranch(root, g);
if (at.error) throw new CheckoutError(project, `the root ${root} can't be read: ${at.error}`);
if (!at.branch) throw new CheckoutError(project, `the root ${root} has no branch checked out`);
// A project id never starts with a dot, so the temp name is no project's.
const tmp = join(workspace, `.${project}.clone-${randomBytes(4).toString("hex")}`);
try {
- const r = await git(["clone", "-q", "--no-hardlinks", "--template=", "--branch", at.branch, "--", root, tmp], env);
+ const r = await g(["clone", "-q", "--no-hardlinks", "--template=", "--branch", at.branch, "--", root, tmp]);
if (!r.ok) throw new CheckoutError(project, `git clone of ${root} failed: ${r.why}`);
renameSync(tmp, dest);
} catch (e) {
@@ -201,33 +238,35 @@ async function clone(project, root, workspace, dest, env) {
return { outcome: "cloned", branch: at.branch };
}
-async function update(root, dest, env) {
- const problem = await unsafe(dest, root, env);
+async function update(root, dest, g) {
+ const problem = await unsafe(dest, root, g);
if (problem) return { outcome: "kept", reason: problem };
// ls-files reads only the index: no filter, hook or submodule runs.
- const staged = await git([...inClone(dest), "ls-files", "--stage", "-z"], env);
+ const staged = await g([...inClone(dest), "ls-files", "--stage", "-z"]);
if (!staged.ok) return { outcome: "kept", reason: "status-failed" };
if (staged.out.split("\0").some((e) => e.startsWith("160000 "))) return { outcome: "kept", reason: "submodule" };
- const at = await rootBranch(root, env);
+ const at = await rootBranch(root, g);
if (at.error) return { outcome: "kept", reason: "root-unreadable" };
if (!at.branch) return { outcome: "kept", reason: "root-detached" };
- const head = await git([...inClone(dest), "symbolic-ref", "-q", "--short", "HEAD"], env);
+ const head = await g([...inClone(dest), "symbolic-ref", "-q", "--short", "HEAD"]);
if (!head.ok || head.out.trim() !== at.branch) return { outcome: "kept", reason: "other-branch" };
- const status = await git(["--no-optional-locks", ...inClone(dest), "status", "--porcelain=v1", "-z", "--untracked-files=all", "--ignore-submodules=all"], env);
+ const status = await g(["--no-optional-locks", ...inClone(dest), "status", "--porcelain=v1", "-z", "--untracked-files=all", "--ignore-submodules=all"]);
if (!status.ok) return { outcome: "kept", reason: "status-failed" };
if (status.out !== "") return { outcome: "kept", reason: "dirty" };
- const before = (await git([...inClone(dest), "rev-parse", "HEAD"], env)).out.trim();
- if (!(await git([...inClone(dest), "fetch", "-q", "--no-tags", "--no-recurse-submodules", "origin"], env)).ok) return { outcome: "kept", reason: "fetch-failed" };
- if (!(await git([...inClone(dest), "merge", "-q", "--ff-only", "--no-edit", `refs/remotes/origin/${at.branch}`], env)).ok) return { outcome: "kept", reason: "not-fast-forward" };
- const after = (await git([...inClone(dest), "rev-parse", "HEAD"], env)).out.trim();
+ const before = (await g([...inClone(dest), "rev-parse", "HEAD"])).out.trim();
+ if (!(await g([...inClone(dest), "fetch", "-q", "--no-tags", "--no-recurse-submodules", "origin"])).ok) return { outcome: "kept", reason: "fetch-failed" };
+ if (!(await g([...inClone(dest), "merge", "-q", "--ff-only", "--no-edit", `refs/remotes/origin/${at.branch}`])).ok) return { outcome: "kept", reason: "not-fast-forward" };
+ const after = (await g([...inClone(dest), "rev-parse", "HEAD"])).out.trim();
return { outcome: before === after ? "current" : "updated", branch: at.branch };
}
-// await ensureCheckouts({ workspace, projects, env }) for the business file's
-// projects ({ <id>: { root } }). Returns one { project, path, outcome,
-// reason?, branch? } per project, outcome cloned, updated, current or kept.
-// Rejects with CheckoutError when a missing clone can't be made.
-export async function ensureCheckouts({ workspace, projects, env = process.env }) {
+// await ensureCheckouts({ workspace, projects, env, limits }) for the
+// business file's projects ({ <id>: { root } }). Returns one { project,
+// path, outcome, reason?, branch? } per project, outcome cloned, updated,
+// current or kept. Rejects with CheckoutError when a missing clone can't be
+// made. `limits` ({ timeoutMs, graceMs }) is for the tests.
+export async function ensureCheckouts({ workspace, projects, env = process.env, limits = GIT_LIMITS }) {
+ const g = (args, extra) => git(args, env, extra, limits);
const out = [];
for (const [project, { root }] of Object.entries(projects)) {
const dest = join(workspace, project);
@@ -238,7 +277,7 @@ export async function ensureCheckouts({ workspace, projects, env = process.env }
if (e.code !== "ENOENT") throw new CheckoutError(project, `${dest} can't be read: ${e.code}`);
exists = false;
}
- out.push({ project, path: dest, ...(exists ? await update(root, dest, env) : await clone(project, root, workspace, dest, env)) });
+ out.push({ project, path: dest, ...(exists ? await update(root, dest, g) : await clone(project, root, workspace, dest, g)) });
}
return out;
}
diff --git a/packages/seat/src/working.mjs b/packages/seat/src/working.mjs
index 3b8787972..59e621fa1 100644
--- a/packages/seat/src/working.mjs
+++ b/packages/seat/src/working.mjs
@@ -34,22 +34,24 @@
// deletes it.
//
// A git past GIT_TIMEOUT_MS gets SIGTERM, on which git removes its lockfiles,
-// and SIGKILL only if it still runs GIT_GRACE_MS later. A SIGKILL leaves a
-// ref's .lock behind, and that lock refuses every later push to the ref.
+// and SIGKILL only if it still runs GIT_GRACE_MS later (checkout.mjs). Both
+// go to git's process group, so nothing git started outlives it. A SIGKILL
+// leaves a ref's .lock behind, and that lock refuses every later push to the
+// ref.
//
// The lock is a promise chain in this process, so it holds only while one
-// launcher serves a dataRoot. startHost refuses a second bus host while
-// bus-host/host.json names a live one; that check isn't atomic.
+// launcher serves a dataRoot. startHost refuses a second bus host for the
+// dataRoot: it takes bus-host/host.lock with a hard link, which fails if
+// the file exists.
//
// review, under the same lock: main again from the root, then a bundle of
// the requested commit that leaves out the commits the reviewer said it has
// (those the repository knows), written to the file the caller names.
-import { spawn } from "node:child_process";
import { randomBytes } from "node:crypto";
import { closeSync, lstatSync, mkdirSync, openSync, readSync, renameSync, rmSync } from "node:fs";
import { dirname, join } from "node:path";
-import { gitEnv, GIT_TIMEOUT_MS } from "./checkout.mjs";
+import { gitEnv, GIT_LIMITS, runGit } from "./checkout.mjs";
// The largest bundle the host takes or sends.
export const BUNDLE_MAX = 32 * 1024 * 1024;
@@ -59,9 +61,6 @@ const V2 = "# v2 git bundle\n";
const OID = /^[0-9a-f]{40}([0-9a-f]{24})?$/;
const TASK_REF = /^vikunja:([1-9][0-9]*)\/([1-9][0-9]*)$/;
const PROJECT = /^[a-z0-9][a-z0-9._-]{0,63}$/;
-// How long a git that got SIGTERM has to exit before SIGKILL.
-export const GIT_GRACE_MS = 5000;
-const LIMITS = { timeoutMs: GIT_TIMEOUT_MS, graceMs: GIT_GRACE_MS };
export class WorkingError extends Error {
constructor(code, message = code) {
@@ -84,28 +83,9 @@ const SAFE = [
// Ahead of the fetch from a bundle. Tests replace it to show it matters.
export const FSCK = ["-c", "transfer.fsckObjects=true"];
-// `limits` is { timeoutMs, graceMs }; the tests shorten them.
-function git(args, env, { pre = [], limits = LIMITS } = {}) {
- return new Promise((resolve) => {
- const child = spawn("git", [...SAFE, ...pre, ...args], { env: gitEnv(env), stdio: ["ignore", "pipe", "pipe"] });
- let out = "";
- let err = "";
- let kill = null;
- const term = setTimeout(() => {
- child.kill("SIGTERM");
- kill = setTimeout(() => child.kill("SIGKILL"), limits.graceMs);
- }, limits.timeoutMs);
- const settle = (r) => {
- clearTimeout(term);
- clearTimeout(kill);
- resolve(r);
- };
- child.stdout.setEncoding("utf8").on("data", (d) => (out += d));
- child.stderr.setEncoding("utf8").on("data", (d) => (err += d));
- child.on("error", (e) => settle({ ok: false, out, err, why: e.code ?? e.message }));
- child.on("close", (code, signal) => settle({ ok: code === 0, out, err, why: signal ? `killed by ${signal}` : err.trim().split("\n").pop() || `exit ${code}` }));
- });
-}
+// `limits` is { timeoutMs, graceMs }; the tests shorten them. runGit
+// (checkout.mjs) runs git in its own process group and signals the group.
+const git = (args, env, { pre = [], limits = GIT_LIMITS } = {}) => runGit([...SAFE, ...pre, ...args], gitEnv(env), limits);
// git with this call's env and limits, for the helpers below.
const bound = (env, limits) => (args, opts = {}) => git(args, env, { ...opts, limits });
@@ -187,7 +167,7 @@ const tip = async (repo, ref, g) => {
// Resolves { ref, commit, previous }; rejects with WorkingError. `record`
// runs under the lock once the ref moved, and a throw moves it back. `fsck`
// and `limits` are for the tests.
-export function push({ dataRoot, business, project, root, taskRef, branch, bundle, record = async () => {}, env = process.env, fsck = FSCK, limits = LIMITS }) {
+export function push({ dataRoot, business, project, root, taskRef, branch, bundle, record = async () => {}, env = process.env, fsck = FSCK, limits = GIT_LIMITS }) {
if (!PROJECT.test(project ?? "")) refuse("invalid-request", "not a project id");
if (!validBranch(branch)) refuse("invalid-request", "not a branch name");
const ref = `refs/heads/${workRef(taskRef)}`;
@@ -240,7 +220,7 @@ export function push({ dataRoot, business, project, root, taskRef, branch, bundl
// Writes a bundle of `commit` to `out`, with one ref, refs/mosaic/review.
// `have` lists commits the reviewer's clone holds; those the repository
// doesn't know are skipped. Resolves { ref, commit }.
-export function review({ dataRoot, business, project, root, taskRef, commit, have = [], out, env = process.env, limits = LIMITS }) {
+export function review({ dataRoot, business, project, root, taskRef, commit, have = [], out, env = process.env, limits = GIT_LIMITS }) {
if (!PROJECT.test(project ?? "") || !OID.test(commit ?? "")) refuse("invalid-request");
if (!Array.isArray(have) || have.length > HAVE_MAX || have.some((h) => !OID.test(h ?? ""))) refuse("invalid-request", "have must list commit ids");
const repo = workingRepo(dataRoot, business, project);
diff --git a/packages/seat/tests/checkout.test.mjs b/packages/seat/tests/checkout.test.mjs
index 8dea05826..8e2344165 100644
--- a/packages/seat/tests/checkout.test.mjs
+++ b/packages/seat/tests/checkout.test.mjs
@@ -331,3 +331,80 @@ test("every project in the business file gets its own clone at its own branch",
assert.deepEqual(list.map((c) => [c.project, c.path, c.outcome, c.branch]), [["alpha", join(workspace, "alpha"), "cloned", "main"], ["beta", join(workspace, "beta"), "cloned", "dev"]]);
assert.equal(git(join(workspace, "beta"), "symbolic-ref", "--short", "HEAD"), "dev");
});
+
+// A git on PATH that, for `git clone`, runs `body` in place of the real one.
+// Every other git is the real one.
+function slowClone(w, body) {
+ const bin = join(w.dir, "bin");
+ mkdirSync(bin);
+ const real = spawnSync("sh", ["-c", "command -v git"], { encoding: "utf8" }).stdout.trim();
+ writeFileSync(join(bin, "git"), `#!/bin/bash\ncase " $* " in *" clone "*) ${body} ;; esac\nexec '${real}' "$@"\n`, { mode: 0o755 });
+ return { PATH: `${bin}:${process.env.PATH}` };
+}
+
+// True once `pid` has gone, or is a zombie waiting for init to reap it.
+async function gone(pid, ms = 3000) {
+ const end = Date.now() + ms;
+ for (;;) {
+ let stat = null;
+ try {
+ stat = readFileSync(`/proc/${pid}/stat`, "utf8");
+ } catch {}
+ if (stat === null || stat.slice(stat.lastIndexOf(")") + 2)[0] === "Z") return true;
+ if (Date.now() > end) return false;
+ await new Promise((r) => setTimeout(r, 50));
+ }
+}
+
+// SIGKILL to a sleep a test left behind, if it still runs.
+const kill = (pid) => {
+ try {
+ process.kill(pid, "SIGKILL");
+ } catch {}
+};
+
+// The pid a wrapper wrote to `file`, once it is there.
+async function pidIn(file) {
+ for (let i = 0; i < 100 && !existsSync(file); i++) await new Promise((r) => setTimeout(r, 20));
+ return Number(readFileSync(file, "utf8").trim());
+}
+
+test("a clone past its time gets SIGTERM and refuses the launch, with nothing left in the workspace", async (t) => {
+ const w = world(t);
+ const env = slowClone(w, "exec sleep 30");
+ await assert.rejects(ensureCheckouts({ workspace: w.workspace, projects: { stack: { root: w.root } }, env, limits: { timeoutMs: 500, graceMs: 20000 } }), (e) => e instanceof CheckoutError && /killed by SIGTERM/.test(e.message));
+ assert.deepEqual(readdirSync(w.workspace), []);
+});
+
+test("a clone that ignores SIGTERM gets SIGKILL once the grace period is over", { timeout: 20000 }, async (t) => {
+ const w = world(t);
+ const env = slowClone(w, "trap '' TERM; sleep 30");
+ const started = Date.now();
+ await assert.rejects(ensureCheckouts({ workspace: w.workspace, projects: { stack: { root: w.root } }, env, limits: { timeoutMs: 300, graceMs: 700 } }), (e) => e instanceof CheckoutError && /killed by SIGKILL/.test(e.message));
+ assert.ok(Date.now() - started >= 1000, "SIGKILL waited out the grace period");
+});
+
+// git is the leader of its own process group, and the group gets the signal:
+// a process git started (here a sleep) doesn't outlive the kill.
+test("a timeout signals git's process group, so a process git started dies with it", { timeout: 20000 }, async (t) => {
+ const w = world(t);
+ const file = join(w.dir, "child.pid");
+ const env = slowClone(w, `sleep 30 </dev/null >/dev/null 2>&1 & echo $! > '${file}'; wait`);
+ await assert.rejects(ensureCheckouts({ workspace: w.workspace, projects: { stack: { root: w.root } }, env, limits: { timeoutMs: 500, graceMs: 20000 } }), /killed by SIGTERM/);
+ const pid = await pidIn(file);
+ t.after(() => kill(pid));
+ assert.equal(await gone(pid), true, "the sleep git started is gone");
+});
+
+// git dies of SIGTERM, but a process it started ignores SIGTERM and holds
+// none of git's output, so git's close comes at once. The group gets
+// SIGKILL then, not after the grace period.
+test("once a signalled git has closed, its group gets SIGKILL, so a process that ignored SIGTERM dies too", { timeout: 20000 }, async (t) => {
+ const w = world(t);
+ const file = join(w.dir, "child.pid");
+ const env = slowClone(w, `(trap '' TERM; exec sleep 30) </dev/null >/dev/null 2>&1 & echo $! > '${file}'; exec sleep 30`);
+ await assert.rejects(ensureCheckouts({ workspace: w.workspace, projects: { stack: { root: w.root } }, env, limits: { timeoutMs: 500, graceMs: 20000 } }), /killed by SIGTERM/);
+ const pid = await pidIn(file);
+ t.after(() => kill(pid));
+ assert.equal(await gone(pid, 1500), true, "the sleep that ignored SIGTERM is gone well inside the grace period");
+});
diff --git a/roles/reviewer.md b/roles/reviewer.md
index a349ad540..782aa156e 100644
--- a/roles/reviewer.md
+++ b/roles/reviewer.md
@@ -9,6 +9,7 @@ candidates and record a verdict. You don't change the code you review.
acceptance criteria. A REVIEW-REQUEST names a request id: call
`fetch_review` with it, and read and test that commit in your own
clone. Run the suites yourself when the request says they pass.
+- A `nothing-to-fetch` refusal means a clone of yours has the commit; see "Git socket" in `packages/cli/README.md`.
- Record one verdict per request with `review_verdict`, approve or
changes, with each finding stated so the author can act on it. The bus
takes one verdict per request; a changed mind needs a new request.
Reference in New Issue View Git Blame Copy Permalink