stack/packages/cli/src/git-socket.mjs

// The host's git socket (#1587), `<dataRoot>/bus-host/git.sock` (0600). A
// coder's runner pushes its work to the business's working repository
// through it, a reviewer's runner fetches a commit it was asked to review,
// and a run holding an approved merge decision lands its task on the
// repository's main (row 137, #1631). The bundle is the only thing that
// crosses: the host never runs git in a session's clone and never opens a
// path in a session's workspace (packages/seat/src/working.mjs says why). The
// sandbox binds the socket only for a role that holds git.push.working or
// review.verdict, or a merge decision a landing could use (landHeld).
//
// Every connection starts with one JSON line of at most 4096 bytes.
//
//   push   {"cap", "op": "push", "task_ref", "project", "branch", "size"}
//          The host answers {"ok": true, "send": <size>} or a refusal. On
//          "send" the runner writes exactly <size> bytes of a v2 git bundle,
//          and the host answers {"ok": true, "result": {ref, commit,
//          previous}} or a refusal.
//   fetch  {"cap", "op": "fetch", "task_ref", "request", "have": [<oid>...]}
//          The host answers {"ok": true, "result": {project, ref, commit,
//          size}} followed by <size> bytes of a bundle whose one ref is
//          refs/mosaic/review, or a refusal.
//   land   {"cap", "op": "land", "task_ref", "decision"}
//          The host answers {"ok": true, "result": {landing: "started",
//          commit, base, reviewed, patch_id}} once everything that can
//          refuse before the gate has passed, or a refusal. `commit` is the
//          squash that will land. The gate runs after the answer, and its
//          outcome is the broker's: recordLand or landRefused, each on the
//          task's trail and told to the role as a RESULT message.
//
// A refusal is {"ok": false, "error": "<code>"}, and the connection ends.
// The host checks, in order: identity (a role of this business), the request
// (task ref, project, branch, size, at most 32 MiB: bundle-too-large), then
// the broker: git.push.working for a push, the review.requested for a fetch
// (packages/bus/src/broker.mjs). Only then does it take any bundle bytes.
// A push is recorded, as the action.allowed for git.push.working with its
// commit, under the repository's lock once the ref moved; when the broker
// refuses the record, the ref goes back.
// 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. 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.
//
// A landing runs land-work's steps (scripts/land-work.sh): the broker's
// checkLand, then prepareLand (working.mjs) under the repository's lock, the
// gate check (gate-missing), the gate in the squash's worktree with the
// root's node_modules linked, and finishLand, which moves main and records.
// One landing runs at a time; another refuses land-busy. Its logs stay in
// bus-host/land/<id>/, and its worktree and Docker network go on every path.
// close() stops a running gate, and the landing is refused host-closing.

import { spawn } from "node:child_process";
import { randomBytes } from "node:crypto";
import { closeSync, createReadStream, existsSync, lstatSync, mkdirSync, openSync, rmSync, statSync, symlinkSync, writeSync, chmodSync } from "node:fs";
import { createServer } from "node:net";
import { join } from "node:path";
import { gitSocketPath } from "../../seat/src/session.mjs";
import { gateEnv, isMosaicStack, rootHits, runGate } from "../../seat/src/landing.mjs";
import { BUNDLE_MAX, dropLand, finishLand, FSCK, HAVE_MAX, prepareLand, push, review, validBranch, workRef, WorkingError } from "../../seat/src/working.mjs";
import { CliError } from "./errors.mjs";

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;
// 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})?$/;

// A refusal the host decided; it goes to the broker's refuse().
class Refusal extends Error {
  constructor(code, message = code) {
    super(message);
    this.code = code;
  }
}
// A refusal the broker decided and recorded itself.
class Recorded extends Error {
  constructor(code) {
    super(code);
    this.code = code;
  }
}

// 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);
      socket.off("timeout", onGone);
      socket.setTimeout(0);
      socket.pause();
      err ? reject(err) : resolve(value);
    };
    const onGone = () => done(new Refusal("invalid-request", "no request line"));
    const onData = (chunk) => {
      buf = Buffer.concat([buf, chunk]);
      const nl = buf.indexOf(0x0a);
      if (nl < 0 ? buf.length > HEADER_MAX : nl > HEADER_MAX) return done(new Refusal("invalid-request", "request line too long"));
      if (nl < 0) return;
      let req;
      try {
        req = JSON.parse(buf.subarray(0, nl).toString("utf8"));
      } catch {
        return done(new Refusal("invalid-request", "request line isn't JSON"));
      }
      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);
    socket.setTimeout(HEADER_TIMEOUT_MS);
    socket.once("timeout", onGone);
  });
}

