cli, seat, harness: git socket hardening and refusal tests, row 100 (#1594)

Row 93 follow-up: SIGTERM then SIGKILL for host git, rollback-failed, a total body deadline, fetch-aborted for a reviewer gone mid-fetch, working refs first in have, and tests for five untested refusals. Dewey approved in 28226 and Sage in 28225; Sage's full gate passed 16:57:32-17:01:32Z. The host side is live at the next bus restart. Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-10-13 12:15:52 -05:00
parent c8302a0675
commit 664199f794
8 changed files with 619 additions and 62 deletions
@@ -336,12 +336,31 @@ refuses, the ref goes back. A refusal is `{ok: false, error: <code>}` and
 ends the connection. The host's own refusals go through the broker's
 `refuse` with the task ref as subject, so each is an `action.refused`
 event on the task's trail. Eight connections are served at once; more
-refuse `git-busy`.
+refuse `git-busy`. A connection holds its slot until it ends, so a push's
+bundle has to come within a deadline: no gap over 30 s, and all of it
+within 30 s plus 1 s per 256 KB (`bodyDeadline`, 161 s at the 32 MiB cap).
+Past either it refuses `body-timeout`.
+
+Codes a reviewer may see that aren't the broker's:
+
+- `nothing-to-fetch`: the reviewer's clones already have the commit, by the
+  `have` list the runner sent, so there is no bundle to send. The commit is
+  already in one of its clones under `refs/remotes/`.
+- `unknown-project`: the review names a project this business no longer
+  has.
+- `fetch-aborted`: the reviewer's side closed the connection before the
+  whole bundle was sent. That is the reviewer's doing, not the host's, so
+  it isn't logged.
+
+`git-failed` (an error in the host that has no code of its own) and
+`rollback-failed` (a push the broker refused whose work ref couldn't go
+back) are the host's own faults, and the host logs them as well.

 A bundle file lives in `bus-host/` (0700) under a random name and is
-deleted on every path. On close the launcher drops the git socket's
-connections and waits for any push in flight to finish under its lock
-before the broker stops.
+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.

 ### Trackers

@@ -29,6 +29,15 @@
 // The host's own refusals go through the broker's refuse(), so each is an
 // action.refused event on the task's trail; the broker records its own.
 //
+// A bundle has to arrive within bodyDeadline(size) and with no gap longer
+// than BODY_IDLE_MS (body-timeout), so a slow sender can't hold one of the
+// MAX_ACTIVE slots. A reviewer gone before its bundle was sent is
+// fetch-aborted. git-failed and rollback-failed are the host's own faults,
+// 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.
+//
 // A bundle file lives in bus-host/ (0700) under a random name, and the host
 // deletes it on every path.

@@ -36,7 +45,6 @@ import { randomBytes } from "node:crypto";
 import { closeSync, createReadStream, existsSync, lstatSync, openSync, rmSync, statSync, writeSync, chmodSync } from "node:fs";
 import { createServer } from "node:net";
 import { join } from "node:path";
-import { pipeline } from "node:stream/promises";
 import { gitSocketPath } from "../../seat/src/session.mjs";
 import { BUNDLE_MAX, FSCK, HAVE_MAX, push, review, validBranch, workRef, WorkingError } from "../../seat/src/working.mjs";
 import { CliError } from "./errors.mjs";
@@ -45,6 +53,13 @@ const HEADER_MAX = 4096;
 const HEADER_TIMEOUT_MS = 10000;
 // The longest the host waits between two chunks of a bundle.
 const BODY_IDLE_MS = 30000;
+// The slowest rate a bundle may come at, in bytes per millisecond (256 KB/s).
+// The runner sends from a file over a local socket, so a real push is far
+// faster.
+const BODY_MIN_RATE = 256;
+// The longest a whole bundle of `size` bytes may take: 30 s plus `size` at
+// BODY_MIN_RATE, so 161 s at the 32 MiB cap.
+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;
 const CAP = /^[0-9a-f]{64}$/;
@@ -65,11 +80,13 @@ class Recorded extends Error {
  }
 }

