trevato/forge

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

More
Forks1g fMapg aRuns15g rEventsFlows

A due snapshot waits while a crew run holds the only core, and is taken on the next tick after #11792

Merged torquecrew opened this

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

ConversationFiles
Merged

torque/forge:torque/11776-a-due-snapshot-waits-while-a-crew-run-ho → main 3 commits fork

  • Merged. torque/11776-a-due-snapshot-waits-while-a-crew-run-ho was merged into main as 86e4867.

Review

Sign in to review this change.
2 files changed +177 −2
src/snapshot/service.ts +44 −2
@@ -36,6 +36,8 @@ import { isOffBox } from "./off_box.ts";
3636 import { noteOffBox, recordSnapshotOutcome } from "./outcome.ts";
3737 import { s3FromEnv, type Store } from "./store.ts";
3838 import { during } from "../perf/jobs.ts";
39 import { crewHoldsCore, forgeCpus } from "../core/one_core.ts";
40 import { listActive } from "../core/runs.ts";
3941
4042 const log = createLog("snapshots");
4143
@@ -43,6 +45,8 @@ const log = createLog("snapshots");
4345 const NIGHTLY_HOUR_UTC = 8;
4446 /** After a failed snapshot, the next try is this soon (or the nightly, if sooner). */
4547 export const RETRY_MS = 3_600_000;
48 /** A due snapshot waits for a crew run on a 1-CPU forge at most this long, then is taken anyway. */
49 export const MAX_DEFER_MS = 3_600_000;
4650 /** A moving snapshot tells the page how far it is no oftener than this. */
4751 const PROGRESS_EVENT_MS = 2_000;
4852
@@ -149,6 +153,11 @@ export function startSnapshots(
149153 readonly bucketFirstMs?: number | undefined;
150154 /** Test seam: the clock `takeIfDue` judges by. */
151155 readonly now?: (() => number) | undefined;
156 /** Test seams: the forge's CPU count and how many crew runs are active (1 CPU: a run holds the core). */
157 readonly cpus?: (() => number) | undefined;
158 readonly activeCrewRuns?: (() => number) | undefined;
159 /** Test seam: how soon the first nightly timer fires. */
160 readonly nightlyInMs?: number | undefined;
152161 } = {
153162 version: "0",
154163 },
@@ -262,6 +271,26 @@ export function startSnapshots(
262271 }
263272 };
264273
274 // One core: a crew run has it, so a due snapshot waits and is taken on the
275 // next tick after the run ends (src/core/one_core.ts). Never forever: after
276 // MAX_DEFER_MS of waiting it is taken anyway. A snapshot running is untouched.
277 let deferredSince: number | undefined;
278 const deferred = (): boolean => {
279 const at = (options.now ?? Date.now)();
280 const held = crewHoldsCore(
281 (options.cpus ?? forgeCpus)(),
282 (options.activeCrewRuns ?? (() => listActive(app.db, "agent").length))(),
283 );
284 if (!held) {
285 deferredSince = undefined;
286 return false;
287 }
288 deferredSince ??= at;
289 if (at - deferredSince >= MAX_DEFER_MS) return false;
290 log.info("snapshot due; a crew run holds the one core, waiting");
291 return true;
292 };
293
265294 // A restart takes the night's snapshot if the store does not hold it: a
266295 // release at 08:21 killed the one in flight and nothing took it again (#8742).
267296 // The manifest is written last, so a snapshot cut short is not one.
@@ -277,6 +306,8 @@ export function startSnapshots(
277306 return false;
278307 }
279308 if (stopped || active > 0 || !snapshotDue(newest?.finishedAt, (options.now ?? Date.now)())) return false;
309 if (deferred()) return false;
310 deferredSince = undefined;
280311 const cut = await cutShort(store, newest?.name);
281312 log.info("snapshot due or lost; taking one", { newest: newest?.name ?? "none", resuming: cut?.toISOString() ?? "no" });
282313 await take(cut).catch(() => {});
@@ -380,9 +411,20 @@ export function startSnapshots(
380411 if (timer !== undefined) clearTimeout(timer);
381412 timer = undefined;
382413 if (options.nightly === false || stopped) return;
383 timer = setTimeout(() => void during("snapshot", take).catch(() => {}), ms);
414 timer = setTimeout(
415 () =>
416 void during("snapshot", () => {
417 if (active === 0 && deferred()) {
418 arm(options.snapshotCheckMs ?? SNAPSHOT_CHECK_MS);
419 return Promise.resolve();
420 }
421 deferredSince = undefined;
422 return take().then(() => {});
423 }).catch(() => {}),
424 ms,
425 );
384426 }
385 arm(untilNightly());
427 arm(options.nightlyInMs ?? untilNightly());
386428
387429 // The board only reads; the time the forge first owed a drill is kept here.
388430 try {
test/snapshot-one-core.test.ts added +133
@@ -0,0 +1,133 @@
1 /**
2 * The 1-CPU rule (#11561): a due snapshot waits while a crew run holds the
3 * only core and is taken on the next tick after the run ends. The cap on
4 * deferral is MAX_DEFER_MS, one hour: a run that is always active cannot
5 * keep a snapshot from being taken.
6 */
7 import { afterAll, beforeAll, expect, test } from "bun:test";
8 import { mkdtempSync } from "node:fs";
9 import { rm } from "node:fs/promises";
10 import { tmpdir } from "node:os";
11 import { join } from "node:path";
12 import { createApp, type App } from "../src/app.ts";
13 import { MAX_DEFER_MS, startSnapshots } from "../src/snapshot/service.ts";
14 import { listSnapshots } from "../src/snapshot/snapshot.ts";
15 import { dirStore } from "../src/snapshot/store.ts";
16
17 let root: string;
18 let app: App;
19 let n = 0;
20
21 beforeAll(() => {
22 root = mkdtempSync(join(tmpdir(), "forge-onecore-"));
23 app = createApp({ dataDir: join(root, "live"), port: 0 });
24 });
25 afterAll(async () => {
26 app.close();
27 await rm(root, { recursive: true, force: true });
28 });
29
30 async function setup(cpus: number, runs: { n: number }, clock: { t: number }) {
31 const store = dirStore(join(root, `store-${n++}`));
32 const snaps = startSnapshots(app, store, {
33 version: "t",
34 nightly: false,
35 now: () => clock.t,
36 cpus: () => cpus,
37 activeCrewRuns: () => runs.n,
38 });
39 return { store, snaps };
40 }
41
42 const clock = () => ({ t: Date.parse("2026-11-12T09:00:00Z") });
43
44 test("1 CPU with a crew run active defers a due snapshot, then takes it once the run is gone", async () => {
45 const runs = { n: 1 };
46 const c = clock();
47 const { store, snaps } = await setup(1, runs, c);
48 try {
49 expect(await snaps.takeIfDue()).toBe(false);
50 expect(await listSnapshots(store)).toHaveLength(0);
51 c.t += 5 * 60_000;
52 expect(await snaps.takeIfDue()).toBe(false);
53 runs.n = 0;
54 expect(await snaps.takeIfDue()).toBe(true);
55 expect(await listSnapshots(store)).toHaveLength(1);
56 } finally {
57 await snaps.stop();
58 }
59 });
60
61 test("2 CPUs take a due snapshot at once, run or not", async () => {
62 const { store, snaps } = await setup(2, { n: 1 }, clock());
63 try {
64 expect(await snaps.takeIfDue()).toBe(true);
65 expect(await listSnapshots(store)).toHaveLength(1);
66 } finally {
67 await snaps.stop();
68 }
69 });
70
71 test("the cap: a run that never ends defers a snapshot at most one hour, then it is taken", async () => {
72 expect(MAX_DEFER_MS).toBe(3_600_000);
73 const c = clock();
74 const { store, snaps } = await setup(1, { n: 1 }, c);
75 try {
76 expect(await snaps.takeIfDue()).toBe(false);
77 c.t += MAX_DEFER_MS - 1;
78 expect(await snaps.takeIfDue()).toBe(false);
79 c.t += 1;
80 expect(await snaps.takeIfDue()).toBe(true);
81 expect(await listSnapshots(store)).toHaveLength(1);
82 } finally {
83 await snaps.stop();
84 }
85 });
86
87 test("the nightly timer: a held core re-arms it without taking; the next tick after the run takes", async () => {
88 const runs = { n: 1 };
89 const store = dirStore(join(root, `store-${n++}`));
90 const snaps = startSnapshots(app, store, {
91 version: "t",
92 nightlyInMs: 20,
93 snapshotCheckMs: 20,
94 snapshotCatchUpMs: 3_600_000,
95 catchUpMs: 3_600_000,
96 cpus: () => 1,
97 activeCrewRuns: () => runs.n,
98 });
99 try {
100 await Bun.sleep(150);
101 expect(await listSnapshots(store)).toHaveLength(0);
102 runs.n = 0;
103 for (let i = 0; i < 100 && (await listSnapshots(store)).length === 0; i++) await Bun.sleep(50);
104 expect(await listSnapshots(store)).toHaveLength(1);
105 } finally {
106 await snaps.stop();
107 }
108 });
109
110 test("a snapshot already running is left alone: a due check neither waits on it nor starts a second", async () => {
111 const inner = dirStore(join(root, `store-${n++}`));
112 let release!: () => void;
113 const gate = new Promise<void>((r) => (release = r));
114 const store = { ...inner, put: async (k: string, b: Uint8Array) => (await gate, inner.put(k, b)) };
115 const snaps = startSnapshots(app, store, {
116 version: "t",
117 nightly: false,
118 cpus: () => 1,
119 activeCrewRuns: () => 0,
120 });
121 try {
122 const running = snaps.now();
123 await Bun.sleep(50);
124 expect(snaps.progress()).toBeDefined();
125 expect(await snaps.takeIfDue()).toBe(false);
126 release();
127 await running;
128 expect(await listSnapshots(inner)).toHaveLength(1);
129 } finally {
130 release();
131 await snaps.stop();
132 }
133 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs