jason.woltjeandClaude Opus 5.5
465bbdfbbc
Row 103, the row 100 follow-up. Built by Darkwing, packet
agents/darkwing/work/queue-103/r1/, candidate manifest
6e5ac8c0782bc595d7c0d711d457738ffb42f1ba5f38253912943dd55e55888a.
Sage approved on #1597, comment 28252; Dewey approved, comment 28260.
Host git runs through runGit: a detached group, SIGTERM at the limit,
SIGKILL after the grace, and one more SIGKILL to the group after a
signalled git closes. git-socket close() waits at most 2 s. startHost
claims bus-host/host.lock by hard link before the broker fork.
packages/cli/README.md and tests/host.test.mjs carry row 85's hunks
too; row 85 lands first.
Sage's gate rerun on 3a1e11b83 with rows 85, 102 and 103 applied,
Docker up: node suites bus 128, business 67, cli 165, control-board
124, conversation 182, discord 178, harness 73, ledger 78, mosaic 69,
queue 155, runs 43, seat 113, tasks 92, webui 108; scripts auth 15,
conductor 17, config 24, discord 66, extension-package 18,
foundation 44, queue 28, release 14, task 98. All pass, 0 fail.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
629 lines
32 KiB
JavaScript
629 lines
32 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 { open } from "node:fs/promises";
|
|
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 { CLOSE_WAIT_MS, createGitSocket, letGo, 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 bundle with no bytes for bodyIdle refuses body-timeout long before its deadline", async (t) => {
|
|
// The deadline is a minute away, so only the idle limit can end it.
|
|
const w = await setup(t, { bodyDeadline: () => 60000, bodyIdle: 300 });
|
|
const { coder } = w.roles;
|
|
const started = Date.now();
|
|
// Sends one byte of 1000 after "send", then nothing.
|
|
const stalled = new Promise((resolve, reject) => {
|
|
const s = connect(w.sock.path);
|
|
let buf = "";
|
|
s.on("connect", () => s.write(pushReq(coder.cap, 1000)));
|
|
s.on("data", (b) => {
|
|
const sent = buf === `${JSON.stringify({ ok: true, send: 1000 })}\n`;
|
|
buf += b;
|
|
if (!sent && buf === `${JSON.stringify({ ok: true, send: 1000 })}\n`) s.write("x");
|
|
});
|
|
s.on("close", () => 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();
|
|
});
|
|
const { last, ms } = await stalled;
|
|
assert.deepEqual(JSON.parse(last), { ok: false, error: "body-timeout" });
|
|
assert.ok(ms >= 300 && ms < 5000, `${ms} ms`);
|
|
assert.ok((await refusedOn(w)).includes("coder body-timeout"));
|
|
assert.equal(w.refs(), "");
|
|
});
|
|
|
|
// A FIFO as the bundle holds send()'s loop open on a read, so a test can end
|
|
// the connection between the last chunk and the loop's end: the arm after the
|
|
// loop. A plain close there was served; a failed write was fetch-aborted.
|
|
test("send: a close once the loop is past its last chunk is served, and a failed write there is fetch-aborted", async (t) => {
|
|
const dir = mkdtempSync(join(tmpdir(), "mosaic-send-"));
|
|
t.after(() => rmSync(dir, { recursive: true, force: true }));
|
|
const fifo = join(dir, "b.fifo");
|
|
assert.equal(spawnSync("mkfifo", [fifo]).status, 0);
|
|
|
|
// `fail` is how the stand-in socket ends: destroyed as it is, or by
|
|
// failing its held write as a socket whose reader left does.
|
|
const run = async (fail) => {
|
|
let pending = null;
|
|
const sock = new Duplex({ read() {}, write: (c, e, cb) => (pending = cb), final: (cb) => cb() });
|
|
sock.on("error", () => {});
|
|
const sent = send(sock, fifo);
|
|
// Settled before the FIFO ends, so neither outcome is unhandled.
|
|
const outcome = sent.then(() => "served", (e) => e.code);
|
|
const writer = await open(fifo, "w");
|
|
await writer.write("ten bytes!");
|
|
await until(() => pending, "the write");
|
|
if (fail) pending(Object.assign(new Error("write EPIPE"), { code: "EPIPE" }));
|
|
else sock.destroy();
|
|
await until(() => sock.destroyed, "the socket's end");
|
|
await writer.close();
|
|
return outcome;
|
|
};
|
|
assert.equal(await run(false), "served");
|
|
assert.equal(await run(true), "fetch-aborted");
|
|
});
|
|
|
|
// A peer that never reads leaves the host's last write unfinished; this
|
|
// stand-in never finishes one.
|
|
test("letGo waits for an answer to leave for at most its cap, then destroys the socket", async (t) => {
|
|
const stuck = new Duplex({ read() {}, write() {}, final() {} });
|
|
stuck.on("error", () => {});
|
|
stuck.end("host-closing\n");
|
|
const quick = new Duplex({ read() {}, write: (c, e, cb) => cb(), final: (cb) => cb() });
|
|
quick.on("error", () => {});
|
|
quick.end("host-closing\n");
|
|
const started = Date.now();
|
|
let timer = null;
|
|
const hung = new Promise((r) => (timer = setTimeout(() => r("hung"), 3000)));
|
|
t.after(() => clearTimeout(timer));
|
|
assert.equal(await Promise.race([letGo([stuck, quick], 300).then(() => "let go"), hung]), "let go");
|
|
const ms = Date.now() - started;
|
|
assert.ok(ms >= 290 && ms < 3000, `${ms} ms`);
|
|
assert.ok(stuck.destroyed && quick.destroyed);
|
|
assert.ok(CLOSE_WAIT_MS > 0 && CLOSE_WAIT_MS <= 10000, `CLOSE_WAIT_MS is ${CLOSE_WAIT_MS}`);
|
|
});
|
|
|
|
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);
|
|
});
|