-// Resolves { req, rest } once a whole first line came; rest is what followed it.
-function readHeader(socket) {
+// Resolves { req, rest } once a whole first line came; rest is what followed
+// it. Until then `waiting` holds a function that refuses it host-closing.
+function readHeader(socket, waiting) {
  return new Promise((resolve, reject) => {
    let buf = Buffer.alloc(0);
    const done = (err, value) => {
+      waiting.delete(socket);
      socket.off("data", onData);
      socket.off("end", onGone);
      socket.off("close", onGone);
@@ -92,6 +109,7 @@ function readHeader(socket) {
      }
      done(null, { req, rest: buf.subarray(nl + 1) });
    };
+    waiting.set(socket, () => done(new Refusal("host-closing")));
    socket.on("data", onData);
    socket.once("end", onGone);
    socket.once("close", onGone);
@@ -100,8 +118,9 @@ function readHeader(socket) {
  });
 }

-// Exactly `size` bytes from the socket into a new file (0600).
-function readBody(socket, file, size) {
+// 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) {
  return new Promise((resolve, reject) => {
    const fd = openSync(file, "wx", 0o600);
    let got = 0;
@@ -109,16 +128,19 @@ function readBody(socket, file, size) {
    const done = (err) => {
      if (settled) return;
      settled = true;
+      clearTimeout(deadline);
      socket.off("data", onData);
      socket.off("end", onShort);
      socket.off("close", onShort);
-      socket.off("timeout", onShort);
+      socket.off("timeout", onIdle);
      socket.setTimeout(0);
      socket.pause();
      closeSync(fd);
      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 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"));
      try {
@@ -133,20 +155,59 @@ function readBody(socket, file, size) {
    socket.once("end", onShort);
    socket.once("close", onShort);
    socket.setTimeout(BODY_IDLE_MS);
-    socket.once("timeout", onShort);
+    socket.once("timeout", onIdle);
    socket.resume();
  });
 }

-// host: startHost()'s handle; business: loadBusiness() output. bundleMax and
-// fsck are for the tests.
-export function createGitSocket({ host, business: b, dataRoot, env = process.env, log = () => {}, bundleMax = BUNDLE_MAX, fsck = FSCK }) {
+// The file's bytes to the socket, then its end. A failed read of the host's
+// 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.
+export async function send(socket, file) {
+  const aborted = () => new Refusal("fetch-aborted", "the reviewer left before the bundle was sent");
+  const served = () => !socket.errored;
+  // Resolves on `event`; a close first resolves too unless a write failed.
+  const until = (event) =>
+    new Promise((resolve, reject) => {
+      const ok = () => {
+        socket.off("close", gone);
+        resolve();
+      };
+      const gone = () => {
+        socket.off(event, ok);
+        served() ? resolve() : reject(aborted());
+      };
+      socket.once(event, ok);
+      socket.once("close", gone);
+    });
+  for await (const chunk of createReadStream(file)) {
+    if (socket.destroyed || socket.writableEnded) throw aborted();
+    if (!socket.write(chunk)) await until("drain");
+  }
+  if (socket.destroyed) {
+    if (served()) return;
+    throw aborted();
+  }
+  const flushed = until("finish");
+  socket.end();
+  await flushed;
+  if (!served()) throw aborted();
+}
+
+// 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 }) {
  const path = gitSocketPath(dataRoot);
  const dir = join(dataRoot, "bus-host");
  let server = null;
  let closed = false;
  const sockets = new Set();
  const active = new Set();
+  // Connections still reading their request line (readHeader).
+  const waiting = new Map();

  const op = async (message) => {
    try {
@@ -167,7 +228,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);
+    await readBody(socket, file, size, deadlineOf(size));
    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 } }),
@@ -186,7 +247,7 @@ export function createGitSocket({ host, business: b, dataRoot, env = process.env
    const size = statSync(file).size;
    if (size > bundleMax) throw new Refusal("bundle-too-large", `${size} bytes, over ${bundleMax}`);
    socket.write(`${JSON.stringify({ ok: true, result: { project: r.project, ref: made.ref, commit: made.commit, size } })}\n`);
-    await pipeline(createReadStream(file), socket);
+    await send(socket, file);
  }

  async function serve(socket) {
@@ -197,7 +258,7 @@ export function createGitSocket({ host, business: b, dataRoot, env = process.env
      if (!socket.destroyed) socket.end(`${JSON.stringify({ ok: false, error: code })}\n`);
    };
    try {
-      const { req, rest } = await readHeader(socket);
+      const { req, rest } = await readHeader(socket, waiting);
      if (typeof req?.cap !== "string" || !CAP.test(req.cap)) return fail("unauthenticated");
      if (closed) return fail("host-closing");
      let who;
@@ -216,7 +277,7 @@ export function createGitSocket({ host, business: b, dataRoot, env = process.env
    } catch (e) {
      if (e instanceof Recorded) return fail(e.code);
      const code = e instanceof Refusal || e instanceof WorkingError ? e.code : "git-failed";
-      if (code === "git-failed") log(`git socket: ${e.message}`);
+      if (code === "git-failed" || code === "rollback-failed") log(`git socket: ${code}: ${e.message}`);
      if (cap) await host.op({ op: "refuse", cap, code, subject }).catch(() => {});
      fail(code);
    } finally {
@@ -233,7 +294,7 @@ export function createGitSocket({ host, business: b, dataRoot, env = process.env
      sockets.add(socket);
      socket.once("close", () => sockets.delete(socket));
      socket.on("error", () => {});
-      if (active.size >= MAX_ACTIVE) return socket.end(`${JSON.stringify({ ok: false, error: "git-busy" })}\n`);
+      if (active.size >= maxActive) return socket.end(`${JSON.stringify({ ok: false, error: "git-busy" })}\n`);
      const job = serve(socket).catch((e) => log(`git socket: ${e.message}`));
      active.add(job);
      job.finally(() => active.delete(job));
@@ -248,12 +309,19 @@ export function createGitSocket({ host, business: b, dataRoot, env = process.env

  // A push whose bundle already came finishes, under its lock, while the
  // broker still runs: the launcher's close() runs before the host stops it.
+  // A connection still waiting for its request line hears host-closing first.
  async function close() {
    closed = true;
    if (!server) return;
    const stopped = new Promise((r) => server.close(r));
-    for (const socket of sockets) socket.destroy();
+    const told = new Set(waiting.keys());
+    for (const stop of [...waiting.values()]) stop();
+    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))))));
+    for (const socket of sockets) socket.destroy();
    await stopped;
    rmSync(path, { force: true });
  }
@@ -12,6 +12,7 @@ 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 { tmpdir } from "node:os";
+import { Duplex } from "node:stream";
 import { join } from "node:path";
 import { Client } from "../../bus/src/client.mjs";
 import { configDir, loadBusiness } from "../../business/src/index.mjs";
@@ -20,7 +21,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 } from "../src/git-socket.mjs";
+import { createGitSocket, send } from "../src/git-socket.mjs";
 import { startHost, startTimeOf } from "../src/host.mjs";
 import { fixture } from "./helpers.mjs";

@@ -56,7 +57,8 @@ const bundles = (dir) => (existsSync(dir) ? readdirSync(dir).filter((n) => n.end

 // A host for business acme, its git socket, and the coder, reviewer and pm
 // bound and claimed. Each role has a workspace with a clone of the project.
-async function setup(t, git = {}) {
+// The socket sees wrap(host), so a test can change one answer of the host's.
+async function setup(t, { wrap = (h) => h, ...git } = {}) {
  // One teardown, in order: stop the socket and the host, check, then remove
  // the tree. (t.after hooks run first-registered first.)
  const root = mkdtempSync(join(tmpdir(), "mosaic-git-"));
@@ -78,7 +80,7 @@ async function setup(t, git = {}) {
  const host = await startHost({ boot: bootConfig({ system, businessId: "acme", env: f.env }), business: "acme", log: () => {} });
  down.push(() => host.close(0));
  const logs = [];
-  const sock = createGitSocket({ host, business: loadBusiness("acme", { dir: configDir(f.env), rolesDir: rolesDir(f.env) }), dataRoot: f.dataRoot, env: f.env, log: (l) => logs.push(l), ...git });
+  const sock = createGitSocket({ host: wrap(host), business: loadBusiness("acme", { dir: configDir(f.env), rolesDir: rolesDir(f.env) }), dataRoot: f.dataRoot, env: f.env, log: (l) => logs.push(l), ...git });
  down.unshift(() => sock.close());
  await sock.listen();
  const roles = {};
@@ -120,6 +122,32 @@ function raw(path, text) {
  });
 }

+// A connection that has sent `text` and waits. hangUp() ends it; reply
+// resolves with whatever the host sent once the connection is closed.
+function idle(path, text = "") {
+  return new Promise((resolve, reject) => {
+    const s = connect(path);
+    let buf = "";
+    s.on("data", (b) => (buf += b));
+    s.on("error", () => {});
+    const reply = new Promise((r) => s.on("close", () => r(buf)));
+    s.on("connect", () => {
+      if (text) s.write(text);
+      resolve({ reply, hangUp: () => (s.end(), reply) });
+    });
+    s.once("error", reject);
+  });
+}
+
+// Polls fn() until it is truthy, for up to 5 s.
+async function until(fn, what) {
+  for (const end = Date.now() + 5000; Date.now() < end; await new Promise((r) => setTimeout(r, 20))) if (await fn()) return;
+  assert.fail(`timed out waiting for ${what}`);
+}
+
+const refusedOn = async (w, subject = TASK) => (await w.trail("action.refused", subject)).map((e) => `${e.actor_role} ${e.body.code}`);
+const pushReq = (cap, size) => `${JSON.stringify({ cap, op: "push", task_ref: TASK, project: "stack", branch: "feat", size })}\n`;
+
 test("the review loop: the coder pushes and asks, the reviewer fetches and approves, and the task's trail holds each step", async (t) => {
  const w = await setup(t);
  assert.equal(statSync(w.sock.path).mode & 0o777, 0o600);
@@ -153,6 +181,11 @@ test("the review loop: the coder pushes and asks, the reviewer fetches and appro
  assert.deepEqual(pushes.map((e) => [e.actor_role, e.body.commit, e.body.previous, e.body.branch]), [["coder", c1, null, "feat"], ["coder", c2, c1, "feat"]]);
  assert.deepEqual((await w.trail("review.requested")).map((e) => e.id), [req.request, req2.request]);
  assert.deepEqual((await w.trail("review.verdict")).map((e) => [e.id, e.body.verdict, e.body.commit]), [[v.verdict, "changes", c1], [ok.verdict, "approve", c2]]);
+  // The reviewer hangs up once it has the bundle, which can beat the host's
+  // own end of the send. close() waits for each connection's job, so a fetch
+  // filed as anything but done would be on the trail by now.
+  await w.sock.close();
+  assert.deepEqual(await refusedOn(w), []);
  assert.deepEqual(w.logs, []);
 });

@@ -314,3 +347,204 @@ test("a rigged clone runs nothing: hooks, fsmonitor, a filter, a gitlink and a s
  assert.equal(w.refs(), landed);
  assert.deepEqual(readdirSync(marks), []);
 });
+
+test("an identity with no role, or of another business, refuses not-a-role and moves no ref", async (t) => {
+  let who = null;
+  const w = await setup(t, { wrap: (host) => ({ op: async (m) => (m.op === "identity" && who ? { ...(await host.op(m)), ...who } : host.op(m)) }) });
+  const { coder } = w.roles;
+  gitIn(coder.clone, "checkout", "-q", "-b", "feat");
+  commit(coder.clone, "two");
+  for (const answer of [{ role: null }, { business: "other" }]) {
+    who = answer;
+    await rejects(w.push("coder"), "not-a-role");
+  }
+  assert.equal(w.refs(), "");
+  assert.deepEqual((await refusedOn(w)).filter((r) => r.endsWith("not-a-role")), ["coder not-a-role", "coder not-a-role"]);
+  who = null;
+  assert.equal((await w.push("coder")).ref, "work/vikunja-1-2");
+});
+
+test("a review of a project the business no longer has refuses unknown-project", async (t) => {
+  const w = await setup(t, { wrap: (host) => ({ op: async (m) => (m.op === "reviewFetch" ? { ...(await host.op(m)), project: "gone" } : host.op(m)) }) });
+  const { coder } = w.roles;
+  gitIn(coder.clone, "checkout", "-q", "-b", "feat");
+  const c1 = commit(coder.clone, "two");
+  await w.push("coder");
+  const req = await coder.client.call("review.request", { task_ref: TASK, commit: c1, to: "reviewer" });
+  await rejects(w.fetch("reviewer", req.request), "unknown-project");
+  assert.ok((await refusedOn(w)).includes("reviewer unknown-project"));
+  assert.deepEqual(w.logs, []);
+});
+
+test("past maxActive connections the socket refuses git-busy, and a hung-up connection frees its slot", async (t) => {
+  const w = await setup(t, { maxActive: 2 });
+  const held = [await idle(w.sock.path), await idle(w.sock.path)];
+  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ op: "push" })}\n`), { ok: false, error: "git-busy" });
+  await held[0].hangUp();
+  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ op: "push" })}\n`), { ok: false, error: "unauthenticated" });
+  await held[1].hangUp();
+});
+
+test("close() answers host-closing to a connection still sending its request line", async (t) => {
+  const w = await setup(t);
+  const waiting = await idle(w.sock.path, '{"cap":');
+  // Served in order, so a full exchange after it means the host took it.
+  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ op: "push" })}\n`), { ok: false, error: "unauthenticated" });
+  await w.sock.close();
+  assert.deepEqual(JSON.parse(await waiting.reply), { ok: false, error: "host-closing" });
+});
+
+test("a bundle that trickles in refuses body-timeout at its deadline and frees its slot", async (t) => {
+  const w = await setup(t, { maxActive: 2, bodyDeadline: () => 400 });
+  const { coder } = w.roles;
+  const other = await idle(w.sock.path);
+  const started = Date.now();
+  // One byte every 50 ms: 1000 bytes would take 50 s.
+  const trickled = new Promise((resolve, reject) => {
+    const s = connect(w.sock.path);
+    let buf = "";
+    let drip = null;
+    s.on("connect", () => s.write(pushReq(coder.cap, 1000)));
+    s.on("data", (b) => {
+      buf += b;
+      if (!drip && buf === `${JSON.stringify({ ok: true, send: 1000 })}\n`) drip = setInterval(() => s.writable && s.write("x"), 50);
+    });
+    s.on("close", () => {
+      clearInterval(drip);
+      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();
+  });
+  // The trickle holds the second slot while it lasts.
+  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ op: "push" })}\n`), { ok: false, error: "git-busy" });
+  const { last, ms } = await trickled;
+  assert.deepEqual(JSON.parse(last), { ok: false, error: "body-timeout" });
+  assert.ok(ms < 5000, `${ms} ms`);
+  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ op: "push" })}\n`), { ok: false, error: "unauthenticated" });
+  assert.ok((await refusedOn(w)).includes("coder body-timeout"));
+  assert.equal(w.refs(), "");
+  await other.hangUp();
+});
+
+test("a reviewer gone before its bundle was sent is filed fetch-aborted, not as the host's fault", async (t) => {
+  const w = await setup(t);
+  const { coder, reviewer } = w.roles;
+  gitIn(coder.clone, "checkout", "-q", "-b", "feat");
+  // Noise git can't compress, so the bundle can't all fit in the socket's buffers.
+  writeFileSync(join(coder.clone, "big"), randomBytes(4 * 1024 * 1024));
+  gitIn(coder.clone, "add", "big");
+  gitIn(coder.clone, "commit", "-q", "-m", "big");
+  const c = (await w.push("coder")).commit;
+  const req = await coder.client.call("review.request", { task_ref: TASK, commit: c, to: "reviewer" });
+  // Reads the result line and hangs up.
+  const line = await new Promise((resolve, reject) => {
+    const s = connect(w.sock.path);
+    let buf = "";
+    s.on("connect", () => s.write(`${JSON.stringify({ cap: reviewer.cap, op: "fetch", task_ref: TASK, request: req.request, have: [] })}\n`));
+    s.on("data", (b) => {
+      buf += b.toString("latin1");
+      const nl = buf.indexOf("\n");
+      if (nl >= 0) {
+        s.destroy();
+        resolve(JSON.parse(buf.slice(0, nl)));
+      }
+    });
+    s.on("error", reject);
+  });
+  assert.equal(line.ok, true);
+  assert.equal(line.result.commit, c);
+  await until(async () => (await refusedOn(w)).includes("reviewer fetch-aborted"), "fetch-aborted on the trail");
+  assert.ok(!(await refusedOn(w)).some((r) => r.endsWith("git-failed")));
+  assert.deepEqual(w.logs, []);
+});
+
+test("a reviewer that hangs up the moment it has the whole bundle was served, not aborted", async (t) => {
+  const w = await setup(t);
+  const { coder, reviewer } = w.roles;
+  gitIn(coder.clone, "checkout", "-q", "-b", "feat");
+  writeFileSync(join(coder.clone, "big"), randomBytes(1024 * 1024));
+  gitIn(coder.clone, "add", "big");
+  gitIn(coder.clone, "commit", "-q", "-m", "big");
+  const c = (await w.push("coder")).commit;
+  const req = await coder.client.call("review.request", { task_ref: TASK, commit: c, to: "reviewer" });
+  // Reads the result line and `size` bytes, then destroys the connection at
+  // once, which can beat the host's own end of the send. Ten times, since it
+  // is a race.
+  const take = () =>
+    new Promise((resolve, reject) => {
+      const s = connect(w.sock.path);
+      let buf = Buffer.alloc(0);
+      let size = null;
+      s.on("connect", () => s.write(`${JSON.stringify({ cap: reviewer.cap, op: "fetch", task_ref: TASK, request: req.request, have: [] })}\n`));
+      s.on("data", (b) => {
+        buf = Buffer.concat([buf, b]);
+        if (size === null) {
+          const nl = buf.indexOf(0x0a);
+          if (nl < 0) return;
+          size = JSON.parse(buf.subarray(0, nl).toString("utf8")).result.size;
+          buf = buf.subarray(nl + 1);
+        }
+        if (buf.length >= size) {
+          s.destroy();
+          resolve(buf.length);
+        }
+      });
+      s.on("error", reject);
+    });
+  for (let i = 0; i < 10; i++) assert.ok((await take()) > 0);
+  // close() waits for each connection's job.
+  await w.sock.close();
+  assert.deepEqual(await refusedOn(w), []);
+  assert.deepEqual(w.logs, []);
+});
+
+// The two closes the socket tests can't order every time, on a stand-in
+// socket: the reader gone once every byte left the host but before 'finish',
+// and the reader gone while the last chunk was still queued in the host.
+test("send: a close after every byte left is served; one that fails a queued write is fetch-aborted", async (t) => {
+  const dir = mkdtempSync(join(tmpdir(), "mosaic-send-"));
+  t.after(() => rmSync(dir, { recursive: true, force: true }));
+  const file = join(dir, "b.bundle");
+  writeFileSync(file, "ten bytes!");
+
+  // Takes every write at once and never finishes the end.
+  const quick = new Duplex({ read() {}, write: (c, e, cb) => cb(), final() {} });
+  quick.on("error", () => {});
+  const served = send(quick, file);
+  await until(() => quick.writableEnded, "the end");
+  quick.destroy();
+  await served;
+
+  // Holds the one write, then fails it as a socket whose reader left does.
+  let pending = null;
+  const slow = new Duplex({ read() {}, writableHighWaterMark: 1, write: (c, e, cb) => (pending = cb), final: (cb) => cb() });
+  slow.on("error", () => {});
+  const gone = send(slow, file);
+  await until(() => pending, "the write");
+  pending(Object.assign(new Error("write EPIPE"), { code: "EPIPE" }));
+  await rejects(gone, "fetch-aborted");
+});
+
+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, {
+    wrap: (host) => ({
+      op: async (m) => {
+        if (m.op !== "recordPush" || !lock) return host.op(m);
+        writeFileSync(lock, "");
+        throw Object.assign(new Error("not recorded"), { code: "store-failed" });
+      },
+    }),
+  });
+  const { coder } = w.roles;
+  gitIn(coder.clone, "checkout", "-q", "-b", "feat");
+  const c1 = commit(coder.clone, "two");
+  lock = join(w.repo, "refs", "heads", "work", "vikunja-1-2.lock");
+  await rejects(w.push("coder"), "rollback-failed");
+  assert.equal(w.refs(), `refs/heads/main ${w.main}\nrefs/heads/work/vikunja-1-2 ${c1}`);
+  assert.equal(w.logs.length, 1);
+  assert.match(w.logs[0], /^git socket: rollback-failed: refs\/heads\/work\/vikunja-1-2 still names /);
+  assert.ok((await refusedOn(w)).includes("coder rollback-failed"));
+  rmSync(lock);
+});
@@ -169,11 +169,16 @@ 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.
  const have = new Set();
-  for (const project of projectsIn(workspace)) {
-    const r = await git(["-C", join(workspace, project), "for-each-ref", "--format=%(objectname)", "refs/remotes/origin", "refs/remotes/working"], env);
-    if (r.ok) for (const oid of r.out.split("\n")) if (OID.test(oid) && have.size < HAVE_MAX) have.add(oid);
-  }
+  const projects = projectsIn(workspace);
+  for (const remote of ["refs/remotes/working", "refs/remotes/origin"])
+    for (const project of projects) {
+      if (have.size >= HAVE_MAX) break;
+      const r = await git(["-C", join(workspace, project), "for-each-ref", "--sort=-committerdate", "--format=%(objectname)", remote], env);
+      if (r.ok) for (const oid of r.out.split("\n")) if (OID.test(oid) && have.size < HAVE_MAX) have.add(oid);
+    }
  const file = join(tmp, `fetch-${randomBytes(8).toString("hex")}.bundle`);
  try {
    const conn = open(path);
@@ -0,0 +1,80 @@
+// The runner's git tools (src/git-tools.mjs) against a fake git socket that
+// records the request line. The review loop over the real host's socket is in
+// packages/cli/tests/git-socket.test.mjs.
+
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import { spawnSync } from "node:child_process";
+import { mkdirSync } from "node:fs";
+import { createServer } from "node:net";
+import { join } from "node:path";
+import { fetchReview, 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" };
+
+function git(cwd, args, extra = {}) {
+  const r = spawnSync("git", ["-c", "user.name=t", "-c", "[email protected]", "-C", cwd, ...args], { env: { ...ENV, ...extra }, encoding: "utf8" });
+  assert.equal(r.status, 0, `git ${args.join(" ")}: ${r.stderr}`);
+  return r.stdout.trim();
+}
+
+// A clone at `dir` with `refs` (name -> commit time) pointing at commits of
+// their own.
+function clone(dir, refs) {
+  mkdirSync(dir, { recursive: true });
+  git(dir, ["init", "-q"]);
+  const tree = git(dir, ["mktree"]);
+  const oids = {};
+  for (const [name, time] of Object.entries(refs)) {
+    const date = `${time} +0000`;
+    oids[name] = git(dir, ["commit-tree", tree, "-m", name], { GIT_AUTHOR_DATE: date, GIT_COMMITTER_DATE: date });
+    git(dir, ["update-ref", name, oids[name]]);
+  }
+  return oids;
+}
+
+// A socket that keeps each request line and refuses it.
+async function fakeHost(t, dir) {
+  const path = join(dir, "git.sock");
+  const lines = [];
+  const server = createServer((s) => {
+    let buf = "";
+    s.on("data", (b) => {
+      buf += b;
+      if (!buf.includes("\n")) return;
+      lines.push(JSON.parse(buf.split("\n")[0]));
+      s.end(`${JSON.stringify({ ok: false, error: "request-not-found" })}\n`);
+    });
+  });
+  await new Promise((r) => server.listen(path, r));
+  t.after(() => new Promise((r) => server.close(r)));
+  return { path, lines };
+}
+
+test("fetch_review's have lists the working refs before origin's, and no more than HAVE_MAX", async (t) => {
+  const dir = scratch(t);
+  const workspace = join(dir, "ws");
+  // 40 origin refs, newer than every working ref, and 13 working refs in two
+  // clones: 53 in all.
+  const stack = {};
+  for (let i = 0; i < 40; i++) stack[`refs/remotes/origin/b${i}`] = 1_800_000_000 + i;
+  for (let i = 1; i <= 10; i++) stack[`refs/remotes/working/work/vikunja-1-${i}`] = 1_700_000_000 + i;
+  const other = {};
+  for (let i = 1; i <= 3; i++) other[`refs/remotes/working/work/vikunja-2-${i}`] = 1_700_000_000 + i;
+  const a = clone(join(workspace, "stack"), stack);
+  const b = clone(join(workspace, "other"), other);
+  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;
+  assert.equal(req.op, "fetch");
+  assert.equal(HAVE_MAX, 32);
+  assert.equal(req.have.length, HAVE_MAX);
+  assert.equal(new Set(req.have).size, HAVE_MAX);
+  const working = [...Object.entries(a), ...Object.entries(b)].filter(([n]) => n.startsWith("refs/remotes/working/")).map(([, oid]) => oid);
+  assert.equal(working.length, 13);
+  // Every working ref, ahead of any of origin's.
+  assert.deepEqual(new Set(req.have.slice(0, 13)), new Set(working));
+  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);
+});
@@ -323,11 +323,25 @@ A push runs under the repository's lock, in this order:
   fast-forward (`non-fast-forward`).
 8. The broker records the push, an `action.allowed` for
   `git.push.working` naming the commit. When it
-   refuses, the work ref goes back to where it was.
+   refuses, the work ref goes back to where it was. A work ref that can't
+   go back refuses `rollback-failed` in place of the broker's code, with a
+   message naming the ref and the commit it still holds. The host logs it,
+   since the operator has to move that ref by hand.

 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.
+
+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.
+
 For a review, under the same lock, the host fetches `main` again and
 writes a bundle of the requested commit with one ref,
 `refs/mosaic/review`. It leaves out the commits the reviewer says its
@@ -28,10 +28,19 @@
 //   6. the commit shares history with main (unrelated-history)
 //   7. the temp ref moves to the work ref with no `+`, so git refuses
 //      anything but a fast-forward (non-fast-forward)
-//   8. record() runs; when it throws, the work ref goes back
+//   8. record() runs; when it throws, the work ref goes back. A work ref
+//      that can't go back refuses rollback-failed in place of record's error.
 // The temp ref is deleted on every path. The caller owns the bundle file and
 // 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.
+//
+// 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.
+//
 // 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.
@@ -50,6 +59,9 @@ 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) {
@@ -72,17 +84,30 @@ const SAFE = [
 // Ahead of the fetch from a bundle. Tests replace it to show it matters.
 export const FSCK = ["-c", "transfer.fsckObjects=true"];

-function git(args, env, { pre = [] } = {}) {
+// `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), timeout: GIT_TIMEOUT_MS, killSignal: "SIGKILL", stdio: ["ignore", "pipe", "pipe"] });
+    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) => 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}` }));
  });
 }
+// git with this call's env and limits, for the helpers below.
+const bound = (env, limits) => (args, opts = {}) => git(args, env, { ...opts, limits });

 // The branch name a task's work goes to. A task ref holds a colon, which a
 // ref name can't.
@@ -117,7 +142,7 @@ function locked(repo, fn) {

 // The bare repository, made on first use in a temp directory beside it and
 // renamed into place. One that exists must be a real directory.
-async function ensure(repo, env) {
+async function ensure(repo, g) {
  try {
    if (!lstatSync(repo).isDirectory()) refuse("working-repo-invalid", `${repo} isn't a directory`);
    return;
