trevato/forge

Public: anyone can read this; the crew can change it.

More
Forks1g fMapg aRuns12g rEventsFlows

Sentinel: disk was degraded 49 minutes today: Disk is nearly full: 10% free. #5730

Merged scoutcrew opened this

Check · Review · Land · Live done — Landed and live in production.

ConversationFiles
Merged

scout/forge:scout/5697-sentinel-disk-was-degraded-49-minutes-to → main 2 commits fork

  • Merged. scout/5697-sentinel-disk-was-degraded-49-minutes-to was merged into main as 10e0443.

Review

Sign in to review this change.
5 files changed +301 −0
src/crew/dispatch.ts +13
@@ -28,6 +28,8 @@ import {
2828 type Issue,
2929 } from "../core/issues.ts";
3030 import { getPull, type Pull } from "../core/pulls.ts";
31 import { claimedClones, sweepOrphanClones } from "./orphans.ts";
32 import { crewRoot } from "./workdir.ts";
3133 import { resolveOpen } from "../core/decisions.ts";
3234 import { accountActive } from "../core/accounts.ts";
3335 import { runningLimitSentence, runningRuns } from "../settings/quotas.ts";
@@ -218,6 +220,16 @@ export async function startCrew(
218220 }
219221 pump(app, ready, inFlight);
220222 }, options.tickMs ?? TICK_MS);
223 // Clones no open issue, pull or run claims fill the volume (#5697).
224 const sweepClones = (): void => {
225 sweepOrphanClones(crewRoot(app.config.dataDir), claimedClones(app.db))
226 .then((count) => {
227 if (count > 0) log.info("removed clones nothing claims", { count });
228 })
229 .catch((error) => log.warn("clone sweep failed", { error: String(error) }));
230 };
231 sweepClones();
232 const cloneTimer = setInterval(() => !stopped && sweepClones(), 60 * 60_000);
221233 requeueInterrupted(app);
222234 sweepAssigned(app);
223235 pump(app, ready, inFlight);
@@ -229,6 +241,7 @@ export async function startCrew(
229241 stop: async (options = {}) => {
230242 stopped = true;
231243 clearInterval(timer);
244 clearInterval(cloneTimer);
232245 unsubscribe();
233246 unfollow();
234247 unsettle();
src/crew/orphans.ts added +72
@@ -0,0 +1,72 @@
1 /**
2 * Clones nothing waits for (#5697). A clone lives for as long as its pull is
3 * open, and the runner and the dispatcher remove it when the pull lands or
4 * closes. A run that ended without a pull, a forge that was down when the
5 * pull closed, or a clone that would not go leaves one behind, and the state
6 * volume fills (`Disk is nearly full`). This is the net under them: it lists
7 * `<crew>/<agent>/<repo id>/<n>/` and removes those `keep` does not claim.
8 */
9
10 import { readdirSync } from "node:fs";
11 import { rm } from "node:fs/promises";
12 import { join } from "node:path";
13 import type { Database } from "bun:sqlite";
14 import { getIssue } from "../core/issues.ts";
15 import { listPulls } from "../core/pulls.ts";
16 import { listActive } from "../core/runs.ts";
17
18 const dirs = (dir: string): string[] => {
19 try {
20 return readdirSync(dir, { withFileTypes: true })
21 .filter((e) => e.isDirectory() && !e.name.startsWith("."))
22 .map((e) => e.name);
23 } catch {
24 return [];
25 }
26 };
27
28 /** Remove every clone `keep(agent, repoId, number)` is false for; how many went. */
29 export async function sweepOrphanClones(
30 crewDir: string,
31 keep: (agent: string, repoId: string, number: number) => boolean,
32 ): Promise<number> {
33 let swept = 0;
34 for (const agent of dirs(crewDir)) {
35 for (const repoId of dirs(join(crewDir, agent))) {
36 for (const name of dirs(join(crewDir, agent, repoId))) {
37 if (!/^[1-9]\d*$/.test(name)) continue;
38 if (keep(agent, repoId, Number(name))) continue;
39 try {
40 await rm(join(crewDir, agent, repoId, name), { recursive: true, force: true });
41 swept++;
42 } catch {
43 // a leftover process still writes there; the next sweep tries again
44 }
45 }
46 }
47 }
48 return swept;
49 }
50
51 /** The task number a branch `<agent>/<n>-…` names, if it does. */
52 function taskOf(agent: string, branch: string): number | undefined {
53 if (!branch.startsWith(`${agent}/`)) return undefined;
54 const m = /^(\d+)(?:-|$)/.exec(branch.slice(agent.length + 1));
55 return m === null ? undefined : Number(m[1]);
56 }
57
58 /**
59 * What still claims a clone: its issue is open, a run is queued or working
60 * on it, or an open pull of the agent's branch for it. `repoId` is the
61 * directory name `workdirFor` gives the repository: the db's `repo.id`.
62 */
63 export function claimedClones(db: Database): (agent: string, repoId: string, number: number) => boolean {
64 return (agent, repoId, number) => {
65 // Read per clone, not once per sweep: the sweep awaits many removals, and
66 // a run queued meanwhile for a just-closed issue must keep its clone.
67 const active = listActive(db, "agent");
68 if (getIssue(db, repoId, number)?.state === "open") return true;
69 if (active.some((r) => r.repoId === repoId && taskOf(agent, r.branch) === number)) return true;
70 return listPulls(db, repoId, { state: "open" }).some((p) => taskOf(agent, p.headBranch) === number);
71 };
72 }
test/orphan-claims.test.ts added +137
@@ -0,0 +1,137 @@
1 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
2 import { existsSync, mkdirSync, mkdtempSync } from "node:fs";
3 import { rm } from "node:fs/promises";
4 import { tmpdir } from "node:os";
5 import { join } from "node:path";
6 import { createApp, type App } from "../src/app.ts";
7 import { createRepo } from "../src/actions/repos.ts";
8 import { createIssue, setIssueState } from "../src/core/issues.ts";
9 import { claimRun, createRun, finishRun } from "../src/core/runs.ts";
10 import { createUser, type User } from "../src/core/users.ts";
11 import { claimedClones, sweepOrphanClones } from "../src/crew/orphans.ts";
12 import { workdirFor } from "../src/crew/runner.ts";
13 import { crewRoot } from "../src/crew/workdir.ts";
14 import type { Repo } from "../src/core/repos.ts";
15 import type { Agent } from "../src/core/agents.ts";
16
17 describe("what claims a clone", () => {
18 let dataDir: string;
19 let app: App;
20 let alice: User;
21 let repo: Repo;
22 let claimed: (agent: string, repoId: string, n: number) => boolean;
23
24 const issue = (closed = false): number => {
25 const made = createIssue(app.db, { repoId: repo.id, authorId: alice.id, title: "t" });
26 if (!made.ok) throw new Error(made.error);
27 if (closed) setIssueState(app.db, made.value.id, "closed");
28 return made.value.number;
29 };
30 const run = (branch: string) =>
31 createRun(app.db, { repoId: repo.id, kind: "agent", branch, reason: "assigned" });
32 const pull = (headBranch: string, closed = false): void => {
33 const made = createIssue(app.db, {
34 repoId: repo.id,
35 authorId: alice.id,
36 title: "p",
37 kind: "pull",
38 });
39 if (!made.ok) throw new Error(made.error);
40 app.db
41 .query(
42 `INSERT INTO pull (issue_id, base_branch, head_branch, base_oid, head_oid, head_repo_id)
43 VALUES (?1, 'main', ?2, 'a', 'b', ?3)`,
44 )
45 .run(made.value.id, headBranch, repo.id);
46 if (closed) setIssueState(app.db, made.value.id, "closed");
47 };
48
49 beforeAll(async () => {
50 dataDir = mkdtempSync(join(process.env.TMPDIR ?? tmpdir(), "claims-"));
51 app = createApp({ dataDir, port: 0 });
52 const a = await createUser(app.db, {
53 origin: "seed",
54 trusted: true,
55 username: "alice",
56 email: "[email protected]",
57 password: "a-good-passphrase",
58 adminIfFirst: true,
59 });
60 if (!a.ok) throw new Error(a.error);
61 alice = a.value;
62 const made = await createRepo(app, alice, { name: "garden", visibility: "private" });
63 if (!made.ok) throw new Error(String(made.error));
64 repo = made.value;
65 claimed = claimedClones(app.db);
66 });
67 afterAll(async () => {
68 await rm(dataDir, { recursive: true, force: true });
69 });
70
71 test("the clone directory is named by the db repo id", () => {
72 const dir = workdirFor(dataDir, { user: { username: "bot" } } as Agent, repo, 7);
73 expect(dir).toBe(join(crewRoot(dataDir), "bot", repo.id, "7"));
74 });
75
76 test("an open issue keeps its clone", () => {
77 const n = issue();
78 expect(claimed("bot", repo.id, n)).toBe(true);
79 });
80
81 test("a closed issue with no run and no pull does not", () => {
82 const n = issue(true);
83 expect(claimed("bot", repo.id, n)).toBe(false);
84 });
85
86 test("a queued or working run on the agent's branch keeps it, a finished one does not", () => {
87 const n = issue(true);
88 const r = run(`bot/${n}-plant-tulips`);
89 expect(claimed("bot", repo.id, n)).toBe(true);
90 claimRun(app.db, r.id);
91 expect(claimed("bot", repo.id, n)).toBe(true);
92 finishRun(app.db, r.id, { state: "ok", summary: "done" });
93 expect(claimed("bot", repo.id, n)).toBe(false);
94 });
95
96 test("an open pull from the agent's branch keeps it, a closed one does not", () => {
97 const n = issue(true);
98 pull(`bot/${n}-fix`);
99 expect(claimed("bot", repo.id, n)).toBe(true);
100 const m = issue(true);
101 pull(`bot/${m}-fix`, true);
102 expect(claimed("bot", repo.id, m)).toBe(false);
103 });
104
105 test("another agent's branch, a different number or another repo does not claim it", () => {
106 const n = issue(true);
107 run(`other/${n}-x`);
108 pull(`other/${n}-x`);
109 run(`bot/${n}0-x`);
110 pull(`bot/${n}0-x`);
111 expect(claimed("bot", repo.id, n)).toBe(false);
112 expect(claimed("bot", "some-other-repo", n)).toBe(false);
113 const k = issue(true);
114 pull(`bot/${k}-x`);
115 expect(claimed("bot", "some-other-repo", k)).toBe(false);
116 });
117
118 test("a run queued after the closure is made still keeps its clone", async () => {
119 const n = issue(true);
120 const dir = join(crewRoot(dataDir), "bot", repo.id, String(n));
121 mkdirSync(dir, { recursive: true });
122 run(`bot/${n}-late`);
123 // `claimed` was built before the run existed
124 expect(await sweepOrphanClones(crewRoot(dataDir), claimed)).toBe(0);
125 expect(existsSync(dir)).toBe(true);
126 });
127
128 test("swept against the db, only the unclaimed directory goes", async () => {
129 const open = issue();
130 const dead = issue(true);
131 const root = crewRoot(dataDir);
132 for (const n of [open, dead]) mkdirSync(join(root, "bot", repo.id, String(n)), { recursive: true });
133 await sweepOrphanClones(root, claimed);
134 expect(existsSync(join(root, "bot", repo.id, String(open)))).toBe(true);
135 expect(existsSync(join(root, "bot", repo.id, String(dead)))).toBe(false);
136 });
137 });
test/orphan-clones.test.ts added +26
@@ -0,0 +1,26 @@
1 import { describe, expect, test } from "bun:test";
2 import { existsSync, mkdirSync, mkdtempSync, writeFileSync } from "node:fs";
3 import { tmpdir } from "node:os";
4 import { join } from "node:path";
5 import { sweepOrphanClones } from "../src/crew/orphans.ts";
6
7 describe("sweeping clones nothing claims", () => {
8 test("only the unclaimed clones go; the cache and odd names stay", async () => {
9 const root = mkdtempSync(join(process.env.TMPDIR ?? tmpdir(), "orphans-"));
10 for (const p of ["scout/r1/10", "scout/r1/11", "scout/r2/12", ".cache/r1/bun", "scout/r1/notes"])
11 mkdirSync(join(root, p), { recursive: true });
12 writeFileSync(join(root, "scout/r1/11/f"), "x");
13 const seen: string[] = [];
14 const swept = await sweepOrphanClones(root, (agent, repo, n) => {
15 seen.push(`${agent}/${repo}/${n}`);
16 return n === 10;
17 });
18 expect(swept).toBe(2);
19 expect(seen.sort()).toEqual(["scout/r1/10", "scout/r1/11", "scout/r2/12"]);
20 expect(existsSync(join(root, "scout/r1/10"))).toBe(true);
21 expect(existsSync(join(root, "scout/r1/11"))).toBe(false);
22 expect(existsSync(join(root, "scout/r2/12"))).toBe(false);
23 expect(existsSync(join(root, ".cache/r1/bun"))).toBe(true);
24 expect(existsSync(join(root, "scout/r1/notes"))).toBe(true);
25 });
26 });
test/orphan-dispatch.test.ts added +53
@@ -0,0 +1,53 @@
1 import { describe, expect, test } from "bun:test";
2 import { existsSync, mkdirSync, mkdtempSync } from "node:fs";
3 import { rm } from "node:fs/promises";
4 import { tmpdir } from "node:os";
5 import { join } from "node:path";
6 import { createApp } from "../src/app.ts";
7 import { startCrew } from "../src/crew/dispatch.ts";
8
9 describe("the crew sweeps orphan clones", () => {
10 test("at start it removes a leftover, and stop() clears the hourly timer", async () => {
11 const dataDir = mkdtempSync(join(process.env.TMPDIR ?? tmpdir(), "orphan-crew-"));
12 const orphan = join(dataDir, "crew", "ghost", "no-such-repo", "42");
13 mkdirSync(orphan, { recursive: true });
14 const app = createApp({ dataDir, port: 0 });
15
16 const set = globalThis.setInterval;
17 const clear = globalThis.clearInterval;
18 const hourly: unknown[] = [];
19 const cleared: unknown[] = [];
20 globalThis.setInterval = ((fn: () => void, ms?: number, ...rest: unknown[]) => {
21 const t = (set as (...a: unknown[]) => unknown)(fn, ms, ...rest);
22 if (ms === 60 * 60_000) hourly.push(t);
23 return t;
24 }) as typeof setInterval;
25 globalThis.clearInterval = ((t: unknown) => {
26 cleared.push(t);
27 return (clear as (t: unknown) => void)(t);
28 }) as typeof clearInterval;
29 let crew;
30 try {
31 crew = await startCrew(app, {
32 forgeUrl: "http://127.0.0.1:1",
33 localUrl: "http://127.0.0.1:1",
34 binary: "/nonexistent/claude",
35 concurrency: 1,
36 budgetUsd: 1,
37 runTimeoutMs: 60_000,
38 networkRetryScale: 0,
39 reviewBudgetMs: 1000,
40 });
41 for (let i = 0; i < 100 && existsSync(orphan); i++) await Bun.sleep(20);
42 expect(existsSync(orphan)).toBe(false);
43 expect(hourly.length).toBe(1);
44 await crew.stop();
45 expect(cleared).toContain(hourly[0]);
46 } finally {
47 globalThis.setInterval = set;
48 globalThis.clearInterval = clear;
49 await crew?.stop();
50 await rm(dataDir, { recursive: true, force: true });
51 }
52 });
53 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs