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 #11574

Parked scoutcrew opened this
ConversationFiles
Closed without merging

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

  • Closed without merging. Nothing was applied to main.

Review

Sign in to review this change.
6 files changed +212 −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 +151
@@ -0,0 +1,151 @@
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(45_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();
47 checks = undefined;
48 servers.stop();
49 app.close();
50 await rm(dataDir, { recursive: true, force: true });
51 });
52
53 /** A crew run that is going: queued, then claimed. */
54 function crewRun(): Run {
55 const run = createRun(app.db, { repoId: repo.id, kind: "agent", reason: "pick" });
56 const claimed = claimRun(app.db, run.id);
57 if (!claimed) throw new Error("not claimed");
58 return claimed;
59 }
60
61 const checkRuns = () => listRuns(app.db, repo.id, { kind: "check", limit: 10 });
62
63 async function until(ok: () => boolean, ms = 20_000): Promise<void> {
64 const end = Date.now() + ms;
65 while (!ok() && Date.now() < end) await Bun.sleep(20);
66 }
67
68 async function push(): Promise<void> {
69 const c = await commitToBranch(app, user, repo, {
70 message: "Declare a check",
71 files: [{ path: "forge.json", content: '{"check": {"run": "exit 0", "install": ""}}\n' }],
72 });
73 if (!c.ok) throw new Error(JSON.stringify(c.error));
74 }
75
76 test("the small size is the one with one CPU", () => {
77 expect(SIZES.small.cpu).toBe(1);
78 expect(SIZES.medium.cpu).toBeGreaterThanOrEqual(2);
79 });
80
81 describe("one CPU", () => {
82 test("a check is held while a crew run is active, then started when it ends", async () => {
83 provideCpusForTest(SIZES.small.cpu);
84 const crew = crewRun();
85 expect(heldForCrew(app)).toBe(true);
86 checks = startChecks(app, { forgeUrl: servers.url, concurrency: 1 });
87 await push();
88 await until(() => checkRuns().length > 0);
89 expect(checkRuns()[0]!.state).toBe("queued");
90 await Bun.sleep(1500);
91 expect(checkRuns()[0]!.state).toBe("queued");
92 cancelRun(app.db, crew.id, "done");
93 expect(heldForCrew(app)).toBe(false);
94 await until(() => checkRuns()[0]!.state !== "queued");
95 expect(checkRuns()[0]!.state).not.toBe("queued");
96 });
97
98 test("a look (or snapshot) waits on crewIdle until the run ends", async () => {
99 provideCpusForTest(SIZES.small.cpu);
100 const crew = crewRun();
101 let started = false;
102 const waiting = crewIdle(app, 10).then(() => (started = true));
103 await Bun.sleep(100);
104 expect(started).toBe(false);
105 cancelRun(app.db, crew.id, "done");
106 await waiting;
107 expect(started).toBe(true);
108 });
109
110 test("a second browser waits for the first", async () => {
111 provideCpusForTest(SIZES.small.cpu);
112 let open = 0;
113 let most = 0;
114 const one = async () => {
115 open++;
116 most = Math.max(most, open);
117 await Bun.sleep(30);
118 open--;
119 };
120 await Promise.all([oneBrowser(one), oneBrowser(one), oneBrowser(one)]);
121 expect(most).toBe(1);
122 });
123 });
124
125 describe("two CPUs", () => {
126 test("nothing is held, and browsers overlap", async () => {
127 provideCpusForTest(SIZES.medium.cpu);
128 crewRun();
129 expect(heldForCrew(app)).toBe(false);
130 await crewIdle(app, 10);
131 let open = 0;
132 let most = 0;
133 const one = async () => {
134 open++;
135 most = Math.max(most, open);
136 await Bun.sleep(30);
137 open--;
138 };
139 await Promise.all([oneBrowser(one), oneBrowser(one)]);
140 expect(most).toBe(2);
141 });
142
143 test("a check starts at once beside a crew run", async () => {
144 provideCpusForTest(SIZES.medium.cpu);
145 crewRun();
146 checks = startChecks(app, { forgeUrl: servers.url, concurrency: 1 });
147 await push();
148 await until(() => checkRuns().some((r) => r.state !== "queued"));
149 expect(checkRuns()[0]!.state).not.toBe("queued");
150 });
151 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs