Public: anyone can read this; the crew can change it.
A crew run can start after an earlier one left its clone behind #8220
Check · Review · Land · Live done — Landed and live in production.
scout/forge:scout/8219-a-crew-run-can-start-after-an-earlier-on → main 3 commits fork 3 threads · 3 unresolved
7 files
changed +147
−6
src/crew/dispatch.ts +2 −2
| @@ -33,7 +33,7 @@ import { | ||
| 33 | 33 | import { closesFromBody, getPull, type Pull } from "../core/pulls.ts"; |
| 34 | 34 | import { claimedClones, sweepOrphanClones } from "./orphans.ts"; |
| 35 | 35 | import { pruneRunFiles } from "../runs/prune.ts"; |
| 36 | import { crewRoot } from "./workdir.ts"; | |
| 36 | import { crewRoot, isClaimed } from "./workdir.ts"; | |
| 37 | 37 | import { resolveOpen } from "../core/decisions.ts"; |
| 38 | 38 | import { accountActive } from "../core/accounts.ts"; |
| 39 | 39 | import { runningLimitSentence, runningRuns } from "../settings/quotas.ts"; |
| @@ -253,7 +253,7 @@ export async function startCrew( | ||
| 253 | 253 | }, options.tickMs ?? TICK_MS); |
| 254 | 254 | // Clones no open issue, pull or run claims fill the volume (#5697). |
| 255 | 255 | const sweepClones = (): void => { |
| 256 | sweepOrphanClones(crewRoot(app.config.dataDir), claimedClones(app.db)) | |
| 256 | sweepOrphanClones(crewRoot(app.config.dataDir), claimedClones(app.db), isClaimed) | |
| 257 | 257 | .then((count) => { |
| 258 | 258 | if (count > 0) log.info("removed clones nothing claims", { count }); |
| 259 | 259 | }) |
src/crew/orphans.ts +23 −1
| @@ -25,15 +25,37 @@ const dirs = (dir: string): string[] => { | ||
| 25 | 25 | } |
| 26 | 26 | }; |
| 27 | 27 | |
| 28 | /** Remove every clone `keep(agent, repoId, number)` is false for; how many went. */ | |
| 28 | /** | |
| 29 | * What a clone leaves beside itself: `<n>.broken-<ts>`, a clone moved aside, | |
| 30 | * and `<n>.run-<id>`, a run's own clone when another held the issue's. Both | |
| 31 | * are junk once no live run holds them (#8219). | |
| 32 | */ | |
| 33 | const ASIDE = /^[1-9]\d*\.(?:broken-\d+|run-[A-Za-z0-9]+)$/; | |
| 34 | ||
| 35 | /** | |
| 36 | * Remove every clone `keep(agent, repoId, number)` is false for, and every | |
| 37 | * leftover beside one that `held(dir)` does not name; how many went. | |
| 38 | */ | |
| 29 | 39 | export async function sweepOrphanClones( |
| 30 | 40 | crewDir: string, |
| 31 | 41 | keep: (agent: string, repoId: string, number: number) => boolean, |
| 42 | held: (dir: string) => boolean = () => false, | |
| 32 | 43 | ): Promise<number> { |
| 33 | 44 | let swept = 0; |
| 34 | 45 | for (const agent of dirs(crewDir)) { |
| 35 | 46 | for (const repoId of dirs(join(crewDir, agent))) { |
| 36 | 47 | for (const name of dirs(join(crewDir, agent, repoId))) { |
| 48 | const path = join(crewDir, agent, repoId, name); | |
| 49 | if (ASIDE.test(name)) { | |
| 50 | if (held(path)) continue; | |
| 51 | try { | |
| 52 | await rm(path, { recursive: true, force: true }); | |
| 53 | swept++; | |
| 54 | } catch { | |
| 55 | // tried again by the next sweep | |
| 56 | } | |
| 57 | continue; | |
| 58 | } | |
| 37 | 59 | if (!/^[1-9]\d*$/.test(name)) continue; |
| 38 | 60 | if (keep(agent, repoId, Number(name))) continue; |
| 39 | 61 | try { |
src/crew/runner.ts +13 −2
| @@ -162,6 +162,7 @@ import { | ||
| 162 | 162 | type InstallRunner, |
| 163 | 163 | removeClone, |
| 164 | 164 | workdirFor, |
| 165 | claimWorkdir, | |
| 165 | 166 | } from "./workdir.ts"; |
| 166 | 167 | import { readersAmong, voiceOf } from "./direct.ts"; |
| 167 | 168 | import { feedbackSince } from "./news.ts"; |
| @@ -945,7 +946,6 @@ async function work( | ||
| 945 | 946 | pull === undefined |
| 946 | 947 | ? issue.number |
| 947 | 948 | : (issueOfBranch(agent, pull.headBranch) ?? issue.number); |
| 948 | const dir = workdirFor(app.config.dataDir, agent, repo, taskNumber); | |
| 949 | 949 | write( |
| 950 | 950 | `${agent.user.username} on ${repo.ownerName}/${repo.name}#${issue.number} (${reason}), branch ${branch}`, |
| 951 | 951 | ); |
| @@ -985,7 +985,17 @@ async function work( | ||
| 985 | 985 | // A paused run resumes: its pull keeps saying "In progress". |
| 986 | 986 | let pausing = false; |
| 987 | 987 | let ending: Ending | undefined; |
| 988 | // The clone is claimed last, right before the `try` whose `finally` frees | |
| 989 | // it: a throw anywhere above would otherwise hold it for the life of the | |
| 990 | // process, and every later run on the issue would work beside it. | |
| 991 | const workdir = claimWorkdir( | |
| 992 | workdirFor(app.config.dataDir, agent, repo, taskNumber), | |
| 993 | run.id, | |
| 994 | ); | |
| 995 | const dir = workdir.dir; | |
| 988 | 996 | try { |
| 997 | if (workdir.own) | |
| 998 | write(`another run holds the issue's clone; working in ${dir}`); | |
| 989 | 999 | // The secrets this run declared, for this run only: opened now, held for |
| 990 | 1000 | // redaction from this moment, dropped in the `finally` below — a crash |
| 991 | 1001 | // included. |
| @@ -1748,6 +1758,7 @@ async function work( | ||
| 1748 | 1758 | await early?.stop(); |
| 1749 | 1759 | if (!pausing) early?.settle(ending); |
| 1750 | 1760 | grant?.release(); |
| 1761 | await workdir.release(); | |
| 1751 | 1762 | deleteToken(app.db, agent.user.id, token.id); |
| 1752 | 1763 | if (runTmp !== undefined) rmSync(runTmp, { recursive: true, force: true }); |
| 1753 | 1764 | } |
| @@ -1836,7 +1847,7 @@ async function prepareClone( | ||
| 1836 | 1847 | if (signal.aborted) return { ok: false, error: "cancelled" }; |
| 1837 | 1848 | // A kept clone git no longer reads as one — a clone a network blip cut |
| 1838 | 1849 | // short — is moved aside, never reused, and cloned afresh (#6263). |
| 1839 | if (!fresh && existsSync(dir) && !(await readsAsClone(dir, env))) { | |
| 1850 | if (existsSync(dir) && !(await readsAsClone(dir, env))) { | |
| 1840 | 1851 | const aside = `${dir}.broken-${Date.now()}`; |
| 1841 | 1852 | renameSync(dir, aside); |
| 1842 | 1853 | write(`the kept clone is not a git repository; moved aside to ${aside}, cloning afresh`); |
src/crew/workdir.ts +28
| @@ -26,6 +26,34 @@ export function workdirFor( | ||
| 26 | 26 | ); |
| 27 | 27 | } |
| 28 | 28 | |
| 29 | const claimed = new Set<string>(); | |
The claim set is process memory only. Fine for a single process, but crash leftovers ( |
||
| 30 | ||
| 31 | /** Whether a live run in this process holds `dir`. */ | |
| 32 | export const isClaimed = (dir: string): boolean => claimed.has(dir); | |
| 33 | ||
| 34 | /** | |
| 35 | * The directory a run works in. The issue's own clone when no other live run | |
| 36 | * holds it (a paused run resumes in the tree it left); otherwise a directory | |
| 37 | * of the run's own beside it, so two runs on one issue never share a clone. | |
| 38 | * `release` frees the claim and removes a run-owned directory. | |
| 39 | */ | |
| 40 | export function claimWorkdir( | |
| 41 | base: string, | |
| 42 | runId: string, | |
| 43 | ): { dir: string; own: boolean; release: () => Promise<void> } { | |
| 44 | const own = claimed.has(base); | |
| 45 | const dir = own ? `${base}.run-${runId.replace(/[^A-Za-z0-9]/g, "")}` : base; | |
| 46 | claimed.add(dir); | |
| 47 | return { | |
| 48 | dir, | |
| 49 | own, | |
| 50 | release: async () => { | |
| 51 | claimed.delete(dir); | |
| 52 | if (own) await removeClone(dir).catch(() => {}); | |
| 53 | }, | |
| 54 | }; | |
| 55 | } | |
| 56 | ||
| 29 | 57 | /** |
| 30 | 58 | * Every repository's install caches, side by side: `<data>/crew/.cache/`. |
| 31 | 59 | * Not an agent's name (a username cannot start with a dot), so no agent's |
test/crew-interrupted.test.ts +37 −1
| @@ -19,7 +19,8 @@ import { | ||
| 19 | 19 | } from "../src/core/runs.ts"; |
| 20 | 20 | import { listPulls } from "../src/core/pulls.ts"; |
| 21 | 21 | import { getAgent } from "../src/core/agents.ts"; |
| 22 | import { workdirFor } from "../src/crew/workdir.ts"; | |
| 22 | import { readRunLog } from "../src/runs/log.ts"; | |
| 23 | import { claimWorkdir, isClaimed, workdirFor } from "../src/crew/workdir.ts"; | |
| 23 | 24 | import { STALE_AFTER_MS } from "../src/runs/lease.ts"; |
| 24 | 25 | import { createUser, type User } from "../src/core/users.ts"; |
| 25 | 26 | import { startCrew, type CrewHandle } from "../src/crew/dispatch.ts"; |
| @@ -468,6 +469,41 @@ echo '{"type":"result","subtype":"success","is_error":false,"result":"Done.","to | ||
| 468 | 469 | expect(existsSync(lock)).toBe(false); |
| 469 | 470 | }, 60_000); |
| 470 | 471 | |
| 472 | test("a leftover directory with files and no .git never fails the run (#8219)", async () => { | |
| 473 | const { repo, clone } = await keptClone("bramble"); | |
| 474 | rmSync(join(clone, ".git"), { recursive: true, force: true }); | |
| 475 | const next = await followUp(repo); | |
| 476 | expect(next.summary ?? "").not.toContain("clone failed"); | |
| 477 | expect(next.state).toBe("ok"); | |
| 478 | expect(existsSync(join(clone, ".git", "HEAD"))).toBe(true); | |
| 479 | }, 60_000); | |
| 480 | ||
| 481 | test("a run that finds the issue's clone held works in a clone of its own (#8219)", async () => { | |
| 482 | const { repo, clone } = await keptClone("orchard-held"); | |
| 483 | // Another live run holds the clone, and has uncommitted work in it. | |
| 484 | const other = claimWorkdir(clone, "otherrun"); | |
| 485 | writeFileSync(join(clone, "held.txt"), "mine"); | |
| 486 | const next = await followUp(repo); | |
| 487 | expect(next.state).toBe("ok"); | |
| 488 | expect(readRunLog(app.config.dataDir, next.id).lines.some((l) => l.includes("another run holds the issue's clone"))).toBe(true); | |
| 489 | // The held clone is untouched (a reset and clean would have taken the file)… | |
| 490 | expect(existsSync(join(clone, "held.txt"))).toBe(true); | |
| 491 | // …and the run's own clone is gone with it. | |
| 492 | expect(readdirSync(dirname(clone)).some((n) => n.startsWith(`${basename(clone)}.run-`))).toBe(false); | |
| 493 | await other.release(); | |
| 494 | }, 60_000); | |
| 495 | ||
| 496 | test("a run that throws after claiming its clone frees the claim (#8219)", async () => { | |
| 497 | const { repo, clone } = await keptClone("orchard-throw"); | |
| 498 | // A file where the clone should be: reading it as a directory throws. | |
| 499 | rmSync(clone, { recursive: true, force: true }); | |
| 500 | writeFileSync(clone, "not a directory"); | |
| 501 | const next = await followUp(repo); | |
| 502 | expect(next.state).toBe("failed"); | |
| 503 | expect(isClaimed(clone)).toBe(false); | |
| 504 | rmSync(clone, { force: true }); | |
| 505 | }, 60_000); | |
| 506 | ||
| 471 | 507 | test("a kept clone git no longer reads is moved aside and cloned afresh (#6263)", async () => { |
| 472 | 508 | const { repo, clone } = await keptClone("hedgerow"); |
| 473 | 509 | // What a clone cut short by a network blip leaves: `.git` without its HEAD. |
test/crew-workdir-claim.test.ts added +32
| @@ -0,0 +1,32 @@ | ||
| 1 | import { expect, test } from "bun:test"; | |
| 2 | import { existsSync, mkdirSync, mkdtempSync } from "node:fs"; | |
| 3 | import { tmpdir } from "node:os"; | |
| 4 | import { join } from "node:path"; | |
| 5 | import { claimWorkdir } from "../src/crew/workdir.ts"; | |
| 6 | ||
| 7 | const base = () => join(mkdtempSync(join(tmpdir(), "claim-")), "7651"); | |
| 8 | ||
| 9 | test("a run takes the issue's clone when nobody holds it", async () => { | |
| 10 | const b = base(); | |
| 11 | const a = claimWorkdir(b, "run1"); | |
| 12 | expect(a.dir).toBe(b); | |
| 13 | expect(a.own).toBe(false); | |
| 14 | await a.release(); | |
| 15 | expect(claimWorkdir(b, "run2").dir).toBe(b); | |
| 16 | }); | |
| 17 | ||
| 18 | test("a concurrent run on the same issue gets a directory of its own, removed when it ends (#8219)", async () => { | |
| 19 | const b = base(); | |
| 20 | const first = claimWorkdir(b, "run1"); | |
| 21 | const second = claimWorkdir(b, "run2"); | |
| 22 | expect(second.dir).not.toBe(first.dir); | |
| 23 | expect(second.own).toBe(true); | |
| 24 | const third = claimWorkdir(b, "run3"); | |
| 25 | expect(third.dir).not.toBe(second.dir); | |
| 26 | mkdirSync(second.dir, { recursive: true }); | |
| 27 | await second.release(); | |
| 28 | expect(existsSync(second.dir)).toBe(false); | |
| 29 | await third.release(); | |
| 30 | await first.release(); | |
| 31 | expect(claimWorkdir(b, "run4").dir).toBe(b); | |
| 32 | }); |
test/orphan-clones.test.ts +12
| @@ -23,4 +23,16 @@ describe("sweeping clones nothing claims", () => { | ||
| 23 | 23 | expect(existsSync(join(root, ".cache/r1/bun"))).toBe(true); |
| 24 | 24 | expect(existsSync(join(root, "scout/r1/notes"))).toBe(true); |
| 25 | 25 | }); |
| 26 | ||
| 27 | test("clones moved aside and a crashed run's own clone go; a held one stays (#8219)", async () => { | |
| 28 | const root = mkdtempSync(join(process.env.TMPDIR ?? tmpdir(), "orphans-")); | |
| 29 | const names = ["10.broken-1730000000000", "10.run-abc", "10.run-live"]; | |
| 30 | for (const n of names) mkdirSync(join(root, "scout/r1", n), { recursive: true }); | |
| 31 | const live = join(root, "scout/r1/10.run-live"); | |
| 32 | const swept = await sweepOrphanClones(root, () => true, (d) => d === live); | |
| 33 | expect(swept).toBe(2); | |
| 34 | expect(existsSync(join(root, "scout/r1", names[0]!))).toBe(false); | |
| 35 | expect(existsSync(join(root, "scout/r1", names[1]!))).toBe(false); | |
| 36 | expect(existsSync(live)).toBe(true); | |
| 37 | }); | |
| 26 | 38 | }); |
new above
offline
Claim is taken outside the try/finally that releases it. If mintRunToken throws below, the claim is never released and the issue's clone looks permanently held. Move the claim inside the try or release on every exit.