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 <[email protected]>

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 });

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;
   }