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