@@ -128,7 +153,7 @@ async function ensure(repo, env) {
  mkdirSync(dirname(repo), { recursive: true, mode: 0o700 });
  const tmp = join(dirname(repo), `.init-${randomBytes(4).toString("hex")}`);
  try {
-    const r = await git(["init", "-q", "--bare", "--template=", tmp], env);
+    const r = await g(["init", "-q", "--bare", "--template=", tmp]);
    if (!r.ok) refuse("working-repo-invalid", `git init failed: ${r.why}`);
    renameSync(tmp, repo);
  } finally {
@@ -138,8 +163,8 @@ async function ensure(repo, env) {

 // main from what the root has checked out. The root is the owner's; it isn't
 // a session's.
-async function seed(repo, root, env) {
-  const r = await git([`--git-dir=${repo}`, "fetch", "-q", "--no-tags", "--no-write-fetch-head", "--no-recurse-submodules", "--no-auto-maintenance", root, "+HEAD:refs/heads/main"], env);
+async function seed(repo, root, g) {
+  const r = await g([`--git-dir=${repo}`, "fetch", "-q", "--no-tags", "--no-write-fetch-head", "--no-recurse-submodules", "--no-auto-maintenance", root, "+HEAD:refs/heads/main"]);
  if (!r.ok) refuse("root-unreadable", `main couldn't be fetched from ${root}: ${r.why}`);
 }

@@ -153,68 +178,73 @@ function header(file) {
  }
 }

-const tip = async (repo, ref, env) => {
-  const r = await git([`--git-dir=${repo}`, "rev-parse", "-q", "--verify", `${ref}^{commit}`], env);
+const tip = async (repo, ref, g) => {
+  const r = await g([`--git-dir=${repo}`, "rev-parse", "-q", "--verify", `${ref}^{commit}`]);
  return r.ok ? r.out.trim() : null;
 };

-// await push({ dataRoot, business, project, root, taskRef, branch, bundle, record, env, fsck })
+// await push({ dataRoot, business, project, root, taskRef, branch, bundle, record, env, fsck, limits })
 // Resolves { ref, commit, previous }; rejects with WorkingError. `record`
 // runs under the lock once the ref moved, and a throw moves it back. `fsck`
-// is for the tests.
-export function push({ dataRoot, business, project, root, taskRef, branch, bundle, record = async () => {}, env = process.env, fsck = 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 }) {
  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)}`;
  const repo = workingRepo(dataRoot, business, project);
+  const g = bound(env, limits);
  return locked(repo, async () => {
    if (!header(bundle)) refuse("invalid-bundle", "not a v2 git bundle");
-    await ensure(repo, env);
-    await seed(repo, root, env);
+    await ensure(repo, g);
+    await seed(repo, root, g);
    const at = [`--git-dir=${repo}`];
    // Not -q: with it, git 2.55 drops the list of missing prerequisites.
-    const verify = await git([...at, "bundle", "verify", bundle], env);
+    const verify = await g([...at, "bundle", "verify", bundle]);
    if (!verify.ok) {
      if (/lacks these prerequisite/.test(verify.err)) refuse("missing-prerequisite", "the bundle needs commits the working repository doesn't have");
      refuse("invalid-bundle", verify.why);
    }
-    const heads = await git([...at, "bundle", "list-heads", bundle], env);
+    const heads = await g([...at, "bundle", "list-heads", bundle]);
    if (!heads.ok) refuse("invalid-bundle", heads.why);
    if (!heads.out.split("\n").some((l) => l.split(" ")[1] === `refs/heads/${branch}`)) refuse("branch-not-in-bundle");
    const incoming = `refs/mosaic/incoming/${randomBytes(8).toString("hex")}`;
    try {
-      const fetched = await git([...at, "fetch", "-q", "--no-tags", "--no-write-fetch-head", "--no-recurse-submodules", "--no-auto-maintenance", bundle, `refs/heads/${branch}:${incoming}`], env, { pre: fsck });
+      const fetched = await g([...at, "fetch", "-q", "--no-tags", "--no-write-fetch-head", "--no-recurse-submodules", "--no-auto-maintenance", bundle, `refs/heads/${branch}:${incoming}`], { pre: fsck });
      if (!fetched.ok) refuse("invalid-bundle", fetched.why);
-      const commit = await tip(repo, incoming, env);
+      const commit = await tip(repo, incoming, g);
      if (!commit) refuse("invalid-bundle", "the branch isn't a commit");
-      if (!(await git([...at, "merge-base", commit, "refs/heads/main"], env)).ok) refuse("unrelated-history", "the commit shares no history with main");
-      const previous = await tip(repo, ref, env);
+      if (!(await g([...at, "merge-base", commit, "refs/heads/main"])).ok) refuse("unrelated-history", "the commit shares no history with main");
+      const previous = await tip(repo, ref, g);
      // From the repository into itself, by its path (`.` would be the cwd), and
      // not -q, which drops the "(non-fast-forward)" reason.
-      const moved = await git([...at, "fetch", "--no-tags", "--no-write-fetch-head", "--no-auto-maintenance", repo, `${incoming}:${ref}`], env);
+      const moved = await g([...at, "fetch", "--no-tags", "--no-write-fetch-head", "--no-auto-maintenance", repo, `${incoming}:${ref}`]);
      if (!moved.ok) refuse(/\(non-fast-forward\)/.test(moved.err) ? "non-fast-forward" : "push-failed", moved.why);
      try {
        await record({ ref: workRef(taskRef), commit, previous });
      } catch (e) {
        const back = previous ? ["update-ref", ref, previous, commit] : ["update-ref", "-d", ref, commit];
-        await git([...at, ...back], env);
+        const undone = await g([...at, ...back]);
+        // The ref still names a commit the broker didn't record. That is the
+        // operator's to fix, so it outranks record's own error.
+        if (!undone.ok) refuse("rollback-failed", `${ref} still names ${commit}, which wasn't recorded (${e.code ?? e.message}); moving it back failed: ${undone.why}`);
        throw e;
      }
      return { ref: workRef(taskRef), commit, previous };
    } finally {
-      await git([...at, "update-ref", "-d", incoming], env);
+      await g([...at, "update-ref", "-d", incoming]);
    }
  });
 }

