From ac430b5cfbc06dc69341d10e710ef684af1ce579 Mon Sep 17 00:00:00 2001 From: Jason Woltje Date: Tue, 13 Oct 2026 11:55:33 -0500 Subject: [PATCH] darkwing: row 100 round 1 packet, git socket hardening and refusal tests on 8e0fbca4 (#1594) Co-Authored-By: Claude Opus 5.5 --- agents/darkwing/work/queue-100/gate.sh | 21 + agents/darkwing/work/queue-100/mut.mjs | 6 + agents/darkwing/work/queue-100/mutants.sh | 55 + agents/darkwing/work/queue-100/r1/build.md | 116 ++ agents/darkwing/work/queue-100/r1/build.patch | 1051 +++++++++++++++++ .../queue-100/r1/candidate-manifest.sha256 | 8 + agents/darkwing/work/queue-100/run.py | 19 + 7 files changed, 1276 insertions(+) create mode 100644 agents/darkwing/work/queue-100/gate.sh create mode 100644 agents/darkwing/work/queue-100/mut.mjs create mode 100644 agents/darkwing/work/queue-100/mutants.sh create mode 100644 agents/darkwing/work/queue-100/r1/build.md create mode 100644 agents/darkwing/work/queue-100/r1/build.patch create mode 100644 agents/darkwing/work/queue-100/r1/candidate-manifest.sha256 create mode 100644 agents/darkwing/work/queue-100/run.py diff --git a/agents/darkwing/work/queue-100/gate.sh b/agents/darkwing/work/queue-100/gate.sh new file mode 100644 index 000000000..3cde3ecb2 --- /dev/null +++ b/agents/darkwing/work/queue-100/gate.sh @@ -0,0 +1,21 @@ +#!/usr/bin/env bash +# Row 100 gate: every package's own test script, from its directory, then +# every scripts/test-*.sh, sequential, full output kept. +cd ~/darkwing-scratch/r100/wt || exit 2 +W=$PWD +export TMPDIR=~/darkwing-scratch/r100/tmp +O=${GATE_OUT:-~/darkwing-scratch/r100/gate} +mkdir -p "$TMPDIR" "$O" +for d in packages/*/; do + p=$(basename "$d") + t=$(node -e 'const s=require(process.argv[1]).scripts?.test; if (s) console.log(s)' "$W/$d/package.json") + [ -n "$t" ] || { echo "$p: no test script"; continue; } + s=$(date -u +%T) + (cd "$W/$d" && bash -c "$t") > $O/node-$p.txt 2>&1; rc=$? + echo "$p rc=$rc $s-$(date -u +%T) $(grep -E '^ℹ (pass|fail|skipped)' $O/node-$p.txt | tr '\n' ' ') $(grep -cE '^✖|^not ok' $O/node-$p.txt) marks" +done +for f in scripts/test-*.sh; do + n=$(basename $f .sh); s=$(date -u +%T) + if [ $n = test-task ]; then DOCKER_HOST=unix:///nonexistent bash $f > $O/$n.txt 2>&1; else bash $f > $O/$n.txt 2>&1; fi; rc=$? + echo "$n rc=$rc $s-$(date -u +%T) $(tail -1 $O/$n.txt | cut -c1-120)" +done diff --git a/agents/darkwing/work/queue-100/mut.mjs b/agents/darkwing/work/queue-100/mut.mjs new file mode 100644 index 000000000..072bba00d --- /dev/null +++ b/agents/darkwing/work/queue-100/mut.mjs @@ -0,0 +1,6 @@ +// node mut.mjs : replace exactly one occurrence, else exit 3. +import { readFileSync, writeFileSync } from 'node:fs'; +const [f, a, b] = process.argv.slice(2); +const s = readFileSync(f, 'utf8'); const n = s.split(a).length - 1; +if (n !== 1) { console.error(`matches=${n}`); process.exit(3); } +writeFileSync(f, s.replace(a, () => b)); diff --git a/agents/darkwing/work/queue-100/mutants.sh b/agents/darkwing/work/queue-100/mutants.sh new file mode 100644 index 000000000..e5246f1f1 --- /dev/null +++ b/agents/darkwing/work/queue-100/mutants.sh @@ -0,0 +1,55 @@ +#!/usr/bin/env bash +# Row 100 (#1594) mutants. Run from the candidate worktree's root: +# LOG= bash mutants.sh > mutants.txt +# Each mutant edits one site with mut.mjs (exactly one match, else exit 3), +# runs the named test files (DRY=1 only checks the match) in their own process group (run.py, killed after +# 240 s) and restores the file. "killed" means the run failed; "survived" +# means it passed. +set -u +Q=$1 +run() { + local name=$1 file=$2 from=$3 to=$4; shift 4 + cp -- "$file" "$file.orig" + if ! node "$Q/mut.mjs" "$file" "$from" "$to"; then + echo "$name: NO MATCH"; mv -- "$file.orig" "$file"; return + fi + if [ -n "${DRY:-}" ]; then echo "$name: match"; mv -- "$file.orig" "$file"; return; fi + local log="${LOG:-.}/mut-$name.log" + if python3 "$Q/run.py" 240 "$log" node --test "$@"; then + echo "$name: survived" + else + echo "$name: killed ($(grep -m1 -o 'ℹ fail [0-9]*' "$log"; grep -m1 -o 'TIMEOUT.*' "$log"), $(grep -m1 '^✖' "$log" | cut -c1-90))" + fi + mv -- "$file.orig" "$file" +} +WK=packages/seat/src/working.mjs +GS=packages/cli/src/git-socket.mjs +GTL=packages/harness/src/git-tools.mjs +WT=packages/seat/tests/working.test.mjs +GT=packages/cli/tests/git-socket.test.mjs +HT=packages/harness/tests/git-tools.test.mjs +# Timeouts: SIGTERM first, then SIGKILL after the grace period. +run sigkill-first $WK $' child.kill("SIGTERM");\n kill = setTimeout(() => child.kill("SIGKILL"), limits.graceMs);' ' child.kill("SIGKILL");' $WT +run no-escalation $WK $' kill = setTimeout(() => child.kill("SIGKILL"), limits.graceMs);\n' '' $WT +# rollback-failed, refused and logged +run no-rollback-check $WK $' if (!undone.ok) refuse("rollback-failed",' $' if (false) refuse("rollback-failed",' $WT $GT +run rollback-not-logged $GS 'if (code === "git-failed" || code === "rollback-failed") log(' 'if (code === "git-failed") log(' $GT +# The five untested refusals +run no-root-check $WK ' if (!r.ok) refuse("root-unreadable",' ' if (false) refuse("root-unreadable",' $WT +run no-role-check $GS 'if (!who.role || who.business !== b.id) throw' 'if (who.business !== b.id) throw' $GT +run no-business-check $GS 'if (!who.role || who.business !== b.id) throw' 'if (!who.role) throw' $GT +run no-project-check $GS ' if (!p) throw new Refusal("unknown-project");' '' $GT +run no-busy-check $GS 'if (active.size >= maxActive) return' 'if (false) return' $GT +run no-closing-answer $GS $' waiting.set(socket, () => done(new Refusal("host-closing")));\n' '' $GT +# The body's total deadline +run no-body-deadline $GS '), deadlineMs);' '), 2 ** 31 - 1);' $GT +# fetch-aborted +run abort-as-git-failed $GS 'const aborted = () => new Refusal("fetch-aborted",' 'const aborted = () => new Error(' $GT +# A reviewer that hangs up once it has every byte was served. The first cut +# filed any close before 'finish' as fetch-aborted; so does this mutant. +run close-is-aborted $GS 'const served = () => !socket.errored;' 'const served = () => socket.writableFinished;' $GT +# A queued write that fails is fetch-aborted. +run errored-ignored $GS 'const served = () => !socket.errored;' 'const served = () => true;' $GT +# have: working refs first, at most HAVE_MAX +run have-origin-first $GTL 'for (const remote of ["refs/remotes/working", "refs/remotes/origin"])' 'for (const remote of ["refs/remotes/origin", "refs/remotes/working"])' $HT +run have-uncapped $GTL 'if (OID.test(oid) && have.size < HAVE_MAX) have.add(oid);' 'if (OID.test(oid)) have.add(oid);' $HT diff --git a/agents/darkwing/work/queue-100/r1/build.md b/agents/darkwing/work/queue-100/r1/build.md new file mode 100644 index 000000000..c3eb37495 --- /dev/null +++ b/agents/darkwing/work/queue-100/r1/build.md @@ -0,0 +1,116 @@ +# Row 100 (#1594) round 1: git socket hardening and refusal tests + +Brief: docs/plans/2026-10-12_pm-test-fixes.md, section "Row 93 follow-up: git socket hardening and refusal tests" (commit 8fda9611, blob 017ac786). +BASE: 8e0fbca420a3161ce98c5e0f3910f500daabebba. Candidate: build.patch against BASE; candidate-manifest.sha256 lists the files. + +## What changed + +- packages/seat/src/working.mjs. A git past GIT_TIMEOUT_MS gets SIGTERM, then SIGKILL after GIT_GRACE_MS (5 s). git removes its lockfiles on SIGTERM. A rollback whose update-ref fails refuses rollback-failed, which outranks record's own error. push and review take `limits` so the tests can shorten both timers. +- packages/cli/src/git-socket.mjs. + - readBody has a total deadline, `bodyDeadline(size) = 30000 + ceil(size / 256)` ms (161 s at the 32 MiB cap), beside the 30 s idle timeout. Both refuse body-timeout. + - The bundle send to a reviewer is a loop, `send()`. It used to be pipeline(), whose ERR_STREAM_PREMATURE_CLOSE became git-failed. A close while chunks are still to go is fetch-aborted, and so is a close after a write of a queued chunk failed (EPIPE). A close after every chunk left with no failed write is a served fetch, because the runner reads exactly `size` bytes and hangs up, often before the host's loop sees the file's end or before 'finish'. + - git-failed and rollback-failed go to the log as `git socket: : `. + - close() answers host-closing to a connection still reading its request line, and drops the rest as before. + - `maxActive` and `bodyDeadline` are options for the tests. +- packages/harness/src/git-tools.mjs. fetch_review's `have` lists every clone's working refs, newest first, before any of origin's, and stops at HAVE_MAX. +- packages/seat/README.md: rollback-failed, the SIGTERM/SIGKILL timing, and the lock's one-launcher-per-dataRoot assumption. +- packages/cli/README.md: body-timeout, what nothing-to-fetch, unknown-project and fetch-aborted mean to a reviewer, the log lines, and host-closing. + +## Tests + +| Brief item | Test | +|---|---| +| not-a-role | git-socket.test: no role, then another business | +| host-closing | git-socket.test: close() while a request line is half sent | +| git-busy | git-socket.test: maxActive 2, a third connection | +| root-unreadable | working.test: a root that doesn't exist | +| unknown-project | git-socket.test: a review naming a project the business lacks | +| timed-out push leaves no lock | working.test: a hook holds the ref lock, SIGTERM, no .lock, next push lands | +| SIGKILL escalation | working.test: a git that ignores SIGTERM | +| rollback-failed | working.test (delete and previous-ref cases) and git-socket.test (logged, trail) | +| trickling client | git-socket.test: body-timeout within the deadline, slot freed | +| fetch-aborted | git-socket.test: a reviewer leaves mid-bundle; no git-failed, no log | +| a served fetch isn't fetch-aborted | git-socket.test: the review loop now closes the socket and checks action.refused is empty; a raw reviewer takes a 1 MiB bundle and hangs up at once, ten times; `send()` against two stand-in sockets, one that never finishes and one whose queued write fails EPIPE | +| have over 32 refs | harness git-tools.test: 53 refs over two clones | + +## Mutants + +16 mutants, each in `agents/darkwing/work/queue-100/mutants.sh`, run with `mut.mjs` (one exact match, else exit 3) and `run.py` (own process group, killpg at 240 s). All 16 killed: + +``` +sigkill-first: killed (ℹ fail 1, ✖ a git past its time gets SIGTERM, leaves no ref lock, and the next push lands (10291.116) +no-escalation: killed (TIMEOUT after 240 s, ✖ a git that ignores SIGTERM gets SIGKILL once the grace period is over (20098.619138ms)) +no-rollback-check: killed (ℹ fail 2, ✖ a record that fails and can't be undone refuses rollback-failed, which the host logs (39) +rollback-not-logged: killed (ℹ fail 1, ✖ a record that fails and can't be undone refuses rollback-failed, which the host logs (41) +no-root-check: killed (ℹ fail 2, ✖ a root that can't be read refuses root-unreadable (35.076601ms)) +no-role-check: killed (ℹ fail 1, ✖ an identity with no role, or of another business, refuses not-a-role and moves no ref (2) +no-business-check: killed (ℹ fail 1, ✖ an identity with no role, or of another business, refuses not-a-role and moves no ref (2) +no-project-check: killed (ℹ fail 1, ✖ a review of a project the business no longer has refuses unknown-project (259.074398ms)) +no-busy-check: killed (ℹ fail 2, ✖ past maxActive connections the socket refuses git-busy, and a hung-up connection frees i) +no-closing-answer: killed (ℹ fail 1, ✖ close() answers host-closing to a connection still sending its request line (180.301962m) +no-body-deadline: killed (ℹ fail 1, ✖ a bundle that trickles in refuses body-timeout at its deadline and frees its slot (5212.) +abort-as-git-failed: killed (ℹ fail 2, ✖ a reviewer gone before its bundle was sent is filed fetch-aborted, not as the host's fau) +close-is-aborted: killed (ℹ fail 2, ✖ a reviewer that hangs up the moment it has the whole bundle was served, not aborted (508) +errored-ignored: killed (ℹ fail 1, ✖ send: a close after every byte left is served; one that fails a queued write is fetch-ab) +have-origin-first: killed (ℹ fail 1, ✖ fetch_review's have lists the working refs before origin's, and no more than HAVE_MAX (1) +have-uncapped: killed (ℹ fail 1, ✖ fetch_review's have lists the working refs before origin's, and no more than HAVE_MAX (1) +``` + +no-escalation hits the 240 s cap: its test fails at its own 20 s timeout, and the wrapper git that ignores SIGTERM keeps node --test alive until run.py kills the group. The sweep after each run found none of my processes left. + +### What the mutants found + +The first cut of `send()` waited for 'finish' and filed any earlier close as fetch-aborted. abort-as-git-failed showed it: the review loop test logged `git-failed: the reviewer left before the bundle was sent` for both of its fetches, so with the real code each successful fetch was on the trail as `reviewer fetch-aborted`. The review loop test checked the log but not action.refused, so it passed. Instrumenting showed the close usually comes before the loop reads the file's end, not only before 'finish'. The fix is above; the review loop now checks action.refused after close(), and the three tests in the table cover the race. A mutant that counted bytes (`sent === size`) survived as an equivalent: the loop's `destroyed` check already refuses a close with chunks to go, so I dropped the counter. + +## Gate + +Run in a clean worktree at BASE with the candidate applied and node_modules symlinked, 16:49:50Z to 16:54:16Z on 2026-10-13, by `agents/darkwing/work/queue-100/gate.sh` (test-task under DOCKER_HOST=unix:///nonexistent). Every suite exited 0 with no failures: + +``` +bus rc=0 16:49:50-16:49:54 ℹ pass 127 ℹ fail 0 ℹ skipped 0 0 marks +business rc=0 16:49:54-16:49:55 ℹ pass 67 ℹ fail 0 ℹ skipped 0 0 marks +cli rc=0 16:49:55-16:50:12 ℹ pass 146 ℹ fail 0 ℹ skipped 0 0 marks +control-board rc=0 16:50:12-16:50:13 ℹ pass 124 ℹ fail 0 ℹ skipped 0 0 marks +conversation rc=0 16:50:13-16:51:01 ℹ pass 182 ℹ fail 0 ℹ skipped 0 0 marks +discord rc=0 16:51:01-16:51:03 ℹ pass 178 ℹ fail 0 ℹ skipped 0 0 marks +harness rc=0 16:51:03-16:51:14 ℹ pass 70 ℹ fail 0 ℹ skipped 0 0 marks +mosaic rc=0 16:51:14-16:51:16 ℹ pass 69 ℹ fail 0 ℹ skipped 0 0 marks +queue rc=0 16:51:16-16:51:45 ℹ pass 150 ℹ fail 0 ℹ skipped 0 0 marks +runs rc=0 16:51:45-16:51:45 ℹ pass 43 ℹ fail 0 ℹ skipped 0 0 marks +seat rc=0 16:51:45-16:51:50 ℹ pass 109 ℹ fail 0 ℹ skipped 0 0 marks +tasks rc=0 16:51:50-16:51:56 ℹ pass 90 ℹ fail 0 ℹ skipped 0 0 marks +webui rc=0 16:51:56-16:52:42 ℹ pass 107 ℹ fail 0 ℹ skipped 0 0 marks +test-auth rc=0 16:52:42-16:52:42 selftest: 15 passed, 0 failed +test-conductor rc=0 16:52:42-16:52:47 selftest: 17 passed, 0 failed +test-config rc=0 16:52:47-16:52:48 selftest: 24 passed, 0 failed +test-discord rc=0 16:52:48-16:52:55 discord suite: 66 passed, 0 failed +test-extension-package rc=0 16:52:55-16:52:56 extension package selftest: 18 passed, 0 failed +test-foundation rc=0 16:52:56-16:53:32 selftest: 44 passed, 0 failed +test-queue rc=0 16:53:32-16:54:03 queue suite: 27 passed, 0 failed +test-release rc=0 16:54:03-16:54:14 selftest: 14 passed, 0 failed +test-task rc=0 16:54:14-16:54:16 selftest: 26 passed, 0 failed +ledger rc=0 run separately: pass 78, fail 0 +``` + +packages/ledger has tests but no package.json, so gate.sh prints a MODULE_NOT_FOUND line and "ledger: no test script" for it. I ran `node --test packages/ledger/tests/` on its own; 78 passed. The ledger tests don't touch the row's files. + +The cli suite also passed eight runs in a row after the `send()` fix (five before the counter came out, three on the final code). + +BASE..origin/refactor (a50c9d23a) touches none of the eight manifest paths. build.patch applied to a fresh BASE checkout reproduces candidate-manifest.sha256. + +## Choices + +- host-closing comes from the waiting map in close(). The `if (closed)` check after readHeader stays as a guard; with the map it is unreachable in practice. +- body-timeout is a new code for both timeouts. The idle timeout refused invalid-request before. broker.refuse takes any `/^[a-z-]{1,64}$/` code, so there is no schema change. The 30 s idle path has no test of its own; the deadline test uses the same done(). +- fetch-aborted isn't logged: a reviewer leaving is not a host fault. +- `have` sorts by committer date, newest first, inside each remote. + +## Not owned, for the reviewers + +roles/reviewer.md and the harness README and tools.mjs are what a reviewer reads, and row 100 doesn't own them. The nothing-to-fetch text is in packages/cli/README.md. Pointing roles/reviewer.md at it is a follow-up. + +## Follow-ups (not in this row) + +- packages/seat/src/checkout.mjs git() and the harness runner's git() still use spawn's timeout with SIGKILL. +- The kill goes to the direct git child, not its process group; a hook or helper git spawns may outlive it. +- startHost's host.json single-host check isn't atomic (the README says so). diff --git a/agents/darkwing/work/queue-100/r1/build.patch b/agents/darkwing/work/queue-100/r1/build.patch new file mode 100644 index 000000000..448acc8d5 --- /dev/null +++ b/agents/darkwing/work/queue-100/r1/build.patch @@ -0,0 +1,1051 @@ +diff --git a/packages/cli/README.md b/packages/cli/README.md +index d37f2d3f4..fb964dd44 100644 +--- a/packages/cli/README.md ++++ b/packages/cli/README.md +@@ -336,12 +336,31 @@ refuses, the ref goes back. A refusal is `{ok: false, error: }` 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 + +diff --git a/packages/cli/src/git-socket.mjs b/packages/cli/src/git-socket.mjs +index 5ac6b208e..f3ea43117 100644 +--- a/packages/cli/src/git-socket.mjs ++++ b/packages/cli/src/git-socket.mjs +@@ -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 }); + } +diff --git a/packages/cli/tests/git-socket.test.mjs b/packages/cli/tests/git-socket.test.mjs +index e401ec1bf..998a0d795 100644 +--- a/packages/cli/tests/git-socket.test.mjs ++++ b/packages/cli/tests/git-socket.test.mjs +@@ -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); ++}); +diff --git a/packages/harness/src/git-tools.mjs b/packages/harness/src/git-tools.mjs +index 57b0bfa7e..50fd5336e 100644 +--- a/packages/harness/src/git-tools.mjs ++++ b/packages/harness/src/git-tools.mjs +@@ -169,11 +169,16 @@ export async function pushWorking({ path, cap, workspace, task_ref, project, bra + // Resolves { project, ref, commit, clone }; ref is the clone's remote ref. + export async function fetchReview({ path, cap, workspace, task_ref, request, tmp = "/tmp", env = process.env }) { + // Commits the host may leave out: what the clones' remotes already hold. ++ // The working refs come first, newest first, since a review is most often ++ // of a commit built on one; origin's fill what is left of HAVE_MAX. + const have = new Set(); +- for (const project of projectsIn(workspace)) { +- const r = await git(["-C", join(workspace, project), "for-each-ref", "--format=%(objectname)", "refs/remotes/origin", "refs/remotes/working"], env); +- if (r.ok) for (const oid of r.out.split("\n")) if (OID.test(oid) && have.size < HAVE_MAX) have.add(oid); +- } ++ const projects = projectsIn(workspace); ++ for (const remote of ["refs/remotes/working", "refs/remotes/origin"]) ++ for (const project of projects) { ++ if (have.size >= HAVE_MAX) break; ++ const r = await git(["-C", join(workspace, project), "for-each-ref", "--sort=-committerdate", "--format=%(objectname)", remote], env); ++ if (r.ok) for (const oid of r.out.split("\n")) if (OID.test(oid) && have.size < HAVE_MAX) have.add(oid); ++ } + const file = join(tmp, `fetch-${randomBytes(8).toString("hex")}.bundle`); + try { + const conn = open(path); +diff --git a/packages/harness/tests/git-tools.test.mjs b/packages/harness/tests/git-tools.test.mjs +new file mode 100644 +index 000000000..b66fc2874 +--- /dev/null ++++ b/packages/harness/tests/git-tools.test.mjs +@@ -0,0 +1,80 @@ ++// The runner's git tools (src/git-tools.mjs) against a fake git socket that ++// records the request line. The review loop over the real host's socket is in ++// packages/cli/tests/git-socket.test.mjs. ++ ++import { test } from "node:test"; ++import assert from "node:assert/strict"; ++import { spawnSync } from "node:child_process"; ++import { mkdirSync } from "node:fs"; ++import { createServer } from "node:net"; ++import { join } from "node:path"; ++import { fetchReview, HAVE_MAX } from "../src/git-tools.mjs"; ++import { scratch } from "./helpers.mjs"; ++ ++const ENV = { PATH: process.env.PATH, HOME: "/nonexistent", GIT_CONFIG_NOSYSTEM: "1", GIT_CONFIG_GLOBAL: "/dev/null", LC_ALL: "C" }; ++ ++function git(cwd, args, extra = {}) { ++ const r = spawnSync("git", ["-c", "user.name=t", "-c", "user.email=t@example.invalid", "-C", cwd, ...args], { env: { ...ENV, ...extra }, encoding: "utf8" }); ++ assert.equal(r.status, 0, `git ${args.join(" ")}: ${r.stderr}`); ++ return r.stdout.trim(); ++} ++ ++// A clone at `dir` with `refs` (name -> commit time) pointing at commits of ++// their own. ++function clone(dir, refs) { ++ mkdirSync(dir, { recursive: true }); ++ git(dir, ["init", "-q"]); ++ const tree = git(dir, ["mktree"]); ++ const oids = {}; ++ for (const [name, time] of Object.entries(refs)) { ++ const date = `${time} +0000`; ++ oids[name] = git(dir, ["commit-tree", tree, "-m", name], { GIT_AUTHOR_DATE: date, GIT_COMMITTER_DATE: date }); ++ git(dir, ["update-ref", name, oids[name]]); ++ } ++ return oids; ++} ++ ++// A socket that keeps each request line and refuses it. ++async function fakeHost(t, dir) { ++ const path = join(dir, "git.sock"); ++ const lines = []; ++ const server = createServer((s) => { ++ let buf = ""; ++ s.on("data", (b) => { ++ buf += b; ++ if (!buf.includes("\n")) return; ++ lines.push(JSON.parse(buf.split("\n")[0])); ++ s.end(`${JSON.stringify({ ok: false, error: "request-not-found" })}\n`); ++ }); ++ }); ++ await new Promise((r) => server.listen(path, r)); ++ t.after(() => new Promise((r) => server.close(r))); ++ return { path, lines }; ++} ++ ++test("fetch_review's have lists the working refs before origin's, and no more than HAVE_MAX", async (t) => { ++ const dir = scratch(t); ++ const workspace = join(dir, "ws"); ++ // 40 origin refs, newer than every working ref, and 13 working refs in two ++ // clones: 53 in all. ++ const stack = {}; ++ for (let i = 0; i < 40; i++) stack[`refs/remotes/origin/b${i}`] = 1_800_000_000 + i; ++ for (let i = 1; i <= 10; i++) stack[`refs/remotes/working/work/vikunja-1-${i}`] = 1_700_000_000 + i; ++ const other = {}; ++ for (let i = 1; i <= 3; i++) other[`refs/remotes/working/work/vikunja-2-${i}`] = 1_700_000_000 + i; ++ const a = clone(join(workspace, "stack"), stack); ++ const b = clone(join(workspace, "other"), other); ++ const host = await fakeHost(t, dir); ++ await assert.rejects(fetchReview({ path: host.path, cap: "c".repeat(64), workspace, task_ref: "vikunja:1/2", request: "r1", tmp: dir, env: ENV }), (e) => e.code === "request-not-found"); ++ const [req] = host.lines; ++ assert.equal(req.op, "fetch"); ++ assert.equal(HAVE_MAX, 32); ++ assert.equal(req.have.length, HAVE_MAX); ++ assert.equal(new Set(req.have).size, HAVE_MAX); ++ const working = [...Object.entries(a), ...Object.entries(b)].filter(([n]) => n.startsWith("refs/remotes/working/")).map(([, oid]) => oid); ++ assert.equal(working.length, 13); ++ // Every working ref, ahead of any of origin's. ++ assert.deepEqual(new Set(req.have.slice(0, 13)), new Set(working)); ++ const origin = new Set(Object.entries(a).filter(([n]) => n.startsWith("refs/remotes/origin/")).map(([, oid]) => oid)); ++ for (const oid of req.have.slice(13)) assert.ok(origin.has(oid), oid); ++}); +diff --git a/packages/seat/README.md b/packages/seat/README.md +index e993b9f09..8b830a956 100644 +--- a/packages/seat/README.md ++++ b/packages/seat/README.md +@@ -323,11 +323,25 @@ A push runs under the repository's lock, in this order: + fast-forward (`non-fast-forward`). + 8. The broker records the push, an `action.allowed` for + `git.push.working` naming the commit. When it +- refuses, the work ref goes back to where it was. ++ refuses, the work ref goes back to where it was. A work ref that can't ++ go back refuses `rollback-failed` in place of the broker's code, with a ++ message naming the ref and the commit it still holds. The host logs it, ++ since the operator has to move that ref by hand. + + The temp ref is deleted on every path, and so is the bundle file in + `bus-host/`. No other ref moves, and the root is only read. + ++The repository's lock is a promise chain inside the host's process, so it ++assumes one launcher per dataRoot. `startHost` refuses a second bus host ++while `bus-host/host.json` names a live one, but that check isn't atomic: ++two launchers started at the same moment could both pass it, and their ++pushes to one repository would no longer be serialized. ++ ++A host git call that runs past 100 s (`GIT_TIMEOUT_MS`) gets SIGTERM. git ++removes its lockfiles on SIGTERM. Only a git still running 5 s later ++(`GIT_GRACE_MS`) gets SIGKILL, which can leave a ref's `.lock` behind; that ++lock refuses every later push to the ref until someone removes it. ++ + For a review, under the same lock, the host fetches `main` again and + writes a bundle of the requested commit with one ref, + `refs/mosaic/review`. It leaves out the commits the reviewer says its +diff --git a/packages/seat/src/working.mjs b/packages/seat/src/working.mjs +index 62b4c73f0..3b8787972 100644 +--- a/packages/seat/src/working.mjs ++++ b/packages/seat/src/working.mjs +@@ -28,10 +28,19 @@ + // 6. the commit shares history with main (unrelated-history) + // 7. the temp ref moves to the work ref with no `+`, so git refuses + // anything but a fast-forward (non-fast-forward) +-// 8. record() runs; when it throws, the work ref goes back ++// 8. record() runs; when it throws, the work ref goes back. A work ref ++// that can't go back refuses rollback-failed in place of record's error. + // The temp ref is deleted on every path. The caller owns the bundle file and + // deletes it. + // ++// A git past GIT_TIMEOUT_MS gets SIGTERM, on which git removes its lockfiles, ++// and SIGKILL only if it still runs GIT_GRACE_MS later. A SIGKILL leaves a ++// ref's .lock behind, and that lock refuses every later push to the ref. ++// ++// The lock is a promise chain in this process, so it holds only while one ++// launcher serves a dataRoot. startHost refuses a second bus host while ++// bus-host/host.json names a live one; that check isn't atomic. ++// + // review, under the same lock: main again from the root, then a bundle of + // the requested commit that leaves out the commits the reviewer said it has + // (those the repository knows), written to the file the caller names. +@@ -50,6 +59,9 @@ const V2 = "# v2 git bundle\n"; + const OID = /^[0-9a-f]{40}([0-9a-f]{24})?$/; + const TASK_REF = /^vikunja:([1-9][0-9]*)\/([1-9][0-9]*)$/; + const PROJECT = /^[a-z0-9][a-z0-9._-]{0,63}$/; ++// How long a git that got SIGTERM has to exit before SIGKILL. ++export const GIT_GRACE_MS = 5000; ++const LIMITS = { timeoutMs: GIT_TIMEOUT_MS, graceMs: GIT_GRACE_MS }; + + export class WorkingError extends Error { + constructor(code, message = code) { +@@ -72,17 +84,30 @@ const SAFE = [ + // Ahead of the fetch from a bundle. Tests replace it to show it matters. + export const FSCK = ["-c", "transfer.fsckObjects=true"]; + +-function git(args, env, { pre = [] } = {}) { ++// `limits` is { timeoutMs, graceMs }; the tests shorten them. ++function git(args, env, { pre = [], limits = LIMITS } = {}) { + return new Promise((resolve) => { +- const child = spawn("git", [...SAFE, ...pre, ...args], { env: gitEnv(env), timeout: GIT_TIMEOUT_MS, killSignal: "SIGKILL", stdio: ["ignore", "pipe", "pipe"] }); ++ const child = spawn("git", [...SAFE, ...pre, ...args], { env: gitEnv(env), stdio: ["ignore", "pipe", "pipe"] }); + let out = ""; + let err = ""; ++ let kill = null; ++ const term = setTimeout(() => { ++ child.kill("SIGTERM"); ++ kill = setTimeout(() => child.kill("SIGKILL"), limits.graceMs); ++ }, limits.timeoutMs); ++ const settle = (r) => { ++ clearTimeout(term); ++ clearTimeout(kill); ++ resolve(r); ++ }; + child.stdout.setEncoding("utf8").on("data", (d) => (out += d)); + child.stderr.setEncoding("utf8").on("data", (d) => (err += d)); +- child.on("error", (e) => resolve({ ok: false, out, err, why: e.code ?? e.message })); +- child.on("close", (code, signal) => resolve({ ok: code === 0, out, err, why: signal ? `killed by ${signal}` : err.trim().split("\n").pop() || `exit ${code}` })); ++ child.on("error", (e) => settle({ ok: false, out, err, why: e.code ?? e.message })); ++ child.on("close", (code, signal) => settle({ ok: code === 0, out, err, why: signal ? `killed by ${signal}` : err.trim().split("\n").pop() || `exit ${code}` })); + }); + } ++// git with this call's env and limits, for the helpers below. ++const bound = (env, limits) => (args, opts = {}) => git(args, env, { ...opts, limits }); + + // The branch name a task's work goes to. A task ref holds a colon, which a + // ref name can't. +@@ -117,7 +142,7 @@ function locked(repo, fn) { + + // The bare repository, made on first use in a temp directory beside it and + // renamed into place. One that exists must be a real directory. +-async function ensure(repo, env) { ++async function ensure(repo, g) { + try { + if (!lstatSync(repo).isDirectory()) refuse("working-repo-invalid", `${repo} isn't a directory`); + return; +@@ -128,7 +153,7 @@ async function ensure(repo, env) { + mkdirSync(dirname(repo), { recursive: true, mode: 0o700 }); + const tmp = join(dirname(repo), `.init-${randomBytes(4).toString("hex")}`); + try { +- const r = await git(["init", "-q", "--bare", "--template=", tmp], env); ++ const r = await g(["init", "-q", "--bare", "--template=", tmp]); + if (!r.ok) refuse("working-repo-invalid", `git init failed: ${r.why}`); + renameSync(tmp, repo); + } finally { +@@ -138,8 +163,8 @@ async function ensure(repo, env) { + + // main from what the root has checked out. The root is the owner's; it isn't + // a session's. +-async function seed(repo, root, env) { +- const r = await git([`--git-dir=${repo}`, "fetch", "-q", "--no-tags", "--no-write-fetch-head", "--no-recurse-submodules", "--no-auto-maintenance", root, "+HEAD:refs/heads/main"], env); ++async function seed(repo, root, g) { ++ const r = await g([`--git-dir=${repo}`, "fetch", "-q", "--no-tags", "--no-write-fetch-head", "--no-recurse-submodules", "--no-auto-maintenance", root, "+HEAD:refs/heads/main"]); + if (!r.ok) refuse("root-unreadable", `main couldn't be fetched from ${root}: ${r.why}`); + } + +@@ -153,68 +178,73 @@ function header(file) { + } + } + +-const tip = async (repo, ref, env) => { +- const r = await git([`--git-dir=${repo}`, "rev-parse", "-q", "--verify", `${ref}^{commit}`], env); ++const tip = async (repo, ref, g) => { ++ const r = await g([`--git-dir=${repo}`, "rev-parse", "-q", "--verify", `${ref}^{commit}`]); + return r.ok ? r.out.trim() : null; + }; + +-// await push({ dataRoot, business, project, root, taskRef, branch, bundle, record, env, fsck }) ++// await push({ dataRoot, business, project, root, taskRef, branch, bundle, record, env, fsck, limits }) + // Resolves { ref, commit, previous }; rejects with WorkingError. `record` + // runs under the lock once the ref moved, and a throw moves it back. `fsck` +-// is for the tests. +-export function push({ dataRoot, business, project, root, taskRef, branch, bundle, record = async () => {}, env = process.env, fsck = FSCK }) { ++// and `limits` are for the tests. ++export function push({ dataRoot, business, project, root, taskRef, branch, bundle, record = async () => {}, env = process.env, fsck = FSCK, limits = LIMITS }) { + if (!PROJECT.test(project ?? "")) refuse("invalid-request", "not a project id"); + if (!validBranch(branch)) refuse("invalid-request", "not a branch name"); + const ref = `refs/heads/${workRef(taskRef)}`; + const repo = workingRepo(dataRoot, business, project); ++ const g = bound(env, limits); + return locked(repo, async () => { + if (!header(bundle)) refuse("invalid-bundle", "not a v2 git bundle"); +- await ensure(repo, env); +- await seed(repo, root, env); ++ await ensure(repo, g); ++ await seed(repo, root, g); + const at = [`--git-dir=${repo}`]; + // Not -q: with it, git 2.55 drops the list of missing prerequisites. +- const verify = await git([...at, "bundle", "verify", bundle], env); ++ const verify = await g([...at, "bundle", "verify", bundle]); + if (!verify.ok) { + if (/lacks these prerequisite/.test(verify.err)) refuse("missing-prerequisite", "the bundle needs commits the working repository doesn't have"); + refuse("invalid-bundle", verify.why); + } +- const heads = await git([...at, "bundle", "list-heads", bundle], env); ++ const heads = await g([...at, "bundle", "list-heads", bundle]); + if (!heads.ok) refuse("invalid-bundle", heads.why); + if (!heads.out.split("\n").some((l) => l.split(" ")[1] === `refs/heads/${branch}`)) refuse("branch-not-in-bundle"); + const incoming = `refs/mosaic/incoming/${randomBytes(8).toString("hex")}`; + try { +- const fetched = await git([...at, "fetch", "-q", "--no-tags", "--no-write-fetch-head", "--no-recurse-submodules", "--no-auto-maintenance", bundle, `refs/heads/${branch}:${incoming}`], env, { pre: fsck }); ++ const fetched = await g([...at, "fetch", "-q", "--no-tags", "--no-write-fetch-head", "--no-recurse-submodules", "--no-auto-maintenance", bundle, `refs/heads/${branch}:${incoming}`], { pre: fsck }); + if (!fetched.ok) refuse("invalid-bundle", fetched.why); +- const commit = await tip(repo, incoming, env); ++ const commit = await tip(repo, incoming, g); + if (!commit) refuse("invalid-bundle", "the branch isn't a commit"); +- if (!(await git([...at, "merge-base", commit, "refs/heads/main"], env)).ok) refuse("unrelated-history", "the commit shares no history with main"); +- const previous = await tip(repo, ref, env); ++ if (!(await g([...at, "merge-base", commit, "refs/heads/main"])).ok) refuse("unrelated-history", "the commit shares no history with main"); ++ const previous = await tip(repo, ref, g); + // From the repository into itself, by its path (`.` would be the cwd), and + // not -q, which drops the "(non-fast-forward)" reason. +- const moved = await git([...at, "fetch", "--no-tags", "--no-write-fetch-head", "--no-auto-maintenance", repo, `${incoming}:${ref}`], env); ++ const moved = await g([...at, "fetch", "--no-tags", "--no-write-fetch-head", "--no-auto-maintenance", repo, `${incoming}:${ref}`]); + if (!moved.ok) refuse(/\(non-fast-forward\)/.test(moved.err) ? "non-fast-forward" : "push-failed", moved.why); + try { + await record({ ref: workRef(taskRef), commit, previous }); + } catch (e) { + const back = previous ? ["update-ref", ref, previous, commit] : ["update-ref", "-d", ref, commit]; +- await git([...at, ...back], env); ++ const undone = await g([...at, ...back]); ++ // The ref still names a commit the broker didn't record. That is the ++ // operator's to fix, so it outranks record's own error. ++ if (!undone.ok) refuse("rollback-failed", `${ref} still names ${commit}, which wasn't recorded (${e.code ?? e.message}); moving it back failed: ${undone.why}`); + throw e; + } + return { ref: workRef(taskRef), commit, previous }; + } finally { +- await git([...at, "update-ref", "-d", incoming], env); ++ await g([...at, "update-ref", "-d", incoming]); + } + }); + } + +-// await review({ dataRoot, business, project, root, taskRef, commit, have, out, env }) ++// await review({ dataRoot, business, project, root, taskRef, commit, have, out, env, limits }) + // Writes a bundle of `commit` to `out`, with one ref, refs/mosaic/review. + // `have` lists commits the reviewer's clone holds; those the repository + // doesn't know are skipped. Resolves { ref, commit }. +-export function review({ dataRoot, business, project, root, taskRef, commit, have = [], out, env = process.env }) { ++export function review({ dataRoot, business, project, root, taskRef, commit, have = [], out, env = process.env, limits = LIMITS }) { + if (!PROJECT.test(project ?? "") || !OID.test(commit ?? "")) refuse("invalid-request"); + if (!Array.isArray(have) || have.length > HAVE_MAX || have.some((h) => !OID.test(h ?? ""))) refuse("invalid-request", "have must list commit ids"); + const repo = workingRepo(dataRoot, business, project); ++ const g = bound(env, limits); + return locked(repo, async () => { + try { + if (!lstatSync(repo).isDirectory()) refuse("working-repo-invalid"); +@@ -222,18 +252,18 @@ export function review({ dataRoot, business, project, root, taskRef, commit, hav + if (e instanceof WorkingError) throw e; + refuse("unknown-commit", "the working repository doesn't exist"); + } +- await seed(repo, root, env); ++ await seed(repo, root, g); + const at = [`--git-dir=${repo}`]; +- if (!(await tip(repo, commit, env))) refuse("unknown-commit"); ++ if (!(await tip(repo, commit, g))) refuse("unknown-commit"); + const known = []; +- for (const h of have) if (await tip(repo, h, env)) known.push(h); ++ for (const h of have) if (await tip(repo, h, g)) known.push(h); + const temp = "refs/mosaic/review"; + try { +- if (!(await git([...at, "update-ref", temp, commit], env)).ok) refuse("fetch-failed"); +- const made = await git([...at, "bundle", "create", "-q", out, temp, ...(known.length ? ["--not", ...known] : [])], env); ++ if (!(await g([...at, "update-ref", temp, commit])).ok) refuse("fetch-failed"); ++ const made = await g([...at, "bundle", "create", "-q", out, temp, ...(known.length ? ["--not", ...known] : [])]); + if (!made.ok) refuse(/empty bundle/.test(made.err) ? "nothing-to-fetch" : "fetch-failed", made.why); + } finally { +- await git([...at, "update-ref", "-d", temp], env); ++ await g([...at, "update-ref", "-d", temp]); + } + return { ref: workRef(taskRef), commit }; + }); +diff --git a/packages/seat/tests/working.test.mjs b/packages/seat/tests/working.test.mjs +index 68a6de1f7..24fcd9764 100644 +--- a/packages/seat/tests/working.test.mjs ++++ b/packages/seat/tests/working.test.mjs +@@ -8,7 +8,7 @@ import { test } from "node:test"; + import assert from "node:assert/strict"; + import { spawnSync } from "node:child_process"; + import { createHash } from "node:crypto"; +-import { existsSync, lstatSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, readlinkSync, rmSync, writeFileSync } from "node:fs"; ++import { chmodSync, existsSync, lstatSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, readlinkSync, rmSync, writeFileSync } from "node:fs"; + import { tmpdir } from "node:os"; + import { join } from "node:path"; + import { gitEnv } from "../src/checkout.mjs"; +@@ -87,6 +87,34 @@ async function refuses(w, fn, code) { + assert.doesNotMatch(w.refs(), /refs\/mosaic\//, "no temp ref is left"); + } + ++// The .lock files under a repository's refs. ++function locks(repo) { ++ const found = []; ++ const walk = (d) => { ++ for (const name of readdirSync(d)) { ++ const p = join(d, name); ++ if (lstatSync(p).isDirectory()) walk(p); ++ else if (name.endsWith(".lock")) found.push(p.slice(repo.length + 1)); ++ } ++ }; ++ walk(repo); ++ return found; ++} ++ ++// A `git` first in PATH that runs `body` (bash) with REAL as the real git, ++// and the env that puts it there. ++const REAL_GIT = spawnSync("sh", ["-c", "command -v git"], { encoding: "utf8" }).stdout.trim(); ++function wrapGit(w, body) { ++ const bin = join(w.dir, "bin"); ++ mkdirSync(bin); ++ writeFileSync(join(bin, "git"), `#!/bin/bash ++REAL='${REAL_GIT}' ++${body} ++`); ++ chmodSync(join(bin, "git"), 0o755); ++ return { ...process.env, PATH: `${bin}:${process.env.PATH}` }; ++} ++ + async function lands(w, fn) { + const root = fingerprint(w.root); + const r = await fn(); +@@ -236,6 +264,85 @@ test("a record that throws moves the work ref back, or removes a new one", async + assert.equal(w.refs(), `refs/heads/main ${w.first}\n${REF} ${c1}`); + }); + ++test("a work ref that can't go back refuses rollback-failed, naming the ref and the commit", async (t) => { ++ const w = world(t); ++ git(w.clone, "checkout", "-q", "-b", "feat"); ++ const c1 = commit(w.clone, "two"); ++ // record() fails after taking the ref's lock, so moving the ref back fails. ++ const lock = join(w.repo, `${REF}.lock`); ++ const stuck = async () => { ++ writeFileSync(lock, ""); ++ throw new Error("not recorded"); ++ }; ++ const message = (c) => (e) => e instanceof WorkingError && e.code === "rollback-failed" && e.message.includes(`${REF} still names ${c}`) && /not recorded/.test(e.message); ++ // A new ref that can't be removed. ++ await assert.rejects(w.go(w.bundle(), { record: stuck }), message(c1)); ++ assert.equal(w.refs(), `refs/heads/main ${w.first}\n${REF} ${c1}`); ++ rmSync(lock); ++ // A ref that can't go back to the one before. ++ git(w.clone, "update-ref", "refs/remotes/working/work/vikunja-1-2", c1); ++ const c2 = commit(w.clone, "three"); ++ await assert.rejects(w.go(w.bundle(), { record: stuck }), message(c2)); ++ assert.equal(w.refs(), `refs/heads/main ${w.first}\n${REF} ${c2}`); ++ rmSync(lock); ++ assert.deepEqual(locks(w.repo), []); ++}); ++ ++test("a root that can't be read refuses root-unreadable", async (t) => { ++ const w = world(t); ++ git(w.clone, "checkout", "-q", "-b", "feat"); ++ commit(w.clone, "two"); ++ await refuses(w, () => w.go(w.bundle(), { root: join(w.dir, "no-root") }), "root-unreadable"); ++}); ++ ++test("a git past its time gets SIGTERM, leaves no ref lock, and the next push lands", async (t) => { ++ const w = world(t); ++ git(w.clone, "checkout", "-q", "-b", "feat"); ++ const c1 = commit(w.clone, "two"); ++ await w.go(w.bundle()); ++ git(w.clone, "update-ref", "refs/remotes/working/work/vikunja-1-2", c1); ++ // The move of the work ref holds the ref's lock in a reference-transaction ++ // hook until the lock is gone, so the git is still holding it when its time ++ // is up. Only that one fetch gets the hook. ++ const hooks = join(w.dir, "hooks"); ++ mkdirSync(hooks); ++ const lock = join(w.repo, `${REF}.lock`); ++ writeFileSync(join(hooks, "reference-transaction"), `#!/bin/bash\n[ "$1" = prepared ] || exit 0\nfor i in $(seq 200); do [ -e '${lock}' ] || exit 0; sleep 0.05; done\n`); ++ chmodSync(join(hooks, "reference-transaction"), 0o755); ++ const env = wrapGit( ++ w, ++ `self=0 ++for a in "$@"; do case "$a" in refs/mosaic/incoming/*:refs/heads/work/*) self=1;; esac; done ++args=() ++for a in "$@"; do ++ [ $self = 1 ] && [ "$a" = core.hooksPath=/dev/null ] && a='core.hooksPath=${hooks}' ++ args+=("$a") ++done ++exec "$REAL" "\${args[@]}"`, ++ ); ++ commit(w.clone, "three"); ++ const limits = { timeoutMs: 2000, graceMs: 5000 }; ++ await assert.rejects(w.go(w.bundle(), { env, limits }), (e) => e instanceof WorkingError && e.code === "push-failed" && /killed by SIGTERM/.test(e.message)); ++ assert.deepEqual(locks(w.repo), []); ++ assert.equal(w.refs(), `refs/heads/main ${w.first}\n${REF} ${c1}`); ++ const c3 = commit(w.clone, "four"); ++ const r = await lands(w, () => w.go(w.bundle())); ++ assert.deepEqual(r, { ref: "work/vikunja-1-2", commit: c3, previous: c1 }); ++}); ++ ++test("a git that ignores SIGTERM gets SIGKILL once the grace period is over", { timeout: 20000 }, async (t) => { ++ const w = world(t); ++ git(w.clone, "checkout", "-q", "-b", "feat"); ++ commit(w.clone, "two"); ++ // The fetch of main from the root never ends and ignores SIGTERM. ++ const env = wrapGit( ++ w, ++ `for a in "$@"; do [ "$a" = +HEAD:refs/heads/main ] && { trap '' TERM; while :; do sleep 0.05; done; }; done ++exec "$REAL" "$@"`, ++ ); ++ await assert.rejects(w.go(w.bundle(), { env, limits: { timeoutMs: 500, graceMs: 500 } }), (e) => e instanceof WorkingError && e.code === "root-unreadable" && /killed by SIGKILL/.test(e.message)); ++}); ++ + test("a working repository path that isn't a directory refuses", async (t) => { + const w = world(t); + git(w.clone, "checkout", "-q", "-b", "feat"); diff --git a/agents/darkwing/work/queue-100/r1/candidate-manifest.sha256 b/agents/darkwing/work/queue-100/r1/candidate-manifest.sha256 new file mode 100644 index 000000000..1d6f850cd --- /dev/null +++ b/agents/darkwing/work/queue-100/r1/candidate-manifest.sha256 @@ -0,0 +1,8 @@ +9137d8fd6ff173d4f04b4fdc2f4caed1ab4492df9c18d63aa4fd26639d1dad49 packages/cli/README.md +57b401ce3782ae32cf60877913c18b6b4d6db6b5a499498abd6c26ddc5ce1b58 packages/cli/src/git-socket.mjs +8b6c3f9c575b026081c737cce1a0c1f1cd2eff4394092821719ba03e01a659a7 packages/cli/tests/git-socket.test.mjs +f07a62bbe53a237ac8aceab0760f5332a1f1cb6f629cfb2822747ba2a559b9fb packages/harness/src/git-tools.mjs +15dccbb011be0ba9763afc47e7b423a51de6ccac8c90659eb3660798470ada27 packages/harness/tests/git-tools.test.mjs +e1e783be589085801acbf5f536b848d8113ce3e7a7bb4e44c0933c04c604ac41 packages/seat/README.md +d23a4377a50fa8e63c8396a3e4fe35ca5bf9f3c114fe22be55e1be3c49a635c1 packages/seat/src/working.mjs +b0654288c7ccaed52ee2e8e15d7cd8a1efb2fe8414a838491dad338c4d0c9b8b packages/seat/tests/working.test.mjs diff --git a/agents/darkwing/work/queue-100/run.py b/agents/darkwing/work/queue-100/run.py new file mode 100644 index 000000000..8eb9ebdb7 --- /dev/null +++ b/agents/darkwing/work/queue-100/run.py @@ -0,0 +1,19 @@ +# python3 run.py : runs cmd in its own session, kills +# the whole group on timeout and sweeps it after a normal exit. Exit 124 on +# timeout, else the command's code. +import os, signal, subprocess, sys +secs, log, cmd = int(sys.argv[1]), sys.argv[2], sys.argv[3:] +with open(log, "w") as out: + p = subprocess.Popen(cmd, stdout=out, stderr=subprocess.STDOUT, start_new_session=True) + try: + rc = p.wait(timeout=secs) + except subprocess.TimeoutExpired: + os.killpg(p.pid, signal.SIGKILL) + p.wait() + out.write(f"\nTIMEOUT after {secs} s\n") + rc = 124 + try: + os.killpg(p.pid, signal.SIGKILL) + except ProcessLookupError: + pass +sys.exit(rc)