From 465bbdfbbc8cf2152aa05f025f163e36296ff484 Mon Sep 17 00:00:00 2001 From: Jason Woltje Date: Tue, 13 Oct 2026 16:34:47 -0500 Subject: [PATCH] seat, cli, harness: git kill reach, close bound and host.lock (#1597) Row 103, the row 100 follow-up. Built by Darkwing, packet agents/darkwing/work/queue-103/r1/, candidate manifest 6e5ac8c0782bc595d7c0d711d457738ffb42f1ba5f38253912943dd55e55888a. Sage approved on #1597, comment 28252; Dewey approved, comment 28260. Host git runs through runGit: a detached group, SIGTERM at the limit, SIGKILL after the grace, and one more SIGKILL to the group after a signalled git closes. git-socket close() waits at most 2 s. startHost claims bus-host/host.lock by hard link before the broker fork. packages/cli/README.md and tests/host.test.mjs carry row 85's hunks too; row 85 lands first. Sage's gate rerun on 3a1e11b83 with rows 85, 102 and 103 applied, Docker up: node suites bus 128, business 67, cli 165, control-board 124, conversation 182, discord 178, harness 73, ledger 78, mosaic 69, queue 155, runs 43, seat 113, tasks 92, webui 108; scripts auth 15, conductor 17, config 24, discord 66, extension-package 18, foundation 44, queue 28, release 14, task 98. All pass, 0 fail. Co-Authored-By: Claude Opus 5.5 --- packages/cli/README.md | 38 ++++-- packages/cli/src/git-socket.mjs | 44 +++++-- packages/cli/src/host.mjs | 87 ++++++++++++- packages/cli/tests/git-socket.test.mjs | 80 +++++++++++- packages/cli/tests/host.test.mjs | 149 +++++++++++++++++++++- packages/harness/src/git-tools.mjs | 44 ++++++- packages/harness/tests/git-tools.test.mjs | 89 ++++++++++++- packages/seat/README.md | 21 ++- packages/seat/src/checkout.mjs | 95 ++++++++++---- packages/seat/src/working.mjs | 46 ++----- packages/seat/tests/checkout.test.mjs | 77 +++++++++++ roles/reviewer.md | 1 + 12 files changed, 665 insertions(+), 106 deletions(-) diff --git a/packages/cli/README.md b/packages/cli/README.md index 5328b8a51..5baebde27 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: []`. 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 `/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 + (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: `. 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 `/bus-host/host.json` (0600): the pid, the process +6. It writes `/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, `/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 e69b87de1..7dc05a7f9 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 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 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..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 ({ : { 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 ({ : { 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 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 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.