946e2a9 · c375589 · 11cd37f · b52996f · 9ad7f4a
@@ -1,12 +1,16 @@
/**
 * The forge's own memory against its JS heap, hour by hour (#2151): every
 * minute `process.memoryUsage().rss` and `heapStats().heapSize` (bun:jsc)
 * are folded into the hour they fall in (`memory_hour` in `app.db`, under
 * the data directory), as a count, a sum and a max. RSS rising while the
 * minute `process.memoryUsage().rss` and `heapSize()` (bun:jsc) are folded
 * into the hour they fall in (`memory_hour` in `app.db`, under the data
 * directory), as a count, a sum and a max. RSS rising while the
 * heap stays flat is native memory growth. A week of hours is kept.
 *
 * `heapSize()`, never `heapStats().heapSize`: the same byte count, but
 * `heapStats()` first counts every object on the heap by type, which held the
 * loop ~280 ms a minute on a 2M-object heap and grows with it (#11118).
 */
import type { Database } from "bun:sqlite";
import { heapStats } from "bun:jsc";
import { heapSize } from "bun:jsc";
import type { App } from "../app.ts";
import { quietly } from "../core/write_clock.ts";
import { during } from "../perf/jobs.ts";
@@ -64,7 +68,7 @@
export type MemorySampler = { tick(now?: number): void; stop(): void };
export function startMemorySampler(app: App, everyMs = MEMORY_EVERY_MS): MemorySampler {
  const tick = (now = Date.now()) => {
    try {
      recordMemory(app.db, process.memoryUsage().rss, heapStats().heapSize, now);
      recordMemory(app.db, process.memoryUsage().rss, heapSize(), now);
    } catch (error) {
      app.log.warn("memory sample failed", { error: String(error) });
    }
@@ -0,0 +1,1 @@
a431515533fe72b93c083735ebe3fea9db717882d8ee7914618c48c9b84f972f
@@ -0,0 +1,8 @@
// A visitor's views in one day (`flushViews` in `perf/views_store.ts`, the
// VISITOR_DAY_MAX cap) are counted per new visitor on every flush. With only
// `traffic_view_scope_at` that count walks every row of the scope's day, once
// per visitor: a busy minute held the loop for over 100 ms (#11118). This
// index answers it from the visitor's own rows. Additive.
export default `
  CREATE INDEX traffic_view_scope_visitor_at ON traffic_view (scope, visitor, at);
`;
@@ -1371,7 +1371,7 @@
function forgetWorkdir(app: App, repo: Repo, pull: Pull): void {
 * the task again itself when it stops. Runs at boot and on every tick: a
 * forge that fell and was back within the lease sees its runs lapse later.
 */
function requeueInterrupted(app: App): void {
export function requeueInterrupted(app: App): void {
  for (const run of listActive(app.db, "agent")) {
    if (run.state === "running" && leaseLapsed(run, STALE_AFTER_MS))
      requeue(app, run, "interrupted by a restart; queued again");
@@ -1551,7 +1551,7 @@
export function startOrder(app: App): ReturnType<typeof gateQueue> {
}
/** Take queued runs, oldest first, up to the limit — one per issue at a time. */
function pump(
export function pump(
  app: App,
  options: CrewOptions,
  inFlight: Map<string, Promise<void>>,
@@ -91,3 +91,12 @@
test("an app's last pull and last good cursor walk the pull_run indexes", () =>
  expect(cursor.join("\n")).toContain("pull_run_app_ok");
  noScan(cursor);
});
test("a page-view flush counts a visitor's day from the visitor index, not the scope's whole day (#11118)", async () => {
  const { flushViews, noteView } = await import("../src/perf/views_store.ts");
  const origin = { refHost: "", source: "", medium: "", campaign: "", out: "" };
  noteView(app.db, { scope: "s", path: "/", origin, client: "c", agent: "Mozilla/5.0" });
  const lines = plans(() => void flushViews(app.db));
  expect(lines.join("\n")).toContain("traffic_view_scope_visitor_at");
  noScan(lines.filter((l) => l !== "SCAN CONSTANT ROW"));
});
@@ -0,0 +1,303 @@
/**
 * How long each background job holds the loop (#11118): one fixture with a busy forge's
 * volume (thousands of issues and pulls, tens of thousands of runs, comments and events)
 * is built once, and each job the forge runs under `during` ticks against it. A job's slice
 * is the longest stretch it kept the loop from turning: its synchronous call, or for an async
 * job the worst gap a `setImmediate` chain saw while it ran. Seams that leave the process
 * (git, the network, a child process, a cell) are fakes or fail fast, so what is timed is the
 * forge's own synchronous work. Every name passed to `during` in src/ is either timed here or
 * in UNTIMED with the reason, so a new job cannot go unmeasured.
 *
 * The table is printed worst first for the pull. No absolute number is asserted against a
 * raw clock at the 50 ms the issue names (`test/flake.test.ts`, #7824): each job is held only
 * under CEILING_MS, five times that, which a regression crosses and a slow runner does not.
 * The memory sampler's fix is asserted relative to one walk of the same heap in the same run.
 */
import { afterAll, beforeAll, expect, test } from "bun:test";
import { readdirSync, readFileSync, rmSync } from "node:fs";
import { join } from "node:path";
import { heapStats } from "bun:jsc";
import type { App } from "../src/app.ts";
import { bigForge } from "./helpers/big-forge.ts";
import { quietLogs } from "./helpers/globals.ts";
import { startSampler } from "../src/boards/sampler.ts";
import { startMemorySampler } from "../src/boards/memory.ts";
import { tickKeeper } from "../src/flow/keeper.ts";
import { tickClog } from "../src/flow/clog_tick.ts";
import { pump, recheckForDisk, requeueInterrupted, sweepAssigned } from "../src/crew/dispatch.ts";
import type { CrewOptions } from "../src/crew/runner.ts";
import { crewRoot } from "../src/crew/workdir.ts";
import { claimedClones, sweepOrphanClones } from "../src/crew/orphans.ts";
import { pruneRunFiles } from "../src/runs/prune.ts";
import { radarNight } from "../src/nightly/radar.ts";
import { runAheadReads } from "../src/origins/ahead.ts";
import { runGuardSweep } from "../src/origins/credential.ts";
import { sweepMirrorPulls } from "../src/origins/mirror-pulls.ts";
import { sweepReach } from "../src/spawn/reach.ts";
import { checkUpstream as checkParent } from "../src/spawn/upstream.ts";
import { census, censusWords, processRows } from "../src/kernel/census.ts";
import { runDelivery } from "../src/core/webhooks.ts";
import { pruneSessions } from "../src/core/sessions.ts";
import { pruneTokens } from "../src/core/tokens.ts";
import { runSync } from "../src/origins/sync.ts";
import { checkDue } from "../src/actions/upstream_follow.ts";
import { sweepStaleAsks } from "../src/crew/stale_asks.ts";
import { sweepStanding } from "../src/crew/standing.ts";
import { recordAskLedger } from "../src/web/studio/asks.ts";
import { sweepStarts } from "../src/actions/start_sweep.ts";
import { expireCrewHolds } from "../src/actions/holds.ts";
import { plannedRepoIds, tickWeeks } from "../src/flow/retro.ts";
import { getRepoById } from "../src/core/repos.ts";
import { failAbandonedRuns, heartbeatRuns } from "../src/core/runs.ts";
import { sleepBelowFloor, sleepIdleChildren } from "../src/spawn/host.ts";
import { createReadLedger, createReadWriter } from "../src/perf/reads.ts";
import { flushViews, noteView } from "../src/perf/views_store.ts";
import { requeueAbandonedChecks } from "../src/checks/service.ts";
import { releaseOrphans } from "../src/core/run_secret.ts";
import { STALE_AFTER_MS } from "../src/runs/lease.ts";
import { flushFlagUses } from "../src/core/flags.ts";
import { sendBeat } from "../src/spawn/heartbeat.ts";
import { runModelReview } from "../src/actions/model_review.ts";
import { readLandings } from "../src/release/landings.ts";
import { sweepStanding as sweepCallers } from "../src/http/caller.ts";
import { nightlyDue } from "../src/checks/scan_store.ts";
import { startNotifier } from "../src/crew/notify.ts";
import { fireDue } from "../src/apps/feature_schedule.ts";
import { pullDue } from "../src/apps/pull_schedule.ts";
/** Thousands of issues and pulls, as the issue asks; runs, comments and events thinned so the file runs in seconds (#1242). */
const BUSY = { issues: 2000, pulls: 2000, requested: 100, runs: 5000, comments: 5000, events: 10000, reviewComments: 500 };
// A job's warnings (a sandbox without git warns once per pull) are not its slice: the terminal's write is.
quietLogs();
let app: App;
let dir = "";
beforeAll(() => {
  ({ app, dir } = bigForge(BUSY));
  // A crew with work: a queue past the pump's 50-run window, and runs a dead forge left running.
  const at = new Date(Date.now() - 3_600_000).toISOString();
  const run = app.db.prepare(
    "INSERT INTO run (id, repo_id, issue_id, kind, state, reason, summary, created_at, started_at, actor_id) VALUES (?1, 'r', ?2, 'agent', ?3, 'assigned', '', ?4, ?5, 'builder')",
  );
  for (let n = 1; n <= 80; n++) run.run(`q${n}`, `i${n}`, "queued", at, null);
  for (let n = 81; n <= 90; n++) run.run(`s${n}`, `i${n}`, "running", at, at);
  reads = createReadWriter(app.db, readLedger, app.log, Date.now, 1e9);
}, 60_000);
afterAll(() => {
  reads.stop();
  app.db.close();
  rmSync(dir, { recursive: true, force: true });
});
/** The longest the loop stood still while `fn` ran, in ms: its synchronous call, or any continuation after. */
async function slice(fn: () => unknown): Promise<number> {
  let last = performance.now();
  let worst = 0;
  let turning = true;
  const turn = (): void => {
    const at = performance.now();
    worst = Math.max(worst, at - last);
    last = at;
    if (turning) setImmediate(turn);
  };
  setImmediate(turn);
  const start = performance.now();
  const result = fn();
  const sync = performance.now() - start;
  await Promise.resolve(result).catch(() => undefined);
  turning = false;
  await new Promise((resolve) => setImmediate(resolve));
  return Math.max(sync, worst);
}
/** The runs the fixture left running (s81–s90): the leases `lease-renew` renews. */
const RUNNING = Array.from({ length: 10 }, (_, i) => `s${81 + i}`);
const NO_ORIGIN = { refHost: "", source: "", medium: "", campaign: "", out: "" };
const readLedger = createReadLedger();
let reads: { flush(): void; stop(): void };
/** No free slot: the crew's pump gates every queued run and starts none. */
const NO_SLOT: CrewOptions = {
  forgeUrl: "https://forge.test/", localUrl: "http://127.0.0.1:0/", binary: "/nonexistent/claude",
  concurrency: 0, budgetUsd: 0, runTimeoutMs: 1,
};
/** Each job's tick, by the name it runs under in `during`, as the forge's timer calls it. */
function jobs(): Record<string, () => unknown> {
  const notifier = startNotifier(app, {
    forgeUrl: "https://forge.test/", send: async () => true, clockMs: 1e9,
    drainingSince: () => null, causes: () => [], alarms: () => [],
  });
  notifier.stop();
  const sampler = startSampler(app, { everyMs: 1e9 });
  const memory = startMemorySampler(app, 1e9);
  sampler.stop();
  memory.stop();
  return {
    sampler: () => sampler.tick(),
    "memory-sampler": () => memory.tick(),
    keeper: () => tickKeeper(app, new Date(), () => {}),
    clog: () => tickClog(app.db, app.bus),
    // The crew's boot tick and its timer's, together. With no free slot `pump` orders, gates
    // and holds every queued run but starts none: a start spawns Claude Code, not the loop's work.
    crew: () => {
      requeueInterrupted(app);
      recheckForDisk(app);
      sweepAssigned(app);
      pump(app, NO_SLOT, new Map());
    },
    // The orphan sweep reads the crew root (empty here); the claims it checks against are the fixture's.
    "clone-sweep": async () => {
      await sweepOrphanClones(crewRoot(app.config.dataDir), claimedClones(app.db));
      pruneRunFiles(app.config.dataDir);
    },
    // The radar's night: every repository's pins (git fails fast: no repositories on disk), npm faked.
    nightly: () => radarNight(app, async () => "not looked up"),
    // The census of /proc main.ts logs each minute (`processTick`), without the crew's reaping.
    processes: () => censusWords(census(processRows())),
    // The fixture has no origins, mirrors, children or parent link: each is its empty pass, one read.
    ahead: () => runAheadReads(app),
    "guard-sweep": () => runGuardSweep(app),
    "mirror-pulls": () => sweepMirrorPulls(app),
    reach: () => sweepReach(app),
    upstream: () => checkParent(app),
    // Its first pass on start tells and pages everyone (`tell(id, true)`, then `page`), a superset of the clock's tick.
    notify: async () => {
      const told = startNotifier(app, {
        forgeUrl: "https://forge.test/", send: async () => true, clockMs: 1e9,
        drainingSince: () => null, causes: () => [], alarms: () => [],
      });
      await told.settled();
      await told.close();
    },
    // The disk floor and the idle sweep over the hosted children; the fixture hosts none.
    "spawn-floor": () => sleepBelowFloor(app),
    "spawn-idle": () => sleepIdleChildren(app),
    // One renewal of every lease this process holds: the runs a dead forge left running.
    "lease-renew": () => heartbeatRuns(app.db, RUNNING),
    // A minute's reads and page views, buffered then written in one flush: the noting is in the slice too.
    reads: () => {
      for (let n = 0; n < 1000; n++) readLedger.note(Date.now(), { repo: "r", ref: "main", path: `f${n % 400}` }, n % 2 === 0);
      reads.flush();
    },
    views: () => {
      for (let n = 0; n < 1000; n++)
        noteView(app.db, { scope: "r", path: `/p${n % 200}`, origin: NO_ORIGIN, client: `c${n % 300}`, agent: "Mozilla/5.0" });
      flushViews(app.db);
    },
    delivery: () => {
      runDelivery(app.db, app.log, app.bus);
      pruneSessions(app.db);
      pruneTokens(app.db);
    },
    sync: () => runSync(app),
    follow: () => checkDue(app),
    asks: async () => {
      await sweepStaleAsks(app);
      await sweepStanding(app);
      recordAskLedger(app.db);
    },
    starts: () => sweepStarts(app),
    holds: () => expireCrewHolds(app),
    week: () => {
      const repos = plannedRepoIds(app.db).flatMap((id) => getRepoById(app.db, id) ?? []);
      tickWeeks(app.db, repos, new Date());
    },
    lease: () => {
      failAbandonedRuns(app.db, STALE_AFTER_MS, ["deploy"]);
      requeueAbandonedChecks(app, STALE_AFTER_MS);
      releaseOrphans(app.db);
    },
    "flag-uses": () => flushFlagUses(app.db),
    heartbeat: () => sendBeat(app, async () => new Response("")),
    "model-review": () => runModelReview(app),
    landings: () => readLandings(app),
    "notify-page": async () => {
      await notifier.settled(); // the notifier's first pass on start is its own, not the pager's
      await notifier.tick();
    },
    "standing-sweep": () => sweepCallers(app.db, () => {}),
    "nightly-scan": () => nightlyDue(app.db, new Date(Date.now() - 24 * 3_600_000).toISOString(), 25),
    "feature-schedule": () => fireDue(app),
    "pull-schedule": () => pullDue(app),
  };
}
/**
 * The jobs not timed here, each with why. The release keeper's are off unless a cluster turns them on.
 */
const UNTIMED: Record<string, string> = {
  lane: "a closure in startChecks; starting the checks service starts cell checks — follow-up",
  "lane-catch-up": "a closure in startChecks, once on boot; same as lane",
  "verify-sweep": "a closure in startChecks, off unless verifyEveryMs is set; same as lane",
  looks: "a closure in startLooks; needs the apps runtime and chromium, and the fixture has no apps",
  "perf-watcher": "a closure in startPerfWatcher over its own in-memory record; the fixture has no FORGE_REPO to file in",
  probes: "probes each app over HTTP after a deploy and nightly; the fixture has no apps",
  prompts: "a closure in startPromptRoutine; the fixture has no prompt routines",
  snapshot: "copies the database file to disk (VACUUM INTO): I/O off the loop, nightly",
  "snapshot-mirror": "uploads a snapshot to the mirror: network, nightly",
  drill: "restores a snapshot into a scratch forge: nightly, I/O",
  "release-behind": "reads the forge's own release tags with git; needs FORGE_REPO and its repository",
  "release-keeper": "cluster only (FORGE_RELEASE_KEEPER=on): talks to the cluster's API",
  "roll-probe": "cluster only (FORGE_RELEASE_KEEPER=on): one HTTP probe",
  "apps-sweep": "AppsRuntime's sweep over running app processes; the fixture has no apps",
  "cell-watch": "watches running app cells' cgroups; needs cells, skipped in this sandbox mode",
  "egress-idle": "a closure in the egress proxy over its live connections; needs the proxy's unix socket",
  "git-readers": "closes idle git cat-file readers: a map walk over child processes the fixture has none of",
  "github-traffic": "pulls GitHub's traffic API: network, and off unless a GitHub token is configured",
  presence: "writes this process's heartbeat file under the data directory: one small file write",
  "snapshot-journal": "inside a running snapshot upload: writes its journal; nightly, I/O",
  "snapshot-report": "inside a running snapshot upload: logs its progress; nightly",
  "system-live": "reads /proc and the cgroup roots for the system page while someone watches it",
  traffic: "a closure in createTrafficWriter over its in-memory minute ledger; follow-up",
};
/** A job held past this in one slice fails the test: five times the issue's 50 ms, so no runner flakes it. */
const CEILING_MS = 250;
/** Every name src/ passes to `during`: the list a new job joins, timed or not. */
function duringNames(dir = new URL("../src/", import.meta.url).pathname): Set<string> {
  const names = new Set<string>();
  for (const entry of readdirSync(dir, { withFileTypes: true, recursive: true })) {
    if (!entry.isFile() || !entry.name.endsWith(".ts")) continue;
    const text = readFileSync(join(entry.parentPath, entry.name), "utf8");
    for (const m of text.matchAll(/\bduring\(\s*"([^"]+)"/g)) names.add(m[1]!);
  }
  return names;
}
test("every job src/ runs under `during` is timed here or named in UNTIMED with a reason", () => {
  const timed = Object.keys(jobs());
  const named = duringNames();
  expect(named.size).toBeGreaterThan(40);
  const neither = [...named].filter((n) => !timed.includes(n) && !(n in UNTIMED)).sort();
  const both = timed.filter((n) => n in UNTIMED);
  const gone = [...timed, ...Object.keys(UNTIMED)].filter((n) => !named.has(n)).sort();
  expect({ neither, both, gone }).toEqual({ neither: [], both: [], gone: [] });
});
test("every timed job's tick holds the loop under the ceiling on a busy fixture, worst first", async () => {
  const measured: { name: string; ms: number }[] = [];
  for (const [name, tick] of Object.entries(jobs())) {
    await slice(tick); // the first tick prepares its statements; the timer's ticks find them warm
    Bun.gc(true); // the fixture's garbage is collected before the clock runs, not charged to whichever job comes first
    measured.push({ name, ms: await slice(tick) });
  }
  measured.sort((a, b) => b.ms - a.ms);
  console.log(`job slices (ms), worst first:\n${measured.map((j) => `  ${j.name}: ${j.ms.toFixed(1)}`).join("\n")}`);
  expect(measured.filter((j) => j.ms >= CEILING_MS)).toEqual([]);
}, 60_000);
test("the memory sampler reads the heap's size without walking the heap", async () => {
  // A heap with many objects on it, as a forge's is after a day of serving.
  const held = Array.from({ length: 300_000 }, (_, i) => ({ i }));
  const memory = startMemorySampler(app, 1e9);
  memory.stop();
  memory.tick();
  // The baseline, in this run on this heap: one walk of it, what the tick did before #11118.
  const walk = await slice(() => heapStats());
  const tick = await slice(() => memory.tick());
  console.log(`memory-sampler on ${held.length} objects: ${tick.toFixed(1)} ms; heapStats() alone: ${walk.toFixed(1)} ms`);
  expect(tick).toBeLessThan(walk / 4);
}, 60_000);
		@@ -1,12 +1,16 @@
1	1	/**
2	2	* The forge's own memory against its JS heap, hour by hour (#2151): every
3		* minute `process.memoryUsage().rss` and `heapStats().heapSize` (bun:jsc)
4		* are folded into the hour they fall in (`memory_hour` in `app.db`, under
5		* the data directory), as a count, a sum and a max. RSS rising while the
	3	* minute `process.memoryUsage().rss` and `heapSize()` (bun:jsc) are folded
	4	* into the hour they fall in (`memory_hour` in `app.db`, under the data
	5	* directory), as a count, a sum and a max. RSS rising while the
6	6	* heap stays flat is native memory growth. A week of hours is kept.
	7	*
	8	* `heapSize()`, never `heapStats().heapSize`: the same byte count, but
	9	* `heapStats()` first counts every object on the heap by type, which held the
	10	* loop ~280 ms a minute on a 2M-object heap and grows with it (#11118).
7	11	*/
8	12	import type { Database } from "bun:sqlite";
9		import { heapStats } from "bun:jsc";
	13	import { heapSize } from "bun:jsc";
10	14	import type { App } from "../app.ts";
11	15	import { quietly } from "../core/write_clock.ts";
12	16	import { during } from "../perf/jobs.ts";
		@@ -64,7 +68,7 @@ export type MemorySampler = { tick(now?: number): void; stop(): void };
64	68	export function startMemorySampler(app: App, everyMs = MEMORY_EVERY_MS): MemorySampler {
65	69	const tick = (now = Date.now()) => {
66	70	try {
67		recordMemory(app.db, process.memoryUsage().rss, heapStats().heapSize, now);
	71	recordMemory(app.db, process.memoryUsage().rss, heapSize(), now);
68	72	} catch (error) {
69	73	app.log.warn("memory sample failed", { error: String(error) });
70	74	}
		@@ -0,0 +1,1 @@
	1	a431515533fe72b93c083735ebe3fea9db717882d8ee7914618c48c9b84f972f
		@@ -0,0 +1,8 @@
	1	// A visitor's views in one day (`flushViews` in `perf/views_store.ts`, the
	2	// VISITOR_DAY_MAX cap) are counted per new visitor on every flush. With only
	3	// `traffic_view_scope_at` that count walks every row of the scope's day, once
	4	// per visitor: a busy minute held the loop for over 100 ms (#11118). This
	5	// index answers it from the visitor's own rows. Additive.
	6	export default `
	7	CREATE INDEX traffic_view_scope_visitor_at ON traffic_view (scope, visitor, at);
	8	`;
		@@ -1371,7 +1371,7 @@ function forgetWorkdir(app: App, repo: Repo, pull: Pull): void {
1371	1371	* the task again itself when it stops. Runs at boot and on every tick: a
1372	1372	* forge that fell and was back within the lease sees its runs lapse later.
1373	1373	*/
1374		function requeueInterrupted(app: App): void {
	1374	export function requeueInterrupted(app: App): void {
1375	1375	for (const run of listActive(app.db, "agent")) {
1376	1376	if (run.state === "running" && leaseLapsed(run, STALE_AFTER_MS))
1377	1377	requeue(app, run, "interrupted by a restart; queued again");
		@@ -1551,7 +1551,7 @@ export function startOrder(app: App): ReturnType<typeof gateQueue> {
1551	1551	}
1552	1552
1553	1553	/** Take queued runs, oldest first, up to the limit — one per issue at a time. */
1554		function pump(
	1554	export function pump(
1555	1555	app: App,
1556	1556	options: CrewOptions,
1557	1557	inFlight: Map<string, Promise<void>>,
		@@ -91,3 +91,12 @@ test("an app's last pull and last good cursor walk the pull_run indexes", () =>
91	91	expect(cursor.join("\n")).toContain("pull_run_app_ok");
92	92	noScan(cursor);
93	93	});
	94
	95	test("a page-view flush counts a visitor's day from the visitor index, not the scope's whole day (#11118)", async () => {
	96	const { flushViews, noteView } = await import("../src/perf/views_store.ts");
	97	const origin = { refHost: "", source: "", medium: "", campaign: "", out: "" };
	98	noteView(app.db, { scope: "s", path: "/", origin, client: "c", agent: "Mozilla/5.0" });
	99	const lines = plans(() => void flushViews(app.db));
	100	expect(lines.join("\n")).toContain("traffic_view_scope_visitor_at");
	101	noScan(lines.filter((l) => l !== "SCAN CONSTANT ROW"));
	102	});
		@@ -0,0 +1,303 @@
	1	/**
	2	* How long each background job holds the loop (#11118): one fixture with a busy forge's
	3	* volume (thousands of issues and pulls, tens of thousands of runs, comments and events)
	4	* is built once, and each job the forge runs under `during` ticks against it. A job's slice
	5	* is the longest stretch it kept the loop from turning: its synchronous call, or for an async
	6	* job the worst gap a `setImmediate` chain saw while it ran. Seams that leave the process
	7	* (git, the network, a child process, a cell) are fakes or fail fast, so what is timed is the
	8	* forge's own synchronous work. Every name passed to `during` in src/ is either timed here or
	9	* in UNTIMED with the reason, so a new job cannot go unmeasured.
	10	*
	11	* The table is printed worst first for the pull. No absolute number is asserted against a
	12	* raw clock at the 50 ms the issue names (`test/flake.test.ts`, #7824): each job is held only
	13	* under CEILING_MS, five times that, which a regression crosses and a slow runner does not.
	14	* The memory sampler's fix is asserted relative to one walk of the same heap in the same run.
	15	*/
	16	import { afterAll, beforeAll, expect, test } from "bun:test";
	17	import { readdirSync, readFileSync, rmSync } from "node:fs";
	18	import { join } from "node:path";
	19	import { heapStats } from "bun:jsc";
	20	import type { App } from "../src/app.ts";
	21	import { bigForge } from "./helpers/big-forge.ts";
	22	import { quietLogs } from "./helpers/globals.ts";
	23	import { startSampler } from "../src/boards/sampler.ts";
	24	import { startMemorySampler } from "../src/boards/memory.ts";
	25	import { tickKeeper } from "../src/flow/keeper.ts";
	26	import { tickClog } from "../src/flow/clog_tick.ts";
	27	import { pump, recheckForDisk, requeueInterrupted, sweepAssigned } from "../src/crew/dispatch.ts";
	28	import type { CrewOptions } from "../src/crew/runner.ts";
	29	import { crewRoot } from "../src/crew/workdir.ts";
	30	import { claimedClones, sweepOrphanClones } from "../src/crew/orphans.ts";
	31	import { pruneRunFiles } from "../src/runs/prune.ts";
	32	import { radarNight } from "../src/nightly/radar.ts";
	33	import { runAheadReads } from "../src/origins/ahead.ts";
	34	import { runGuardSweep } from "../src/origins/credential.ts";
	35	import { sweepMirrorPulls } from "../src/origins/mirror-pulls.ts";
	36	import { sweepReach } from "../src/spawn/reach.ts";
	37	import { checkUpstream as checkParent } from "../src/spawn/upstream.ts";
	38	import { census, censusWords, processRows } from "../src/kernel/census.ts";
	39	import { runDelivery } from "../src/core/webhooks.ts";
	40	import { pruneSessions } from "../src/core/sessions.ts";
	41	import { pruneTokens } from "../src/core/tokens.ts";
	42	import { runSync } from "../src/origins/sync.ts";
	43	import { checkDue } from "../src/actions/upstream_follow.ts";
	44	import { sweepStaleAsks } from "../src/crew/stale_asks.ts";
	45	import { sweepStanding } from "../src/crew/standing.ts";
	46	import { recordAskLedger } from "../src/web/studio/asks.ts";
	47	import { sweepStarts } from "../src/actions/start_sweep.ts";
	48	import { expireCrewHolds } from "../src/actions/holds.ts";
	49	import { plannedRepoIds, tickWeeks } from "../src/flow/retro.ts";
	50	import { getRepoById } from "../src/core/repos.ts";
	51	import { failAbandonedRuns, heartbeatRuns } from "../src/core/runs.ts";
	52	import { sleepBelowFloor, sleepIdleChildren } from "../src/spawn/host.ts";
	53	import { createReadLedger, createReadWriter } from "../src/perf/reads.ts";
	54	import { flushViews, noteView } from "../src/perf/views_store.ts";
	55	import { requeueAbandonedChecks } from "../src/checks/service.ts";
	56	import { releaseOrphans } from "../src/core/run_secret.ts";
	57	import { STALE_AFTER_MS } from "../src/runs/lease.ts";
	58	import { flushFlagUses } from "../src/core/flags.ts";
	59	import { sendBeat } from "../src/spawn/heartbeat.ts";
	60	import { runModelReview } from "../src/actions/model_review.ts";
	61	import { readLandings } from "../src/release/landings.ts";
	62	import { sweepStanding as sweepCallers } from "../src/http/caller.ts";
	63	import { nightlyDue } from "../src/checks/scan_store.ts";
	64	import { startNotifier } from "../src/crew/notify.ts";
	65	import { fireDue } from "../src/apps/feature_schedule.ts";
	66	import { pullDue } from "../src/apps/pull_schedule.ts";
	67
	68	/** Thousands of issues and pulls, as the issue asks; runs, comments and events thinned so the file runs in seconds (#1242). */
	69	const BUSY = { issues: 2000, pulls: 2000, requested: 100, runs: 5000, comments: 5000, events: 10000, reviewComments: 500 };
	70
	71	// A job's warnings (a sandbox without git warns once per pull) are not its slice: the terminal's write is.
	72	quietLogs();
	73
	74	let app: App;
	75	let dir = "";
	76	beforeAll(() => {
	77	({ app, dir } = bigForge(BUSY));
	78	// A crew with work: a queue past the pump's 50-run window, and runs a dead forge left running.
	79	const at = new Date(Date.now() - 3_600_000).toISOString();
	80	const run = app.db.prepare(
	81	"INSERT INTO run (id, repo_id, issue_id, kind, state, reason, summary, created_at, started_at, actor_id) VALUES (?1, 'r', ?2, 'agent', ?3, 'assigned', '', ?4, ?5, 'builder')",
	82	);
	83	for (let n = 1; n <= 80; n++) run.run(`q${n}`, `i${n}`, "queued", at, null);
	84	for (let n = 81; n <= 90; n++) run.run(`s${n}`, `i${n}`, "running", at, at);
	85	reads = createReadWriter(app.db, readLedger, app.log, Date.now, 1e9);
	86	}, 60_000);
	87	afterAll(() => {
	88	reads.stop();
	89	app.db.close();
	90	rmSync(dir, { recursive: true, force: true });
	91	});
	92
	93	/** The longest the loop stood still while `fn` ran, in ms: its synchronous call, or any continuation after. */
	94	async function slice(fn: () => unknown): Promise<number> {
	95	let last = performance.now();
	96	let worst = 0;
	97	let turning = true;
	98	const turn = (): void => {
	99	const at = performance.now();
	100	worst = Math.max(worst, at - last);
	101	last = at;
	102	if (turning) setImmediate(turn);
	103	};
	104	setImmediate(turn);
	105	const start = performance.now();
	106	const result = fn();
	107	const sync = performance.now() - start;
	108	await Promise.resolve(result).catch(() => undefined);
	109	turning = false;
	110	await new Promise((resolve) => setImmediate(resolve));
	111	return Math.max(sync, worst);
	112	}
	113
	114	/** The runs the fixture left running (s81–s90): the leases `lease-renew` renews. */
	115	const RUNNING = Array.from({ length: 10 }, (_, i) => `s${81 + i}`);
	116	const NO_ORIGIN = { refHost: "", source: "", medium: "", campaign: "", out: "" };
	117	const readLedger = createReadLedger();
	118	let reads: { flush(): void; stop(): void };
	119
	120	/** No free slot: the crew's pump gates every queued run and starts none. */
	121	const NO_SLOT: CrewOptions = {
	122	forgeUrl: "https://forge.test/", localUrl: "http://127.0.0.1:0/", binary: "/nonexistent/claude",
	123	concurrency: 0, budgetUsd: 0, runTimeoutMs: 1,
	124	};
	125
	126	/** Each job's tick, by the name it runs under in `during`, as the forge's timer calls it. */
	127	function jobs(): Record<string, () => unknown> {
	128	const notifier = startNotifier(app, {
	129	forgeUrl: "https://forge.test/", send: async () => true, clockMs: 1e9,
	130	drainingSince: () => null, causes: () => [], alarms: () => [],
	131	});
	132	notifier.stop();
	133	const sampler = startSampler(app, { everyMs: 1e9 });
	134	const memory = startMemorySampler(app, 1e9);
	135	sampler.stop();
	136	memory.stop();
	137	return {
	138	sampler: () => sampler.tick(),
	139	"memory-sampler": () => memory.tick(),
	140	keeper: () => tickKeeper(app, new Date(), () => {}),
	141	clog: () => tickClog(app.db, app.bus),
	142	// The crew's boot tick and its timer's, together. With no free slot `pump` orders, gates
	143	// and holds every queued run but starts none: a start spawns Claude Code, not the loop's work.
	144	crew: () => {
	145	requeueInterrupted(app);
	146	recheckForDisk(app);
	147	sweepAssigned(app);
	148	pump(app, NO_SLOT, new Map());
	149	},
	150	// The orphan sweep reads the crew root (empty here); the claims it checks against are the fixture's.
	151	"clone-sweep": async () => {
	152	await sweepOrphanClones(crewRoot(app.config.dataDir), claimedClones(app.db));
	153	pruneRunFiles(app.config.dataDir);
	154	},
	155	// The radar's night: every repository's pins (git fails fast: no repositories on disk), npm faked.
	156	nightly: () => radarNight(app, async () => "not looked up"),
	157	// The census of /proc main.ts logs each minute (`processTick`), without the crew's reaping.
	158	processes: () => censusWords(census(processRows())),
	159	// The fixture has no origins, mirrors, children or parent link: each is its empty pass, one read.
	160	ahead: () => runAheadReads(app),
	161	"guard-sweep": () => runGuardSweep(app),
	162	"mirror-pulls": () => sweepMirrorPulls(app),
	163	reach: () => sweepReach(app),
	164	upstream: () => checkParent(app),
	165	// Its first pass on start tells and pages everyone (`tell(id, true)`, then `page`), a superset of the clock's tick.
	166	notify: async () => {
	167	const told = startNotifier(app, {
	168	forgeUrl: "https://forge.test/", send: async () => true, clockMs: 1e9,
	169	drainingSince: () => null, causes: () => [], alarms: () => [],
	170	});
	171	await told.settled();
	172	await told.close();
	173	},
	174	// The disk floor and the idle sweep over the hosted children; the fixture hosts none.
	175	"spawn-floor": () => sleepBelowFloor(app),
	176	"spawn-idle": () => sleepIdleChildren(app),
	177	// One renewal of every lease this process holds: the runs a dead forge left running.
	178	"lease-renew": () => heartbeatRuns(app.db, RUNNING),
	179	// A minute's reads and page views, buffered then written in one flush: the noting is in the slice too.
	180	reads: () => {
	181	for (let n = 0; n < 1000; n++) readLedger.note(Date.now(), { repo: "r", ref: "main", path: `f${n % 400}` }, n % 2 === 0);
	182	reads.flush();
	183	},
	184	views: () => {
	185	for (let n = 0; n < 1000; n++)
	186	noteView(app.db, { scope: "r", path: `/p${n % 200}`, origin: NO_ORIGIN, client: `c${n % 300}`, agent: "Mozilla/5.0" });
	187	flushViews(app.db);
	188	},
	189	delivery: () => {
	190	runDelivery(app.db, app.log, app.bus);
	191	pruneSessions(app.db);
	192	pruneTokens(app.db);
	193	},
	194	sync: () => runSync(app),
	195	follow: () => checkDue(app),
	196	asks: async () => {
	197	await sweepStaleAsks(app);
	198	await sweepStanding(app);
	199	recordAskLedger(app.db);
	200	},
	201	starts: () => sweepStarts(app),
	202	holds: () => expireCrewHolds(app),
	203	week: () => {
	204	const repos = plannedRepoIds(app.db).flatMap((id) => getRepoById(app.db, id) ?? []);
	205	tickWeeks(app.db, repos, new Date());
	206	},
	207	lease: () => {
	208	failAbandonedRuns(app.db, STALE_AFTER_MS, ["deploy"]);
	209	requeueAbandonedChecks(app, STALE_AFTER_MS);
	210	releaseOrphans(app.db);
	211	},
	212	"flag-uses": () => flushFlagUses(app.db),
	213	heartbeat: () => sendBeat(app, async () => new Response("")),
	214	"model-review": () => runModelReview(app),
	215	landings: () => readLandings(app),
	216	"notify-page": async () => {
	217	await notifier.settled(); // the notifier's first pass on start is its own, not the pager's
	218	await notifier.tick();
	219	},
	220	"standing-sweep": () => sweepCallers(app.db, () => {}),
	221	"nightly-scan": () => nightlyDue(app.db, new Date(Date.now() - 24 * 3_600_000).toISOString(), 25),
	222	"feature-schedule": () => fireDue(app),
	223	"pull-schedule": () => pullDue(app),
	224	};
	225	}
	226
	227	/**
	228	* The jobs not timed here, each with why. The release keeper's are off unless a cluster turns them on.
	229	*/
	230	const UNTIMED: Record<string, string> = {
	231	lane: "a closure in startChecks; starting the checks service starts cell checks — follow-up",
	232	"lane-catch-up": "a closure in startChecks, once on boot; same as lane",
	233	"verify-sweep": "a closure in startChecks, off unless verifyEveryMs is set; same as lane",
	234	looks: "a closure in startLooks; needs the apps runtime and chromium, and the fixture has no apps",
	235	"perf-watcher": "a closure in startPerfWatcher over its own in-memory record; the fixture has no FORGE_REPO to file in",
	236	probes: "probes each app over HTTP after a deploy and nightly; the fixture has no apps",
	237	prompts: "a closure in startPromptRoutine; the fixture has no prompt routines",
	238	snapshot: "copies the database file to disk (VACUUM INTO): I/O off the loop, nightly",
	239	"snapshot-mirror": "uploads a snapshot to the mirror: network, nightly",
	240	drill: "restores a snapshot into a scratch forge: nightly, I/O",
	241	"release-behind": "reads the forge's own release tags with git; needs FORGE_REPO and its repository",
	242	"release-keeper": "cluster only (FORGE_RELEASE_KEEPER=on): talks to the cluster's API",
	243	"roll-probe": "cluster only (FORGE_RELEASE_KEEPER=on): one HTTP probe",
	244	"apps-sweep": "AppsRuntime's sweep over running app processes; the fixture has no apps",
	245	"cell-watch": "watches running app cells' cgroups; needs cells, skipped in this sandbox mode",
	246	"egress-idle": "a closure in the egress proxy over its live connections; needs the proxy's unix socket",
	247	"git-readers": "closes idle git cat-file readers: a map walk over child processes the fixture has none of",
	248	"github-traffic": "pulls GitHub's traffic API: network, and off unless a GitHub token is configured",
	249	presence: "writes this process's heartbeat file under the data directory: one small file write",
	250	"snapshot-journal": "inside a running snapshot upload: writes its journal; nightly, I/O",
	251	"snapshot-report": "inside a running snapshot upload: logs its progress; nightly",
	252	"system-live": "reads /proc and the cgroup roots for the system page while someone watches it",
	253	traffic: "a closure in createTrafficWriter over its in-memory minute ledger; follow-up",
	254	};
	255
	256	/** A job held past this in one slice fails the test: five times the issue's 50 ms, so no runner flakes it. */
	257	const CEILING_MS = 250;
	258
	259	/** Every name src/ passes to `during`: the list a new job joins, timed or not. */
	260	function duringNames(dir = new URL("../src/", import.meta.url).pathname): Set<string> {
	261	const names = new Set<string>();
	262	for (const entry of readdirSync(dir, { withFileTypes: true, recursive: true })) {
	263	if (!entry.isFile() \|\| !entry.name.endsWith(".ts")) continue;
	264	const text = readFileSync(join(entry.parentPath, entry.name), "utf8");
	265	for (const m of text.matchAll(/\bduring\(\s*"([^"]+)"/g)) names.add(m[1]!);
	266	}
	267	return names;
	268	}
	269
	270	test("every job src/ runs under `during` is timed here or named in UNTIMED with a reason", () => {
	271	const timed = Object.keys(jobs());
	272	const named = duringNames();
	273	expect(named.size).toBeGreaterThan(40);
	274	const neither = [...named].filter((n) => !timed.includes(n) && !(n in UNTIMED)).sort();
	275	const both = timed.filter((n) => n in UNTIMED);
	276	const gone = [...timed, ...Object.keys(UNTIMED)].filter((n) => !named.has(n)).sort();
	277	expect({ neither, both, gone }).toEqual({ neither: [], both: [], gone: [] });
	278	});
	279
	280	test("every timed job's tick holds the loop under the ceiling on a busy fixture, worst first", async () => {
	281	const measured: { name: string; ms: number }[] = [];
	282	for (const [name, tick] of Object.entries(jobs())) {
	283	await slice(tick); // the first tick prepares its statements; the timer's ticks find them warm
	284	Bun.gc(true); // the fixture's garbage is collected before the clock runs, not charged to whichever job comes first
	285	measured.push({ name, ms: await slice(tick) });
	286	}
	287	measured.sort((a, b) => b.ms - a.ms);
	288	console.log(`job slices (ms), worst first:\n${measured.map((j) => ` ${j.name}: ${j.ms.toFixed(1)}`).join("\n")}`);
	289	expect(measured.filter((j) => j.ms >= CEILING_MS)).toEqual([]);
	290	}, 60_000);
	291
	292	test("the memory sampler reads the heap's size without walking the heap", async () => {
	293	// A heap with many objects on it, as a forge's is after a day of serving.
	294	const held = Array.from({ length: 300_000 }, (_, i) => ({ i }));
	295	const memory = startMemorySampler(app, 1e9);
	296	memory.stop();
	297	memory.tick();
	298	// The baseline, in this run on this heap: one walk of it, what the tick did before #11118.
	299	const walk = await slice(() => heapStats());
	300	const tick = await slice(() => memory.tick());
	301	console.log(`memory-sampler on ${held.length} objects: ${tick.toFixed(1)} ms; heapStats() alone: ${walk.toFixed(1)} ms`);
	302	expect(tick).toBeLessThan(walk / 4);
	303	}, 60_000);