// Exactly `size` bytes from the socket into a new file (0600), within
// `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;
    let settled = false;
    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", 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 ${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"));
      try {
        writeSync(fd, chunk);
      } catch (e) {
        return done(e);
      }
      got += chunk.length;
      if (got === size) done();
    };
    socket.on("data", onData);
    socket.once("end", onShort);
    socket.once("close", onShort);
    socket.setTimeout(idleMs);
    socket.once("timeout", onIdle);
    socket.resume();
  });
}

// 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, 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;
  // 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();
}

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

// Removes the Docker network a gate's Compose project left. The project is
// named after the worktree, so the network is <name>_default.
function dropNetwork(name, env) {
  return new Promise((resolve) => {
    const child = spawn("docker", ["network", "rm", name], { env, stdio: "ignore" });
    child.on("error", () => resolve());
    child.on("close", () => resolve());
  });
}
export const LAND_GATE = { classify: isMosaicStack, run: runGate, dropNetwork };

// How the decision was resolved, by the resolution's via: the human through the CLI, or the
// arbiter through the broker (row 137, REQ-DEC-2). readLanding takes no other.
const RESOLVED_VIA = { cli: "via the CLI", broker: "as its arbiter, via the broker" };

// The squash's message: land-work's, applied under the git socket.
function landMessage(f, { old, base, patch_id }) {
  return `${f.subject}

Squash of ${f.work_ref} (${f.task_ref}) at ${f.commit} onto main ${old},
the reviewed diff ${base}..${f.commit}. Resolved ${f.choice} by ${f.by} ${RESOLVED_VIA[f.via] ?? `via ${f.via}`} at ${f.resolved_at}.

Decision: ${f.decision}
Verdict: ${f.verdict}
Work-ref: ${f.work_ref}
Reviewed-commit: ${f.commit}
Base: main ${old}
Patch-id: ${patch_id}
Gate: full, every package suite and every scripts/test-*.sh
Applied-under: git socket land (bus host)
`;
}

// host: startHost()'s handle; business: loadBusiness() output. bundleMax,
// bodyDeadline, bodyIdle, maxActive, fsck and gate (LAND_GATE's shape) 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, gate = LAND_GATE }) {
  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();
  // The running landing, { abort, done }, or null.
  let landing = null;

  const op = async (message) => {
    try {
      return await host.op(message);
    } catch (e) {
      throw new Recorded(e.code ?? "not-authorized");
    }
  };
  const project = (id) => (typeof id === "string" && Object.hasOwn(b.projects, id) ? b.projects[id] : null);

  async function doPush(socket, cap, req, rest, file) {
    const { task_ref, branch, size } = req;
    const p = project(req.project);
    if (!p || !validBranch(branch) || !Number.isSafeInteger(size) || size < 1) throw new Refusal("invalid-request");
    workRef(task_ref);
    if (size > bundleMax) throw new Refusal("bundle-too-large", `${size} bytes, over ${bundleMax}`);
    // The runner waits for "send"; bytes before it break the protocol.
    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), 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 } }),
    });
    socket.end(`${JSON.stringify({ ok: true, result })}\n`);
  }

  async function doFetch(socket, cap, req, file) {
    const { task_ref, request, have = [] } = req;
    if (typeof request !== "string" || !Array.isArray(have) || have.length > HAVE_MAX || have.some((h) => typeof h !== "string" || !OID.test(h))) throw new Refusal("invalid-request");
    workRef(task_ref);
    const r = await op({ op: "reviewFetch", cap, args: { task_ref, request } });
    const p = project(r.project);
    if (!p) throw new Refusal("unknown-project");
    const made = await review({ dataRoot, business: b.id, project: r.project, root: p.root, taskRef: task_ref, commit: r.commit, have, out: file, 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 send(socket, file);
  }

  async function doLand(socket, cap, req) {
    const { task_ref, decision } = req;
    if (typeof decision !== "string") throw new Refusal("invalid-request");
    workRef(task_ref);
    if (landing) throw new Refusal("land-busy", "another landing is running");
    // Taken before the first await, so two requests can't both pass.
    const slot = { abort: new AbortController(), done: null };
    landing = slot;
    try {
      await startLand(socket, cap, task_ref, decision, slot);
    } finally {
      if (!slot.done) landing = null;
    }
  }

  // From the broker's ticket on, each refusal is the landing's (landRefused):
  // the run's cap may be gone by the time the gate ends.
  async function startLand(socket, cap, task_ref, decision, slot) {
    const facts = await op({ op: "checkLand", cap, args: { task_ref, decision } });
    const { ticket } = facts;
    const id = `land-${randomBytes(6).toString("hex")}`;
    const logs = join(dir, "land", id);
    const p = project(facts.project);
    const where = { dataRoot, business: b.id, project: facts.project, env };
    const refused = async (e, notify) => {
      const code = e instanceof Refusal || e instanceof Recorded || e instanceof WorkingError ? e.code : "git-failed";
      if (code === "git-failed" || code === "rollback-failed") log(`git socket: ${id}: ${code}: ${e.message}`);
      // The host's own faults stay in its log; the role gets the code.
      const detail = notify && code !== "git-failed" && e.message !== code ? e.message.slice(0, 1000) : null;
      await host.op({ op: "landRefused", ticket, args: { code, detail, notify } }).catch((x) => log(`git socket: ${id}: refusal ${code} not recorded: ${x.code ?? x.message}`));
      return code;
    };
    let prepared = null;
    try {
      if (!p) throw new Refusal("unknown-project");
      mkdirSync(logs, { recursive: true, mode: 0o700 });
      prepared = await prepareLand({ ...where, root: p.root, taskRef: task_ref, commit: facts.commit, dir: logs, name: id, rootHits, message: (m) => landMessage(facts, m) });
      if (!(await gate.classify(p.root, env))) throw new Refusal("gate-missing", `${p.root} isn't a mosaic-stack checkout with node_modules`);
      symlinkSync(join(p.root, "node_modules"), join(prepared.worktree, "node_modules"));
    } catch (e) {
      if (prepared) await dropLand({ ...where, worktree: prepared.worktree }).catch((x) => log(`git socket: ${id}: ${x.message}`));
      throw new Recorded(await refused(e, false));
    }
    const { commit, base, reviewed, patch_id } = prepared;
    socket.end(`${JSON.stringify({ ok: true, result: { landing: "started", commit, base, reviewed, patch_id } })}\n`);
    slot.done = (async () => {
      try {
        const r = await gate.run({ worktree: prepared.worktree, logs, env: gateEnv(env), signal: slot.abort.signal });
        if (r.aborted || slot.abort.signal.aborted) throw new Refusal("host-closing", "the host closed during the gate; nothing moved");
        if (!r.ok) throw new Refusal("gate-failed", `${r.step} failed, see ${r.log}`);
        await finishLand({ ...where, land: prepared, record: (m) => op({ op: "recordLand", ticket, land: { ...m, reviewed, base, patch_id } }) });
      } catch (e) {
        await refused(e, true);
      } finally {
        await dropLand({ ...where, worktree: prepared.worktree }).catch((x) => log(`git socket: ${id}: ${x.message}`));
        await gate.dropNetwork(`${id}_default`, gateEnv(env));
        landing = null;
      }
    })();
  }

  async function serve(socket) {
    let cap = null;
    let subject = null;
    let file = null;
    const fail = (code) => {
      if (!socket.destroyed) socket.end(`${JSON.stringify({ ok: false, error: code })}\n`);
    };
    try {
      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;
      try {
        who = await host.op({ op: "identity", cap: req.cap });
      } catch (e) {
        return fail(e.code ?? "unauthenticated");
      }
      cap = req.cap;
      if (typeof req.task_ref === "string") subject = req.task_ref;
      if (!who.role || who.business !== b.id) throw new Refusal("not-a-role");
      file = join(dir, `${randomBytes(8).toString("hex")}.bundle`);
      if (req.op === "push") await doPush(socket, cap, req, rest, file);
      else if (req.op === "fetch") await doFetch(socket, cap, req, file);
      else if (req.op === "land") await doLand(socket, cap, req);
      else throw new Refusal("invalid-request", "op is push, fetch or land");
    } 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" || code === "rollback-failed") log(`git socket: ${code}: ${e.message}`);
      if (cap) await host.op({ op: "refuse", cap, code, subject }).catch(() => {});
      fail(code);
    } finally {
      if (file) rmSync(file, { force: true });
    }
  }

  async function listen() {
    if (existsSync(path)) {
      if (!lstatSync(path).isSocket()) throw new CliError(`${path} exists and isn't a socket`, 3);
      rmSync(path);
    }
    server = createServer((socket) => {
      sockets.add(socket);
      socket.once("close", () => sockets.delete(socket));
      socket.on("error", () => {});
      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));
    });
    // bus-host/ is 0700, so the moment before the chmod exposes nothing.
    await new Promise((resolve, reject) => {
      server.once("error", reject);
      server.listen(path, resolve);
    });
    chmodSync(path, 0o600);
  }

  // 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.
  // A landing's gate is stopped, and its refusal recorded the same way.
  async function close() {
    closed = true;
    if (!server) return;
    const stopped = new Promise((r) => server.close(r));
    const told = new Set(waiting.keys());
    for (const stop of [...waiting.values()]) stop();
    for (const socket of sockets) if (!told.has(socket)) socket.destroy();
    landing?.abort.abort();
    await Promise.allSettled([...active]);
    await landing?.done;
    // The refusal is a few bytes; once it is flushed, or the peer is gone,
    // 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 });
  }

  return { path, listen, close };
}