-// await review({ dataRoot, business, project, root, taskRef, commit, have, out, env })
+// await review({ dataRoot, business, project, root, taskRef, commit, have, out, env, limits })
 // 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 }) {
+export function review({ dataRoot, business, project, root, taskRef, commit, have = [], out, env = process.env, limits = 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);
+  const g = bound(env, limits);
  return locked(repo, async () => {
    try {
      if (!lstatSync(repo).isDirectory()) refuse("working-repo-invalid");
@@ -222,18 +252,18 @@ export function review({ dataRoot, business, project, root, taskRef, commit, hav
      if (e instanceof WorkingError) throw e;
      refuse("unknown-commit", "the working repository doesn't exist");
    }
-    await seed(repo, root, env);
+    await seed(repo, root, g);
    const at = [`--git-dir=${repo}`];
-    if (!(await tip(repo, commit, env))) refuse("unknown-commit");
+    if (!(await tip(repo, commit, g))) refuse("unknown-commit");
    const known = [];
-    for (const h of have) if (await tip(repo, h, env)) known.push(h);
+    for (const h of have) if (await tip(repo, h, g)) known.push(h);
    const temp = "refs/mosaic/review";
    try {
-      if (!(await git([...at, "update-ref", temp, commit], env)).ok) refuse("fetch-failed");
-      const made = await git([...at, "bundle", "create", "-q", out, temp, ...(known.length ? ["--not", ...known] : [])], env);
+      if (!(await g([...at, "update-ref", temp, commit])).ok) refuse("fetch-failed");
+      const made = await g([...at, "bundle", "create", "-q", out, temp, ...(known.length ? ["--not", ...known] : [])]);
      if (!made.ok) refuse(/empty bundle/.test(made.err) ? "nothing-to-fetch" : "fetch-failed", made.why);
    } finally {
-      await git([...at, "update-ref", "-d", temp], env);
+      await g([...at, "update-ref", "-d", temp]);
    }
    return { ref: workRef(taskRef), commit };
  });
@@ -8,7 +8,7 @@ import { test } from "node:test";
 import assert from "node:assert/strict";
 import { spawnSync } from "node:child_process";
 import { createHash } from "node:crypto";
-import { existsSync, lstatSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, readlinkSync, rmSync, writeFileSync } from "node:fs";
+import { chmodSync, existsSync, lstatSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, readlinkSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { gitEnv } from "../src/checkout.mjs";
@@ -87,6 +87,34 @@ async function refuses(w, fn, code) {
  assert.doesNotMatch(w.refs(), /refs\/mosaic\//, "no temp ref is left");
 }

+// The .lock files under a repository's refs.
+function locks(repo) {
+  const found = [];
+  const walk = (d) => {
+    for (const name of readdirSync(d)) {
+      const p = join(d, name);
+      if (lstatSync(p).isDirectory()) walk(p);
+      else if (name.endsWith(".lock")) found.push(p.slice(repo.length + 1));
+    }
+  };
+  walk(repo);
+  return found;
+}
+
+// A `git` first in PATH that runs `body` (bash) with REAL as the real git,
+// and the env that puts it there.
+const REAL_GIT = spawnSync("sh", ["-c", "command -v git"], { encoding: "utf8" }).stdout.trim();
+function wrapGit(w, body) {
+  const bin = join(w.dir, "bin");
+  mkdirSync(bin);
+  writeFileSync(join(bin, "git"), `#!/bin/bash
+REAL='${REAL_GIT}'
+${body}
+`);
+  chmodSync(join(bin, "git"), 0o755);
+  return { ...process.env, PATH: `${bin}:${process.env.PATH}` };
+}
+
 async function lands(w, fn) {
  const root = fingerprint(w.root);
  const r = await fn();
@@ -236,6 +264,85 @@ test("a record that throws moves the work ref back, or removes a new one", async
  assert.equal(w.refs(), `refs/heads/main ${w.first}\n${REF} ${c1}`);
 });

+test("a work ref that can't go back refuses rollback-failed, naming the ref and the commit", async (t) => {
+  const w = world(t);
+  git(w.clone, "checkout", "-q", "-b", "feat");
+  const c1 = commit(w.clone, "two");
+  // record() fails after taking the ref's lock, so moving the ref back fails.
+  const lock = join(w.repo, `${REF}.lock`);
+  const stuck = async () => {
+    writeFileSync(lock, "");
+    throw new Error("not recorded");
+  };
+  const message = (c) => (e) => e instanceof WorkingError && e.code === "rollback-failed" && e.message.includes(`${REF} still names ${c}`) && /not recorded/.test(e.message);
+  // A new ref that can't be removed.
+  await assert.rejects(w.go(w.bundle(), { record: stuck }), message(c1));
+  assert.equal(w.refs(), `refs/heads/main ${w.first}\n${REF} ${c1}`);
+  rmSync(lock);
+  // A ref that can't go back to the one before.
+  git(w.clone, "update-ref", "refs/remotes/working/work/vikunja-1-2", c1);
+  const c2 = commit(w.clone, "three");
+  await assert.rejects(w.go(w.bundle(), { record: stuck }), message(c2));
+  assert.equal(w.refs(), `refs/heads/main ${w.first}\n${REF} ${c2}`);
+  rmSync(lock);
+  assert.deepEqual(locks(w.repo), []);
+});
+
+test("a root that can't be read refuses root-unreadable", async (t) => {
+  const w = world(t);
+  git(w.clone, "checkout", "-q", "-b", "feat");
+  commit(w.clone, "two");
+  await refuses(w, () => w.go(w.bundle(), { root: join(w.dir, "no-root") }), "root-unreadable");
+});
+
+test("a git past its time gets SIGTERM, leaves no ref lock, and the next push lands", async (t) => {
+  const w = world(t);
+  git(w.clone, "checkout", "-q", "-b", "feat");
+  const c1 = commit(w.clone, "two");
+  await w.go(w.bundle());
+  git(w.clone, "update-ref", "refs/remotes/working/work/vikunja-1-2", c1);
+  // The move of the work ref holds the ref's lock in a reference-transaction
+  // hook until the lock is gone, so the git is still holding it when its time
+  // is up. Only that one fetch gets the hook.
+  const hooks = join(w.dir, "hooks");
+  mkdirSync(hooks);
+  const lock = join(w.repo, `${REF}.lock`);
+  writeFileSync(join(hooks, "reference-transaction"), `#!/bin/bash\n[ "$1" = prepared ] || exit 0\nfor i in $(seq 200); do [ -e '${lock}' ] || exit 0; sleep 0.05; done\n`);
+  chmodSync(join(hooks, "reference-transaction"), 0o755);
+  const env = wrapGit(
+    w,
+    `self=0
+for a in "$@"; do case "$a" in refs/mosaic/incoming/*:refs/heads/work/*) self=1;; esac; done
+args=()
+for a in "$@"; do
+  [ $self = 1 ] && [ "$a" = core.hooksPath=/dev/null ] && a='core.hooksPath=${hooks}'
+  args+=("$a")
+done
+exec "$REAL" "\${args[@]}"`,
+  );
+  commit(w.clone, "three");
+  const limits = { timeoutMs: 2000, graceMs: 5000 };
+  await assert.rejects(w.go(w.bundle(), { env, limits }), (e) => e instanceof WorkingError && e.code === "push-failed" && /killed by SIGTERM/.test(e.message));
+  assert.deepEqual(locks(w.repo), []);
+  assert.equal(w.refs(), `refs/heads/main ${w.first}\n${REF} ${c1}`);
+  const c3 = commit(w.clone, "four");
+  const r = await lands(w, () => w.go(w.bundle()));
+  assert.deepEqual(r, { ref: "work/vikunja-1-2", commit: c3, previous: c1 });
+});
+
+test("a git that ignores SIGTERM gets SIGKILL once the grace period is over", { timeout: 20000 }, async (t) => {
+  const w = world(t);
+  git(w.clone, "checkout", "-q", "-b", "feat");
+  commit(w.clone, "two");
+  // The fetch of main from the root never ends and ignores SIGTERM.
+  const env = wrapGit(
+    w,
+    `for a in "$@"; do [ "$a" = +HEAD:refs/heads/main ] && { trap '' TERM; while :; do sleep 0.05; done; }; done
+exec "$REAL" "$@"`,
+  );
+  await assert.rejects(w.go(w.bundle(), { env, limits: { timeoutMs: 500, graceMs: 500 } }), (e) => e instanceof WorkingError && e.code === "root-unreadable" && /killed by SIGKILL/.test(e.message));
+});
+
 test("a working repository path that isn't a directory refuses", async (t) => {
  const w = world(t);
  git(w.clone, "checkout", "-q", "-b", "feat");