trevato/forge

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

More
Forks1g fMapg aRunsg rEventsFlows

On a 1-CPU forge the crew run has the core: looks and checks wait for it, and one browser runs at a time #11584

Parked scoutcrew opened this
ConversationFiles
Closed without merging

scout/forge:scout/11561-on-a-1-cpu-forge-the-crew-run-has-the-co → main 6 commits fork

  • Closed without merging. Nothing was applied to main.

Review

Sign in to review this change.
6 files changed +220 −1
src/checks/service.ts +12
@@ -11,6 +11,7 @@
1111 * is that command's business.
1212 */
1313
14 import { heldForCrew, RECHECK_MS } from "../core/one_core.ts";
1415 import { pullDir, runDir } from "../core/pulls.ts";
1516 import { cellHides } from "../spawn/hosted-mark.ts";
1617 import { existsSync, lstatSync, mkdirSync, readFileSync, readdirSync } from "node:fs";
@@ -622,8 +623,18 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
622623 return next;
623624 }
624625
626 let heldTimer: ReturnType<typeof setTimeout> | undefined;
625627 function schedule(): void {
626628 if (stopped) return;
629 // One core: the crew's run has it; queued checks start after it ends.
630 if (heldForCrew(app)) {
631 heldTimer ??= setTimeout(() => {
632 heldTimer = undefined;
633 schedule();
634 }, RECHECK_MS);
635 heldTimer.unref?.();
636 return;
637 }
627638 const limit = capacity();
628639 if (options.gate === undefined) gate.resize(limit);
629640 startVerify();
@@ -1751,6 +1762,7 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
17511762 stop: async (stopOptions) => {
17521763 stopped = true;
17531764 clearInterval(timer);
1765 if (heldTimer !== undefined) clearTimeout(heldTimer);
17541766 if (sweeper !== undefined) clearInterval(sweeper);
17551767 unsubscribe();
17561768 unwatch();
src/core/one_core.ts added +40
@@ -0,0 +1,40 @@
1 /**
2 * One core: on a forge whose size has one CPU (`spawn/sizes.ts`), the crew's
3 * run has the core, and a look, a check or a snapshot starts after it, in
4 * order (#11561). A forge with two or more behaves as it always did.
5 */
6
7 import type { App } from "../app.ts";
8 import { cpuLimit } from "./capacity.ts";
9 import { listActive } from "./runs.ts";
10
11 /** Whether this forge has a single CPU to share. */
12 export const oneCore = (): boolean => cpuLimit() <= 1;
13
14 /** Whether a crew run is going now. */
15 export const crewRunning = (app: App): boolean =>
16 listActive(app.db, "agent").some((run) => run.state === "running");
17
18 /** Whether background work waits: one core, and the crew is using it. */
19 export const heldForCrew = (app: App): boolean => oneCore() && crewRunning(app);
20
21 /** How often a held start looks again. */
22 export const RECHECK_MS = 1000;
23
24 /** Resolves once nothing is held for the crew; at once on a forge with room. */
25 export async function crewIdle(app: App, pollMs: number = RECHECK_MS): Promise<void> {
26 while (heldForCrew(app)) await Bun.sleep(pollMs);
27 }
28
29 let tail: Promise<unknown> = Promise.resolve();
30
31 /**
32 * Runs `fn` alone on a one-core forge: a second browser waits for the first
33 * to end. With more cores it runs at once.
34 */
35 export function oneBrowser<T>(fn: () => Promise<T>): Promise<T> {
36 if (!oneCore()) return fn();
37 const run = tail.catch(() => {}).then(fn);
38 tail = run;
39 return run;
40 }
src/looks/capture.ts +2 −1
@@ -34,6 +34,7 @@
3434 * process starts, and a pod that holds two agent runs has room for one.
3535 */
3636
37 import { oneBrowser } from "../core/one_core.ts";
3738 import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
3839 import { tmpdir } from "node:os";
3940 import { join } from "node:path";
@@ -118,7 +119,7 @@ export async function takeLook(
118119 return err(NO_BROWSER);
119120 const release = await gate.acquire();
120121 try {
121 return await shoot(binary, input);
122 return await oneBrowser(() => shoot(binary, input));
122123 } finally {
123124 release();
124125 }
src/looks/service.ts +4
@@ -73,6 +73,7 @@ import { keepAsMain } from "./main_looks.ts";
7373 import { findChromium, isNoBrowser, takeLook, type Capture } from "./capture.ts";
7474 import { backoff, type BackoffClock } from "./backoff.ts";
7575 import { during } from "../perf/jobs.ts";
76 import { crewIdle } from "../core/one_core.ts";
7677
7778 const log = createLog("looks");
7879
@@ -365,6 +366,9 @@ export function startLooks(app: App, options: LooksOptions): LooksHandle {
365366 await Bun.sleep(15_000);
366367 continue;
367368 }
369 // One core: the crew's run has it, and the look waits its end.
370 await crewIdle(app);
371 if (stopped) return;
368372 try {
369373 await take(next);
370374 } catch (error) {
src/snapshot/service.ts +3
@@ -10,6 +10,7 @@
1010 * is within the hour, not the next night.
1111 */
1212
13 import { crewIdle } from "../core/one_core.ts";
1314 import type { App } from "../app.ts";
1415 import { createLog } from "../kernel/log.ts";
1516 import { firstSeen, reportVerdict, seedFirstSeen } from "./verdict.ts";
@@ -185,6 +186,8 @@ export function startSnapshots(
185186 // very next snapshot's.
186187 const store = current();
187188 if (store === undefined) throw new Nowhere();
189 // One core: the crew's run has it, and the snapshot starts after.
190 await crewIdle(app);
188191 offBox = isOffBox(store);
189192 running = { objects: 0, bytes: 0, elapsedMs: 0 };
190193 startedAt = Date.now();
test/one-core.test.ts added +159
@@ -0,0 +1,159 @@
1 /**
2 * One core (#11561): on a 1-CPU forge a crew run has the core, and a check,
3 * a look and a snapshot start after it; a 2-CPU forge starts them at once.
4 */
5 import { setDefaultTimeout } from "bun:test";
6 import { afterEach, beforeEach, describe, expect, test } from "bun:test";
7 import { mkdtemp, rm } from "node:fs/promises";
8 import { tmpdir } from "node:os";
9 import { join } from "node:path";
10 import { createApp, type App } from "../src/app.ts";
11 import { commitToBranch, createRepo } from "../src/actions/repos.ts";
12 import { startChecks, type ChecksHandle } from "../src/checks/service.ts";
13 import { provideCpusForTest } from "../src/core/capacity.ts";
14 import { crewIdle, heldForCrew, oneBrowser } from "../src/core/one_core.ts";
15 import { cancelRun, claimRun, createRun, listRuns, type Run } from "../src/core/runs.ts";
16 import type { Repo } from "../src/core/repos.ts";
17 import { createUser, type User } from "../src/core/users.ts";
18 import { serve, type Servers } from "../src/http/server.ts";
19 import { SIZES } from "../src/spawn/sizes.ts";
20
21 let dataDir: string;
22 let app: App;
23 let servers: Servers;
24 let checks: ChecksHandle | undefined;
25 let user: User;
26 let repo: Repo;
27
28 setDefaultTimeout(150_000);
29
30 beforeEach(async () => {
31 dataDir = await mkdtemp(join(tmpdir(), "forge-onecore-"));
32 app = createApp({ dataDir, port: 0 });
33 servers = await serve(app);
34 const a = await createUser(app.db, {
35 origin: "seed", trusted: true, username: "alice", email: "[email protected]",
36 password: "a-good-passphrase", adminIfFirst: true,
37 });
38 if (!a.ok) throw new Error(a.error);
39 user = a.value;
40 const made = await createRepo(app, user, { name: "lib", visibility: "public" });
41 if (!made.ok) throw new Error(String(made.error));
42 repo = made.value;
43 });
44 afterEach(async () => {
45 provideCpusForTest(undefined);
46 await checks?.stop({ drain: false });
47 await Bun.sleep(200);
48 checks = undefined;
49 servers.stop();
50 app.close();
51 await rm(dataDir, { recursive: true, force: true });
52 });
53
54 /** A crew run that is going: queued, then claimed. */
55 function crewRun(): Run {
56 const run = createRun(app.db, { repoId: repo.id, kind: "agent", reason: "pick" });
57 const claimed = claimRun(app.db, run.id);
58 if (!claimed) throw new Error("not claimed");
59 return claimed;
60 }
61
62 const checkRuns = () => listRuns(app.db, repo.id, { kind: "check", limit: 10 });
63
64 async function until(ok: () => boolean, ms = 100_000): Promise<void> {
65 const end = Date.now() + ms;
66 const done = () => {
67 try {
68 return ok();
69 } catch {
70 return true; // the app closed under us: the test is over
71 }
72 };
73 while (!done() && Date.now() < end) await Bun.sleep(20);
74 }
75
76 async function push(): Promise<void> {
77 const c = await commitToBranch(app, user, repo, {
78 message: "Declare a check",
79 files: [{ path: "forge.json", content: '{"check": {"run": "exit 0", "install": ""}}\n' }],
80 });
81 if (!c.ok) throw new Error(JSON.stringify(c.error));
82 }
83
84 test("the small size is the one with one CPU", () => {
85 expect(SIZES.small.cpu).toBe(1);
86 expect(SIZES.medium.cpu).toBeGreaterThanOrEqual(2);
87 });
88
89 describe("one CPU", () => {
90 test("a check is held while a crew run is active, then started when it ends", async () => {
91 provideCpusForTest(SIZES.small.cpu);
92 const crew = crewRun();
93 expect(heldForCrew(app)).toBe(true);
94 checks = startChecks(app, { forgeUrl: servers.url, concurrency: 1 });
95 await push();
96 await until(() => checkRuns().length > 0);
97 expect(checkRuns()[0]!.state).toBe("queued");
98 await Bun.sleep(1500);
99 expect(checkRuns()[0]!.state).toBe("queued");
100 cancelRun(app.db, crew.id, "done");
101 expect(heldForCrew(app)).toBe(false);
102 await until(() => checkRuns()[0]!.state !== "queued");
103 expect(checkRuns()[0]!.state).not.toBe("queued");
104 });
105
106 test("a look (or snapshot) waits on crewIdle until the run ends", async () => {
107 provideCpusForTest(SIZES.small.cpu);
108 const crew = crewRun();
109 let started = false;
110 const waiting = crewIdle(app, 10).then(() => (started = true));
111 await Bun.sleep(100);
112 expect(started).toBe(false);
113 cancelRun(app.db, crew.id, "done");
114 await waiting;
115 expect(started).toBe(true);
116 });
117
118 test("a second browser waits for the first", async () => {
119 provideCpusForTest(SIZES.small.cpu);
120 let open = 0;
121 let most = 0;
122 const one = async () => {
123 open++;
124 most = Math.max(most, open);
125 await Bun.sleep(30);
126 open--;
127 };
128 await Promise.all([oneBrowser(one), oneBrowser(one), oneBrowser(one)]);
129 expect(most).toBe(1);
130 });
131 });
132
133 describe("two CPUs", () => {
134 test("nothing is held, and browsers overlap", async () => {
135 provideCpusForTest(SIZES.medium.cpu);
136 crewRun();
137 expect(heldForCrew(app)).toBe(false);
138 await crewIdle(app, 10);
139 let open = 0;
140 let most = 0;
141 const one = async () => {
142 open++;
143 most = Math.max(most, open);
144 await Bun.sleep(30);
145 open--;
146 };
147 await Promise.all([oneBrowser(one), oneBrowser(one)]);
148 expect(most).toBe(2);
149 });
150
151 test("a check starts at once beside a crew run", async () => {
152 provideCpusForTest(SIZES.medium.cpu);
153 crewRun();
154 checks = startChecks(app, { forgeUrl: servers.url, concurrency: 1 });
155 await push();
156 await until(() => checkRuns().some((r) => r.state !== "queued"));
157 expect(checkRuns()[0]!.state).not.toBe("queued");
158 });
159 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs