Files
stack/packages/cli/tests/git-socket.test.mjs
T
jason.woltjeandClaude Opus 5.5 664199f794 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

551 lines
29 KiB
JavaScript

// Row 93 (#1587): the host's git socket inside a real bus host, driven by the
// runner's own git tools (packages/harness/src/git-tools.mjs) over real
// clones. A coder pushes, asks for a review, the reviewer fetches the commit
// and answers, all as the two roles' launches. No sandbox: the socket and the
// tools are what is under test, and each test checks that the project root is
// unchanged and that no bundle file is left in bus-host/ or either tmp.
import { test } from "node:test";
import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { createHash, randomBytes } from "node:crypto";
import { connect } from "node:net";
import { chmodSync, existsSync, lstatSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, readlinkSync, renameSync, rmSync, statSync, symlinkSync, writeFileSync } from "node:fs";
import { 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";
import { gitEnv } from "../../seat/src/checkout.mjs";
import { gitSocketPath } from "../../seat/src/session.mjs";
import { workingRepo } from "../../seat/src/working.mjs";
import { fetchReview, pushWorking } from "../../harness/src/git-tools.mjs";
import { bootConfig, loadSystem, rolesDir } from "../src/config.mjs";
import { createGitSocket, send } from "../src/git-socket.mjs";
import { startHost, startTimeOf } from "../src/host.mjs";
import { fixture } from "./helpers.mjs";
const TASK = "vikunja:1/2";
function gitIn(cwd, ...args) {
const r = spawnSync("git", ["-c", "user.name=test", "-c", "[email protected]", "-C", cwd, ...args], { env: gitEnv(), encoding: "utf8" });
assert.equal(r.status, 0, r.stderr);
return r.stdout.trim();
}
function commit(dir, text, file = "file.txt") {
writeFileSync(join(dir, file), `${text}\n`);
gitIn(dir, "add", file);
gitIn(dir, "commit", "-q", "-m", text);
return gitIn(dir, "rev-parse", "HEAD");
}
function fingerprint(dir) {
const h = createHash("sha256");
const walk = (d) => {
for (const name of readdirSync(d).sort()) {
const p = join(d, name);
const st = lstatSync(p);
h.update(`${p}\0${st.mode}\0`);
if (st.isDirectory()) walk(p);
else if (st.isSymbolicLink()) h.update(readlinkSync(p));
else if (st.isFile()) h.update(readFileSync(p));
}
};
walk(dir);
return h.digest("hex");
}
const bundles = (dir) => (existsSync(dir) ? readdirSync(dir).filter((n) => n.endsWith(".bundle")) : []);
// 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.
// 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-"));
const down = [];
t.after(async () => {
try {
for (const fn of down) await fn();
} finally {
rmSync(root, { recursive: true, force: true });
}
});
const f = fixture(root, "acme");
mkdirSync(f.dataRoot, { mode: 0o700 });
const project = f.doc.projects.stack.root;
gitIn(project, "init", "-q", "-b", "main");
const main = commit(project, "one");
const before = fingerprint(project);
const system = loadSystem({ env: f.env });
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: 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 = {};
for (const role of ["coder", "reviewer", "pm", "cto"]) {
const { cap } = await host.bindLaunch({ business: "acme", role, run: `${role}-test`, harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) });
const client = new Client({ path: host.path, cap });
await client.call("role.claim");
const workspace = join(root, "ws", role);
const tmpDir = join(root, "tmp", role);
mkdirSync(workspace, { recursive: true });
mkdirSync(tmpDir, { recursive: true });
gitIn(workspace, "clone", "-q", project, "stack");
roles[role] = { cap, client, workspace, tmp: tmpDir, clone: join(workspace, "stack") };
}
// Checked after every test: the root as it was, and no bundle anywhere.
down.push(() => {
assert.equal(fingerprint(project), before, "the project root is unchanged");
for (const dir of [join(f.dataRoot, "bus-host"), ...Object.values(roles).map((r) => r.tmp)]) assert.deepEqual(bundles(dir), [], `no bundle left in ${dir}`);
});
const push = (role, args = {}) => pushWorking({ path: sock.path, cap: roles[role].cap, workspace: roles[role].workspace, tmp: roles[role].tmp, task_ref: TASK, project: "stack", branch: "feat", ...args });
const fetch = (role, request, args = {}) => fetchReview({ path: sock.path, cap: roles[role].cap, workspace: roles[role].workspace, tmp: roles[role].tmp, task_ref: TASK, request, ...args });
const repo = workingRepo(f.dataRoot, "acme", "stack");
const refs = () => (existsSync(repo) ? gitIn(root, `--git-dir=${repo}`, "for-each-ref", "--format=%(refname) %(objectname)") : "");
const trail = async (kind, subject = TASK) => (await roles.cto.client.call("trail", { subject })).filter((r) => r.table === "events" && r.kind === kind);
return { root, f, project, main, host, sock, roles, push, fetch, repo, refs, trail, logs };
}
const rejects = (p, code) => assert.rejects(p, (e) => e.code === code || assert.fail(`${e.code}: ${e.message}, wanted ${code}`));
// One raw exchange on the git socket.
function raw(path, text) {
return new Promise((resolve, reject) => {
const s = connect(path);
let buf = "";
s.on("connect", () => s.write(text));
s.on("data", (b) => (buf += b));
s.on("end", () => resolve(JSON.parse(buf.split("\n")[0])));
s.on("error", reject);
});
}
// 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);
assert.equal(w.sock.path, gitSocketPath(w.f.dataRoot));
const { coder, reviewer } = w.roles;
gitIn(coder.clone, "checkout", "-q", "-b", "feat");
const c1 = commit(coder.clone, "two");
assert.deepEqual(await w.push("coder"), { ref: "work/vikunja-1-2", commit: c1, previous: null });
assert.equal(gitIn(coder.clone, "rev-parse", "refs/remotes/working/work/vikunja-1-2"), c1);
assert.equal(w.refs(), `refs/heads/main ${w.main}\nrefs/heads/work/vikunja-1-2 ${c1}`);
// Nothing new: the clone remembers the push.
await rejects(w.push("coder"), "nothing-to-push");
const req = await coder.client.call("review.request", { task_ref: TASK, commit: c1, to: "reviewer" });
const got = await w.fetch("reviewer", req.request);
assert.deepEqual(got, { project: "stack", ref: "working/work/vikunja-1-2", commit: c1, clone: reviewer.clone });
assert.equal(gitIn(reviewer.clone, "rev-parse", "refs/remotes/working/work/vikunja-1-2"), c1);
assert.equal(gitIn(reviewer.clone, "show", "-s", "--format=%s", c1), "two");
const v = await reviewer.client.call("review.verdict", { task_ref: TASK, request: req.request, verdict: "changes", body: "one more" });
assert.equal(v.commit, c1);
// The coder answers with a fast-forward; the bundle carries only the new commit.
const c2 = commit(coder.clone, "three");
assert.deepEqual(await w.push("coder"), { ref: "work/vikunja-1-2", commit: c2, previous: c1 });
const req2 = await coder.client.call("review.request", { task_ref: TASK, commit: c2, to: "reviewer" });
assert.equal((await w.fetch("reviewer", req2.request)).commit, c2);
const ok = await reviewer.client.call("review.verdict", { task_ref: TASK, request: req2.request, verdict: "approve", body: "good" });
// A push is the action.allowed for git.push.working that names its commit.
const pushes = (await w.trail("action.allowed")).filter((e) => e.body.action === "git.push.working");
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, []);
});
test("refusals: no action, non-fast-forward, an unknown commit or request, and a push for another task moves only that task's ref", async (t) => {
const w = await setup(t);
const { coder, reviewer, pm } = w.roles;
gitIn(coder.clone, "checkout", "-q", "-b", "feat");
const c1 = commit(coder.clone, "two");
await w.push("coder");
const refs = w.refs();
// The reviewer and pm hold no git.push.working; the broker refuses before the host takes a byte.
for (const role of ["reviewer", "pm"]) {
gitIn(w.roles[role].clone, "checkout", "-q", "-b", "feat");
commit(w.roles[role].clone, `${role} work`);
await rejects(w.push(role), "decision-required");
}
assert.equal(w.refs(), refs);
// A rewritten branch.
gitIn(coder.clone, "update-ref", "-d", "refs/remotes/working/work/vikunja-1-2");
gitIn(coder.clone, "reset", "-q", "--hard", w.main);
commit(coder.clone, "other two");
await rejects(w.push("coder"), "non-fast-forward");
assert.equal(w.refs(), refs);
// The same branch for another task lands on that task's ref and leaves this one.
const other = await w.push("coder", { task_ref: "vikunja:1/3" });
assert.equal(other.ref, "work/vikunja-1-3");
assert.match(w.refs(), new RegExp(`refs/heads/work/vikunja-1-2 ${c1}`));
await rejects(coder.client.call("review.request", { task_ref: TASK, commit: "e".repeat(40), to: "reviewer" }), "unknown-commit");
const req = await coder.client.call("review.request", { task_ref: TASK, commit: c1, to: "reviewer" });
await rejects(w.fetch("reviewer", "no-such-request"), "request-not-found");
await rejects(w.fetch("reviewer", req.request, { task_ref: "vikunja:1/3" }), "request-not-found");
// The coder and pm hold no review.verdict.
await rejects(w.fetch("coder", req.request), "not-authorized");
await rejects(w.fetch("pm", req.request), "not-authorized");
assert.equal(existsSync(join(reviewer.clone, ".git", "refs", "remotes", "working")), false);
// Each refusal is an action.refused event on the task, from the role that tried.
const refused = (await w.trail("action.refused")).map((e) => `${e.actor_role} ${e.body.code}`);
for (const r of ["reviewer decision-required", "pm decision-required", "coder non-fast-forward", "coder unknown-commit", "reviewer request-not-found", "coder not-authorized", "pm not-authorized"])
assert.ok(refused.includes(r), `${r} in ${refused.join(", ")}`);
});
test("the size cap refuses a push before any byte", async (t) => {
const w = await setup(t, { bundleMax: 64 });
gitIn(w.roles.coder.clone, "checkout", "-q", "-b", "feat");
commit(w.roles.coder.clone, "two");
await rejects(w.push("coder"), "bundle-too-large");
assert.equal(w.refs(), "");
assert.deepEqual((await w.trail("action.allowed")).filter((e) => e.body.action === "git.push.working"), []);
});
test("a fetch over the cap refuses before any byte reaches the reviewer", async (t) => {
const w = await setup(t, { bundleMax: 1024 * 1024 });
const { coder } = w.roles;
gitIn(coder.clone, "checkout", "-q", "-b", "feat");
// Two pushes of 600 KiB of noise each pass a 1 MiB cap; the reviewer has
// neither, so its bundle carries both and is over.
writeFileSync(join(coder.clone, "big"), randomBytes(600 * 1024));
gitIn(coder.clone, "add", "big");
gitIn(coder.clone, "commit", "-q", "-m", "big");
const c = (await w.push("coder")).commit;
writeFileSync(join(coder.clone, "big"), randomBytes(600 * 1024));
gitIn(coder.clone, "commit", "-q", "-am", "bigger");
const c2 = (await w.push("coder")).commit;
assert.notEqual(c, c2);
const req = await coder.client.call("review.request", { task_ref: TASK, commit: c2, to: "reviewer" });
await rejects(w.fetch("reviewer", req.request), "bundle-too-large");
assert.equal(existsSync(join(w.roles.reviewer.clone, ".git", "refs", "remotes", "working")), false);
});
test("the socket refuses a request with no capability, an unknown one, and a bad request line", async (t) => {
const w = await setup(t);
assert.deepEqual(await raw(w.sock.path, "not json\n"), { ok: false, error: "invalid-request" });
assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ op: "push" })}\n`), { ok: false, error: "unauthenticated" });
assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ cap: "f".repeat(64), op: "push" })}\n`), { ok: false, error: "unauthenticated" });
const cap = w.roles.coder.cap;
assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ cap, op: "merge" })}\n`), { ok: false, error: "invalid-request" });
assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ cap, op: "push", task_ref: TASK, project: "nope", branch: "feat", size: 10 })}\n`), { ok: false, error: "invalid-request" });
assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ cap, op: "push", task_ref: "gitea:1", project: "stack", branch: "feat", size: 10 })}\n`), { ok: false, error: "invalid-request" });
// Bytes before "send" are refused, not read.
assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ cap, op: "push", task_ref: TASK, project: "stack", branch: "feat", size: 10 })}\n0123456789`), { ok: false, error: "invalid-request" });
assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ cap, op: "fetch", task_ref: TASK, request: "x", have: ["nope"] })}\n`), { ok: false, error: "invalid-request" });
// A push whose bytes have a bundle's header and then no ref line git can read.
const junk = await new Promise((resolve, reject) => {
const s = connect(w.sock.path);
let buf = "";
s.on("connect", () => s.write(`${JSON.stringify({ cap, op: "push", task_ref: TASK, project: "stack", branch: "feat", size: 20 })}\n`));
s.on("data", (b) => {
buf += b;
if (buf === `${JSON.stringify({ ok: true, send: 20 })}\n`) s.write("# v2 git bundle\nzzz\n");
});
s.on("end", () => resolve(buf.trim().split("\n").pop()));
s.on("error", reject);
});
assert.deepEqual(JSON.parse(junk), { ok: false, error: "invalid-bundle" });
assert.equal(w.refs(), `refs/heads/main ${w.main}`);
});
test("a rigged clone runs nothing: hooks, fsmonitor, a filter, a gitlink and a symlinked .git/config leave no marker", async (t) => {
const w = await setup(t);
const { coder, reviewer } = w.roles;
const marks = join(w.root, "marks");
mkdirSync(marks);
gitIn(coder.clone, "checkout", "-q", "-b", "feat");
commit(coder.clone, "two");
// A gitlink whose .gitmodules would run a command if anything cloned it.
// (An `update = !cmd` line fsck refuses outright; the end of the test pushes one.)
writeFileSync(join(coder.clone, ".gitmodules"), `[submodule "sub"]\n\tpath = sub\n\turl = ext::sh -c touch% ${marks}/submodule\n`);
gitIn(coder.clone, "add", ".gitmodules");
gitIn(coder.clone, "update-index", "--add", "--cacheinfo", `160000,${w.main},sub`);
gitIn(coder.clone, "commit", "-q", "-m", "gitlink");
writeFileSync(join(coder.clone, ".gitattributes"), "* filter=evil\n");
gitIn(coder.clone, "add", ".gitattributes");
gitIn(coder.clone, "commit", "-q", "-m", "attributes");
const tipCommit = gitIn(coder.clone, "rev-parse", "HEAD");
// On a second branch, a .gitmodules with a command as its update setting.
gitIn(coder.clone, "checkout", "-q", "-b", "bad");
writeFileSync(join(coder.clone, ".gitmodules"), `[submodule "sub"]\n\tpath = sub\n\turl = ./sub\n\tupdate = !touch ${marks}/update\n`);
gitIn(coder.clone, "add", ".gitmodules");
gitIn(coder.clone, "commit", "-q", "-m", "update command");
gitIn(coder.clone, "checkout", "-q", "feat");
// Rigged only now, so the test's own git calls above ran none of it.
const rig = (clone, name) => {
const hooks = join(w.root, `hooks-${name}`);
mkdirSync(hooks);
for (const hook of ["reference-transaction", "pre-push", "post-checkout", "post-merge", "post-rewrite", "pre-auto-gc", "fsmonitor-watchman"]) {
writeFileSync(join(hooks, hook), `#!/bin/sh\ntouch '${marks}/${name}-${hook}'\nexit 0\n`);
chmodSync(join(hooks, hook), 0o755);
mkdirSync(join(clone, ".git", "hooks"), { recursive: true });
writeFileSync(join(clone, ".git", "hooks", hook), `#!/bin/sh\ntouch '${marks}/${name}-local-${hook}'\nexit 0\n`);
chmodSync(join(clone, ".git", "hooks", hook), 0o755);
}
const config = join(w.root, `config-${name}`);
writeFileSync(
config,
`${readFileSync(join(clone, ".git", "config"), "utf8")}[core]\n\thooksPath = ${hooks}\n\tfsmonitor = ${hooks}/fsmonitor-watchman\n[filter "evil"]\n\tclean = touch ${marks}/${name}-clean\n\tsmudge = touch ${marks}/${name}-smudge\n\tprocess = touch ${marks}/${name}-process\n[uploadpack]\n\tpackObjectsHook = touch ${marks}/${name}-pack\n[gc]\n\tauto = 1\n`,
);
renameSync(join(clone, ".git", "config"), join(w.root, `config-${name}.orig`));
symlinkSync(config, join(clone, ".git", "config"));
};
rig(coder.clone, "coder");
rig(reviewer.clone, "reviewer");
const pushed = await w.push("coder");
assert.equal(pushed.commit, tipCommit);
const req = await coder.client.call("review.request", { task_ref: TASK, commit: pushed.commit, to: "reviewer" });
assert.equal((await w.fetch("reviewer", req.request)).commit, pushed.commit);
assert.deepEqual(readdirSync(marks), []);
// The gitlink came across as a tree entry and nothing more.
assert.match(gitIn(reviewer.clone, "-c", "core.hooksPath=/dev/null", "ls-tree", pushed.commit, "sub"), /^160000 commit /);
assert.equal(existsSync(join(reviewer.clone, "sub", ".git")), false);
assert.deepEqual(readdirSync(marks), []);
// The update command: the host's fsck refuses the bundle and the work ref stays.
const landed = w.refs();
await rejects(w.push("coder", { branch: "bad" }), "invalid-bundle");
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);
});