Public: anyone can read this; the crew can change it.
Crew runs at once defaults to what the pod's memory holds, as children.max does, so a big machine needs no override #2550
Check · Review · Land · Live done — Landed and live in production.
torque/forge:torque/2548-crew-runs-at-once-defaults-to-what-the-p → main 2 commits fork 1 thread · 1 unresolved
8 files
changed +149
−9
README.md +1 −1
| @@ -870,7 +870,7 @@ same breakdown does not ask it twice. | ||
| 870 | 870 | The forge asks the host for one more binary: `claude`, on `PATH` or at |
| 871 | 871 | `FORGE_CLAUDE`, authenticated as the operator (an `ANTHROPIC_API_KEY` in the |
| 872 | 872 | forge's environment is handed to runs; the machine's own Claude Code login |
| 873 | works when there is none). `FORGE_CREW_CONCURRENCY` (2) and | |
| 873 | works when there is none). `FORGE_CREW_CONCURRENCY` (one per 3 GiB above 2 GiB, 2–16) and | |
| 874 | 874 | `FORGE_CREW_BUDGET_USD` (5) bound it; `FORGE_CREW_CEILING_USD` caps what the |
| 875 | 875 | crew may spend across every repository in a UTC day, whatever each |
| 876 | 876 | repository's own ceiling says; `FORGE_APPS=0` turns apps and crew off. A preview whose pull nobody has |
docs/APPS.md +1 −1
| @@ -1114,7 +1114,7 @@ a link to the milestone (`docs/PLANNING.md`); its progress never does. | ||
| 1114 | 1114 | | `FORGE_APP_DOMAIN` | the host in `FORGE_URL`/`_HOST` | what app hostnames attach to (`example` or `--x`) | |
| 1115 | 1115 | | `FORGE_APPS_CONCURRENCY` | `2` | deploys running their heavy steps at once | |
| 1116 | 1116 | | `FORGE_PREVIEW_IDLE_DAYS` | `3` | days a pull may sit untouched before its preview sleeps; `0` = never | |
| 1117 | | `FORGE_CREW_CONCURRENCY` | `2` | agent runs at once (live: Settings → Capacity) | | |
| 1117 | | `FORGE_CREW_CONCURRENCY` | 1 per 3 GiB above 2 GiB, 2–16 | agent runs at once (live: Settings → Capacity) | | |
| 1118 | 1118 | | `FORGE_CHECKS_CONCURRENCY` | `1` | checks at once (live: Settings → Capacity) | |
| 1119 | 1119 | | `FORGE_CHECKS_PLACEMENT` | `process` | `job` runs a check's commands as a cluster Job (see Placement above) | |
| 1120 | 1120 | | `FORGE_CHECKS_DRAIN_S` | `300` | seconds a roll lets running checks finish first | |
src/core/capacity.ts +16 −2
| @@ -43,7 +43,7 @@ export const CAPACITY: Record<CapacityKey, Definition> = { | ||
| 43 | 43 | env: "FORGE_CREW_CONCURRENCY", |
| 44 | 44 | kind: { type: "count", min: 1, max: 16 }, |
| 45 | 45 | fallback: "2", |
| 46 | help: "How many agent runs go at once. Lowering it lets running ones finish; the next one waits.", | |
| 46 | help: "How many agent runs go at once. Empty: one per 3 GiB of memory above the forge's own 2 GiB, from the pod's limit, at least 2 and at most 16. Lowering it lets running ones finish; the next one waits.", | |
| 47 | 47 | }, |
| 48 | 48 | "checks.concurrency": { |
| 49 | 49 | label: "Checks at once", |
| @@ -109,7 +109,9 @@ export function validCapacity(key: CapacityKey, value: string): boolean { | ||
| 109 | 109 | export function capacityDefault(key: CapacityKey): string { |
| 110 | 110 | const env = process.env[CAPACITY[key].env]?.trim() ?? ""; |
| 111 | 111 | if (env !== "" && validCapacity(key, env)) return env; |
| 112 | return key === "children.max" ? String(childrenForMemory(memoryLimitBytes())) : CAPACITY[key].fallback; | |
| 112 | if (key === "children.max") return String(childrenForMemory(memoryLimitBytes())); | |
| 113 | if (key === "crew.concurrency") return String(crewForMemory(memoryLimitBytes())); | |
| 114 | return CAPACITY[key].fallback; | |
| 113 | 115 | } |
| 114 | 116 | |
| 115 | 117 | const GIB = 1024 ** 3; |
| @@ -126,6 +128,18 @@ export function childrenForMemory(bytes: number): number { | ||
| 126 | 128 | } |
| 127 | 129 | const MIN_CHILDREN = 5; |
| 128 | 130 | |
| 131 | /** | |
| 132 | * One crew run per 3 GiB of room above the forge's own 2 GiB (32 GiB gives | |
| 133 | * 10, 64 GiB the ceiling of 16), never fewer than two: a big machine runs | |
| 134 | * the crew it can hold without an override the sentinel then flags (#2548). | |
| 135 | */ | |
| 136 | export function crewForMemory(bytes: number): number { | |
| 137 | const n = Math.floor((bytes - 2 * GIB) / (3 * GIB)); | |
| 138 | return Math.min(MAX_CREW, Math.max(MIN_CREW, Number.isFinite(n) ? n : MIN_CREW)); | |
| 139 | } | |
| 140 | const MIN_CREW = 2; | |
| 141 | const MAX_CREW = 16; | |
| 142 | ||
| 129 | 143 | /** The pod's memory limit (cgroup v2, then v1), else the machine's memory. */ |
| 130 | 144 | export function memoryLimitBytes(): number { |
| 131 | 145 | for (const f of ["/sys/fs/cgroup/memory.max", "/sys/fs/cgroup/memory/memory.limit_in_bytes"]) { |
src/main.ts +2 −1
| @@ -55,6 +55,7 @@ import { running } from "./kernel/version.ts"; | ||
| 55 | 55 | import { record as recordDegraded } from "./kernel/degraded_history.ts"; |
| 56 | 56 | import { vitals } from "./kernel/vitals.ts"; |
| 57 | 57 | import { snapshotPlace } from "./actions/health.ts"; |
| 58 | import { capacityDefault } from "./core/capacity.ts"; | |
| 58 | 59 | |
| 59 | 60 | // `forge hook <name>` re-enters here from a git hook stub. It relays to the |
| 60 | 61 | // running forge over its socket and exits — it must never boot a server. |
| @@ -379,7 +380,7 @@ const crew = appsOn && !crewHeldOff | ||
| 379 | 380 | forgeUrl, |
| 380 | 381 | localUrl: servers.url, |
| 381 | 382 | binary: process.env.FORGE_CLAUDE ?? "claude", |
| 382 | concurrency: positive(process.env.FORGE_CREW_CONCURRENCY, hosted ? 1 : 2), | |
| 383 | concurrency: positive(process.env.FORGE_CREW_CONCURRENCY, hosted ? 1 : Number(capacityDefault("crew.concurrency"))), | |
| 383 | 384 | budgetUsd: Number(process.env.FORGE_CREW_BUDGET_USD) || 5, |
| 384 | 385 | runTimeoutMs: crewRunMs, |
| 385 | 386 | allowHosts: (process.env.FORGE_CREW_ALLOW_HOSTS ?? "") |
src/snapshot/restore.ts +10 −3
| @@ -238,6 +238,9 @@ export async function restoreSnapshot( | ||
| 238 | 238 | // hundreds of them spent the drill's minutes here, one after another. |
| 239 | 239 | const git = options.git ?? runGit; |
| 240 | 240 | let next = 0; |
| 241 | // Clones finish in any order; a refusal names the earliest bundle in the | |
| 242 | // manifest that failed, so it reads the same however the clones raced. | |
| 243 | const failed: { at: number; error: string }[] = []; | |
| 241 | 244 | await Promise.all( |
| 242 | 245 | Array.from({ length: Math.min(CLONES, bundles.length) }, async () => { |
| 243 | 246 | for (let b = bundles[next++]; b !== undefined; b = bundles[next++]) { |
| @@ -252,9 +255,11 @@ export async function restoreSnapshot( | ||
| 252 | 255 | ]); |
| 253 | 256 | if (!cloned.ok) { |
| 254 | 257 | next = bundles.length; |
| 255 | throw new Error( | |
| 256 | `could not restore ${b.owner}/${b.name}: ${cloned.error.stderr}`, | |
| 257 | ); | |
| 258 | failed.push({ | |
| 259 | at: b.at, | |
| 260 | error: `could not restore ${b.owner}/${b.name}: ${cloned.error.stderr}`, | |
| 261 | }); | |
| 262 | return; | |
| 258 | 263 | } |
| 259 | 264 | // A mirror clone remembers the bundle as its origin; the forge's |
| 260 | 265 | // repositories have no origin of their own. |
| @@ -265,6 +270,8 @@ export async function restoreSnapshot( | ||
| 265 | 270 | } |
| 266 | 271 | }), |
| 267 | 272 | ); |
| 273 | if (failed.length > 0) | |
| 274 | throw new Error(failed.sort((a, b) => a.at - b.at)[0]!.error); | |
| 268 | 275 | const { rmSync } = await import("node:fs"); |
| 269 | 276 | rmSync(join(dataDir, "restore-bundles"), { recursive: true, force: true }); |
| 270 | 277 | if (key !== undefined) { |
test/capacity-crew.test.ts added +98
| @@ -0,0 +1,98 @@ | ||
| 1 | /** | |
| 2 | * Crew runs at once defaults to what the pod's memory holds (#2548): one run | |
| 3 | * per 3 GiB above the forge's own 2 GiB, at least 2 and at most 16. The | |
| 4 | * environment still wins, then a stored value; the sentinel's | |
| 5 | * away-from-default check compares with this default, so a big machine | |
| 6 | * running its share is not flagged. | |
| 7 | */ | |
| 8 | ||
| 9 | import { afterEach, beforeEach, describe, expect, test } from "bun:test"; | |
| 10 | import { mkdtemp, rm } from "node:fs/promises"; | |
| 11 | import { tmpdir } from "node:os"; | |
| 12 | import { join } from "node:path"; | |
| 13 | import { createApp, type App } from "../src/app.ts"; | |
| 14 | import { | |
| 15 | CAPACITY, | |
| 16 | capacityDefault, | |
| 17 | capacityValue, | |
| 18 | crewForMemory, | |
| 19 | memoryLimitBytes, | |
| 20 | setCapacity, | |
| 21 | } from "../src/core/capacity.ts"; | |
| 22 | import { settingsAway } from "../src/crew/sentinel.ts"; | |
| 23 | ||
| 24 | const GiB = 1024 ** 3; | |
| 25 | ||
| 26 | describe("crewForMemory", () => { | |
| 27 | test("one run per 3 GiB above 2 GiB, between 2 and 16", () => { | |
| 28 | expect(crewForMemory(4 * GiB)).toBe(2); // (4 − 2) / 3 = 0 → floor of 2 | |
| 29 | expect(crewForMemory(32 * GiB)).toBe(10); // (32 − 2) / 3 = 10 | |
| 30 | expect(crewForMemory(64 * GiB)).toBe(16); // (64 − 2) / 3 = 20 → ceiling of 16 | |
| 31 | expect(crewForMemory(11 * GiB)).toBe(3); // (11 − 2) / 3 = 3 | |
| 32 | expect(crewForMemory(GiB)).toBe(2); | |
| 33 | expect(crewForMemory(Number.NaN)).toBe(2); | |
| 34 | }); | |
| 35 | ||
| 36 | test("the capacity page says how the default is reached", () => { | |
| 37 | expect(CAPACITY["crew.concurrency"].help).toContain( | |
| 38 | "Empty: one per 3 GiB of memory above the forge's own 2 GiB", | |
| 39 | ); | |
| 40 | }); | |
| 41 | }); | |
| 42 | ||
| 43 | describe("the crew's default, and what overrides it", () => { | |
| 44 | let dataDir: string; | |
| 45 | let app: App; | |
| 46 | let saved: string | undefined; | |
| 47 | ||
| 48 | beforeEach(async () => { | |
| 49 | saved = process.env.FORGE_CREW_CONCURRENCY; | |
| 50 | delete process.env.FORGE_CREW_CONCURRENCY; | |
| 51 | dataDir = await mkdtemp(join(tmpdir(), "forge-capacity-crew-")); | |
| 52 | app = createApp({ dataDir, port: 0 }); | |
| 53 | }); | |
| 54 | afterEach(async () => { | |
| 55 | if (saved === undefined) delete process.env.FORGE_CREW_CONCURRENCY; | |
| 56 | else process.env.FORGE_CREW_CONCURRENCY = saved; | |
| 57 | app.close(); | |
| 58 | await rm(dataDir, { recursive: true, force: true }); | |
| 59 | }); | |
| 60 | ||
| 61 | test("unset, it is the pod's memory's share", () => { | |
| 62 | expect(capacityDefault("crew.concurrency")).toBe(String(crewForMemory(memoryLimitBytes()))); | |
| 63 | }); | |
| 64 | ||
| 65 | test("a set FORGE_CREW_CONCURRENCY wins over memory", () => { | |
| 66 | process.env.FORGE_CREW_CONCURRENCY = "7"; | |
| 67 | expect(capacityDefault("crew.concurrency")).toBe("7"); | |
| 68 | process.env.FORGE_CREW_CONCURRENCY = "99"; // out of range: ignored | |
| 69 | expect(capacityDefault("crew.concurrency")).toBe(String(crewForMemory(memoryLimitBytes()))); | |
| 70 | }); | |
| 71 | ||
| 72 | test("a stored value overrides the default", () => { | |
| 73 | const other = capacityDefault("crew.concurrency") === "5" ? "6" : "5"; | |
| 74 | setCapacity(app.db, "crew.concurrency", other, "system"); | |
| 75 | expect(capacityValue(app.db, "crew.concurrency")).toBe(other); | |
| 76 | setCapacity(app.db, "crew.concurrency", null, "system"); | |
| 77 | expect(capacityValue(app.db, "crew.concurrency")).toBe(capacityDefault("crew.concurrency")); | |
| 78 | }); | |
| 79 | ||
| 80 | test("the sentinel flags a value away from the memory default, not one at it", () => { | |
| 81 | const now = Date.now(); | |
| 82 | const age = () => | |
| 83 | app.db | |
| 84 | .query("UPDATE forge_setting SET updated_at = ?1 WHERE key = 'crew.concurrency'") | |
| 85 | .run(new Date(now - 3 * 86_400_000).toISOString()); | |
| 86 | const crew = () => settingsAway(app.db, now).filter((f) => f.key === "setting:crew.concurrency"); | |
| 87 | const fallback = capacityDefault("crew.concurrency"); | |
| 88 | setCapacity(app.db, "crew.concurrency", fallback, "system"); | |
| 89 | age(); | |
| 90 | expect(crew()).toEqual([]); | |
| 91 | const other = fallback === "5" ? "6" : "5"; | |
| 92 | setCapacity(app.db, "crew.concurrency", other, "system"); | |
| 93 | age(); | |
| 94 | const found = crew(); | |
| 95 | expect(found).toHaveLength(1); | |
| 96 | expect(found[0]!.evidence[0]).toBe(`crew.concurrency = "${other}", default "${fallback}"`); | |
| 97 | }); | |
| 98 | }); |
test/capacity.test.ts +2 −1
| @@ -22,6 +22,7 @@ import { AppsRuntime } from "../src/apps/runtime.ts"; | ||
| 22 | 22 | import { startChecks, type ChecksHandle } from "../src/checks/service.ts"; |
| 23 | 23 | import { |
| 24 | 24 | capacityChanges, |
| 25 | capacityDefault, | |
| 25 | 26 | capacityCount, |
| 26 | 27 | noteBooted, |
| 27 | 28 | capacityEnv, |
| @@ -297,7 +298,7 @@ describe("capacity: only a forge admin reads or changes it", () => { | ||
| 297 | 298 | expect(changes.map((c) => [c.key, c.from, c.to, c.actorName])).toEqual([ |
| 298 | 299 | ["checks.shards", "3", "", "alice"], |
| 299 | 300 | ["cells.egress", "enforce", "report", "alice"], |
| 300 | ["crew.concurrency", "2", "3", "alice"], | |
| 301 | ["crew.concurrency", capacityDefault("crew.concurrency"), "3", "alice"], | |
| 301 | 302 | ["checks.shards", "", "3", "alice"], |
| 302 | 303 | ]); |
| 303 | 304 | const after = await ( |
test/snapshot-restore-clones.test.ts +19
| @@ -11,6 +11,7 @@ import { rm } from "node:fs/promises"; | ||
| 11 | 11 | import { tmpdir } from "node:os"; |
| 12 | 12 | import { join } from "node:path"; |
| 13 | 13 | import { runGit } from "../src/git/env.ts"; |
| 14 | import { err } from "../src/kernel/result.ts"; | |
| 14 | 15 | import { CLONES, restoreSnapshot } from "../src/snapshot/restore.ts"; |
| 15 | 16 | import { dirStore, type Store } from "../src/snapshot/store.ts"; |
| 16 | 17 | |
| @@ -107,4 +108,22 @@ describe("a restore clones in parallel", () => { | ||
| 107 | 108 | "could not restore alice/r6", |
| 108 | 109 | ); |
| 109 | 110 | }); |
| 111 | ||
| 112 | test("of clones that fail together, the refusal names the earliest in the manifest", async () => { | |
| 113 | const store = dirStore(join(root, "store-4")); | |
| 114 | await seed(store, root); | |
| 115 | // r0 is refused last, r2 at once: the error still names r0. | |
| 116 | const git: typeof runGit = async (args, options) => { | |
| 117 | if (args[0] === "clone" && args[3]?.endsWith("alice--r0.bundle")) { | |
| 118 | await Bun.sleep(60); | |
| 119 | return err({ code: 128, stderr: "slow refusal", args }); | |
| 120 | } | |
| 121 | if (args[0] === "clone" && args[3]?.endsWith("alice--r2.bundle")) | |
| 122 | return err({ code: 128, stderr: "fast refusal", args }); | |
| 123 | return runGit(args, options); | |
| 124 | }; | |
| 125 | await expect(restoreSnapshot(store, NAME, join(root, "data-4"), { git })).rejects.toThrow( | |
| 126 | "could not restore alice/r0: slow refusal", | |
| 127 | ); | |
| 128 | }); | |
| 110 | 129 | }); |
new above
offline
Hosted children still boot at 1 here, which the README line does not mention. Also this line is long; consider a small local const.