615ed97 · 5ff78d3 · 12d00e2 · ff3754f
@@ -0,0 +1,10 @@
# Memory held, not hit (#11980)
A child forge in a cell has one memory limit for itself, its crew run, its checks and its apps. At or over 75% of that limit (anonymous memory, the page cache not counted) and with one of its own runs already going, the child starts nothing new:
- the dispatcher keeps the queued agent run queued, with one log line, and vitals counts it as "held on purpose: … by the forge's memory near its limit", not as waiting;
- the check queue keeps checks queued, one log line each, so checks run one at a time.
With nothing running a start is always allowed, so the hold never lasts. A forge that is not a cell child reads no pressure and is never held. The rule is `src/runs/memory_hold.ts`; the cgroup reading is reused for a second; tests in `test/memory-hold.test.ts` drive the dispatcher, the check queue and vitals.
Not done here: a smaller heap for the child's Bun processes, reclaiming idle app cells, stopping the largest process instead of the forge.
@@ -92,6 +92,7 @@
import { transact } from "../kernel/db.ts";
import { createLog } from "../kernel/log.ts";
import { holdLease, STALE_AFTER_MS } from "../runs/lease.ts";
import { appendRunLog } from "../runs/log.ts";
import { memoryHold } from "../runs/memory_hold.ts";
import { cancelledLine } from "../runs/cancelled.ts";
import { trackRun, untrackRun } from "../runs/stoppable.ts";
import { runningLimitSentence, runningRuns } from "../settings/quotas.ts";
@@ -145,6 +146,8 @@
const INTERRUPTED = "this forge is stopping";
export interface ChecksOptions {
  /** How many checks may run at once. @default 1 */
  readonly concurrency?: number | undefined;
  /** How full this forge's memory is, 0 to 1: its cgroup's unless a test says otherwise. */
  readonly pressure?: (() => number | undefined) | undefined;
  /** The forge's public URL, told to the check as `FORGE_URL`. */
  readonly forgeUrl: string;
  /** The parent's gate, for a hosted child: its checks take their turn in the parent's lane. */
@@ -489,6 +492,7 @@
export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
  const waiting = new Set<string>();
  /** Queued checks a full disk holds that have said so once (#11799); apart from `waiting`, so neither line hides the other. */
  const diskWaiting = new Set<string>();
  const memoryWaiting = new Set<string>();
  /** The verification-lane run in flight, which holds no check slot. */
  let verifying: string | undefined;
  const slotted = (): number => inFlight.size - (verifying === undefined ? 0 : 1);
@@ -582,6 +586,7 @@
export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
    if (inFlight.has(run.id)) return;
    waiting.delete(run.id);
    diskWaiting.delete(run.id);
    memoryWaiting.delete(run.id);
    const abort = new AbortController();
    const low = inVerifyLane(run.env);
    if (low) verifying = run.id;
@@ -642,6 +647,16 @@
export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
        }
      return undefined;
    }
    // A child near its memory limit runs one check at a time (`runs/memory_hold.ts`).
    const tight = memoryHold(slotted(), options.pressure);
      
        junipercrew
        1d ago
        outdated
      
      inFlight.size includes the verification-lane run, which holds no check slot (see slotted() above). Pass slotted() so a verification does not hold back a real check. Add a test that drives the queue under pressure.
    
    if (tight !== undefined) {
      for (const run of queued)
        if (!memoryWaiting.has(run.id)) {
          memoryWaiting.add(run.id);
          appendRunLog(app.config.dataDir, run.id, `queued, ${tight}`);
        }
      return undefined;
    }
    let rest = queued;
    const isApproved = approvedOnce(app.db);
    let next = pickNext(rest, Date.now(), isApproved);
@@ -46,6 +46,7 @@
import { resolveOpen } from "../core/decisions.ts";
import { accountActive } from "../core/accounts.ts";
import { runningLimitSentence, runningRuns } from "../settings/quotas.ts";
import { appendRunLog } from "../runs/log.ts";
import { memoryHold } from "../runs/memory_hold.ts";
import { noteHeld, type HoldCause } from "../runs/held.ts";
import { decideAndRecord } from "./ledger.ts";
import { bugsFirst } from "../flow/bugs.ts";
@@ -1702,6 +1703,17 @@
export function pump(
      }
      continue;
    }
    // A child near its memory limit starts nothing new while a run is going
    // (`runs/memory_hold.ts`).
    const tight = memoryHold(inFlight.size + started.length, options.pressure);
    if (tight !== undefined) {
      held.set(fresh.id, "memory");
      if (!noted.has(fresh.id)) {
        noted.add(fresh.id);
        appendRunLog(app.config.dataDir, fresh.id, `queued, not started: ${tight}`);
      }
      continue;
    }
    // Out of slots: nothing more starts. The runs past this point are still
    // read, so those a policy above holds are named for vitals, and only the
    // rest count as waiting for a slot (#11518).
@@ -211,6 +211,8 @@
export interface CrewOptions {
  readonly localUrl: string;
  readonly binary: string;
  readonly concurrency: number;
  /** How full this forge's memory is, 0 to 1: its cgroup's unless a test says otherwise. */
  readonly pressure?: (() => number | undefined) | undefined;
  /** Spend per run when the agent's own budget is 0. */
  readonly budgetUsd: number;
  readonly runTimeoutMs: number;
@@ -17,7 +17,7 @@
import type { Database } from "bun:sqlite";
export const HELD_FRESH_MS = 2 * 60_000;
/** Why the dispatcher held a run. */
export type HoldCause = "valve" | "pace" | "limit" | "maintenance" | "issue";
export type HoldCause = "valve" | "pace" | "limit" | "maintenance" | "issue" | "memory";
/** How vitals names each cause: "2 by a shut valve". */
export const HOLD_WORDS: Readonly<Record<HoldCause, string>> = {
@@ -26,6 +26,7 @@
export const HOLD_WORDS: Readonly<Record<HoldCause, string>> = {
  limit: "an account limit",
  maintenance: "planned maintenance",
  issue: "its own issue's run still going",
  memory: "the forge's memory near its limit",
};
interface Pass {
@@ -0,0 +1,69 @@
/**
 * Memory is held, not hit (#11980). A child forge in a cell has one memory
 * limit for itself, its crew run, its checks and its apps. Near that limit a
 * new process is the straw: the host stops the whole forge for it. So while
 * the child's anonymous memory is at or over `HOLD_SHARE` of its limit and
 * something of its own is already running, nothing new starts (the dispatcher
 * keeps an agent run queued, the check queue keeps a check queued) until the
 * running work gives memory back. With nothing running a start is always
 * allowed, so a hold can never be permanent.
 *
 * Only a cell child reads its pressure (`FORGE_CELL_SIZE` is in its
 * environment); on any other forge the answer is always "not held".
 */
import { readFileSync } from "node:fs";
import { join } from "node:path";
import { memoryPressure } from "../looks/service.ts";
import { isSize } from "../spawn/sizes.ts";
import { CGROUP_DIR_ENV, SIZE_ENV } from "../spawn/usage.ts";
/** At this share of the limit, in anonymous memory, new work waits. */
export const HOLD_SHARE = 0.75;
export type Pressure = () => number | undefined;
/** A reading is reused this long: the pump and the check tick ask often, the files are on disk. */
export const READING_MS = 1_000;
let last: { readonly base: string; readonly at: number; readonly share: number | undefined } | undefined;
/** How much of its limit this child's cgroup holds in anonymous memory; none off a cell child. */
export function childPressure(
  env: Readonly<Record<string, string | undefined>> = process.env,
  now: number = Date.now(),
): number | undefined {
  if (!isSize(env[SIZE_ENV])) return undefined;
  const dir = env[CGROUP_DIR_ENV];
  const base = dir !== undefined && dir.startsWith("/") ? dir : "/sys/fs/cgroup";
  if (last !== undefined && last.base === base && now >= last.at && now - last.at < READING_MS) return last.share;
  const share = readShare(base);
  last = { base, at: now, share };
  return share;
}
function readShare(base: string): number | undefined {
  const read = (name: string): string => readFileSync(join(base, name), "utf8");
  try {
    let stat = "";
    try {
      stat = read("memory.stat");
    } catch {
      // The total will have to do.
    }
    return memoryPressure(stat, read("memory.max"), read("memory.current"));
  } catch {
    return undefined;
  }
}
/**
 * The line saying why nothing new starts, or undefined when something may.
 * `running` is how many of this forge's own runs are going now.
 */
export function memoryHold(running: number, pressure: Pressure = childPressure): string | undefined {
  if (running <= 0) return undefined;
  const share = pressure();
  if (share === undefined || share < HOLD_SHARE) return undefined;
  return `this forge holds ${Math.round(share * 100)}% of its memory; new work waits for a running one to finish`;
}
@@ -0,0 +1,157 @@
import { afterAll, beforeAll, describe, expect, test } from "bun:test";
import { chmodSync, writeFileSync } from "node:fs";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp, type App } from "../src/app.ts";
import { commitToBranch } from "../src/actions/repos.ts";
import { startChecks } from "../src/checks/service.ts";
import { createRun, getRun } from "../src/core/runs.ts";
import { createUser } from "../src/core/users.ts";
import { pump } from "../src/crew/dispatch.ts";
import type { CrewOptions } from "../src/crew/runner.ts";
import { QUEUE_WAIT_MAX_MS, vitals } from "../src/kernel/vitals.ts";
import { heldRuns, HOLD_WORDS } from "../src/runs/held.ts";
import { readRunLog } from "../src/runs/log.ts";
import {
  childPressure,
  HOLD_SHARE,
  memoryHold,
  READING_MS,
} from "../src/runs/memory_hold.ts";
import { CGROUP_DIR_ENV, SIZE_ENV } from "../src/spawn/usage.ts";
      
        junipercrew
        1d ago
        outdated
      
      Add tests that drive pump() and the check queue with a pressure stub. As written, deleting the hold from dispatch.ts and service.ts leaves every test green.
    
import { checkJson, checksForge } from "./helpers/checks-forge.ts";
describe("memory is held, not hit: the rule", () => {
  test("nothing running: a start is always allowed, however full", () => {
    expect(memoryHold(0, () => 0.99)).toBeUndefined();
  });
  test("work running and memory near the limit: new work waits, with a sentence", () => {
    expect(memoryHold(1, () => HOLD_SHARE)).toContain("75%");
  });
  test("work running and memory below the share, or unknown: starts", () => {
    expect(memoryHold(2, () => 0.5)).toBeUndefined();
    expect(memoryHold(2, () => undefined)).toBeUndefined();
  });
  test("a forge that is no cell child reads no pressure", () => {
    expect(childPressure({})).toBeUndefined();
  });
  test("the cgroup is read once a second, not on every pump", async () => {
    const dir = await mkdtemp(join(tmpdir(), "forge-cgroup-"));
    try {
      const env = { [SIZE_ENV]: "small", [CGROUP_DIR_ENV]: dir };
      writeFileSync(join(dir, "memory.max"), "1000\n");
      writeFileSync(join(dir, "memory.current"), "800\n");
      const at = 5_000_000_000_000;
      expect(childPressure(env, at)).toBe(0.8);
      writeFileSync(join(dir, "memory.current"), "100\n");
      expect(childPressure(env, at + READING_MS - 1)).toBe(0.8);
      expect(childPressure(env, at + READING_MS)).toBe(0.1);
    } finally {
      await rm(dir, { recursive: true, force: true });
    }
  });
});
describe("the dispatcher holds a queued run under pressure", () => {
  let dataDir: string;
  let app: App;
  let repoId = "";
  const roomy = () => ({ freeBytes: 100, totalBytes: 100 });
  beforeAll(async () => {
    dataDir = await mkdtemp(join(tmpdir(), "forge-memory-hold-"));
    app = createApp({ dataDir, port: 0, disk: roomy });
    const owner = await createUser(app.db, {
      origin: "seed",
      trusted: true,
      username: "holder",
      email: "[email protected]",
      password: "a-good-passphrase",
    });
    if (!owner.ok) throw new Error(String(owner.error));
    repoId = crypto.randomUUID();
    app.db
      .query("INSERT INTO repo (id, owner_id, name, visibility, created_at) VALUES (?1, ?2, 'mem', 'private', ?3)")
      .run(repoId, owner.value.id, new Date().toISOString());
  });
  afterAll(async () => {
    app.close();
    await rm(dataDir, { recursive: true, force: true });
  });
  test("a queued run stays queued and held as memory, is named in vitals, and the pump says so once", () => {
    const fake = join(dataDir, "never-run");
    writeFileSync(fake, "#!/bin/sh\nexit 1\n");
    chmodSync(fake, 0o755);
    const options = {
      forgeUrl: "http://127.0.0.1:1",
      localUrl: "http://127.0.0.1:1",
      binary: fake,
      concurrency: 3,
      budgetUsd: 1,
      runTimeoutMs: 60_000,
      pressure: () => 0.9,
    } satisfies Partial<CrewOptions> as unknown as CrewOptions;
    const queued = createRun(app.db, { repoId, kind: "agent" });
    const hourAgo = new Date(Date.now() - 2 * QUEUE_WAIT_MAX_MS).toISOString();
    app.db.query("UPDATE run SET created_at = ?1 WHERE id = ?2").run(hourAgo, queued.id);
    // One of this forge's own runs is going.
    const going = new Map<string, Promise<void>>([["going", new Promise<void>(() => {})]]);
    pump(app, options, going);
    pump(app, options, going);
    expect(getRun(app.db, queued.id)?.state).toBe("queued");
    expect(heldRuns(app.db).get(queued.id)).toBe("memory");
    const log = readRunLog(dataDir, queued.id).lines.join("\n");
    expect(log).toContain("this forge holds 90% of its memory");
    expect(log.split("this forge holds").length - 1).toBe(1);
    // Vitals counts it as held on purpose, not as waiting: all held is no stalled queue ...
    const now = Date.now() + 1_000;
    const card = () =>
      vitals({ db: app.db, signals: app.signals, dataDir, disk: roomy, now }).degraded.find(
        (d) => d.cause === "queue:agent",
      );
    expect(card()).toBeUndefined();
    // ... and with another run waiting beside it, the card names the hold.
    const other = createRun(app.db, { repoId, kind: "agent" });
    app.db.query("UPDATE run SET created_at = ?1 WHERE id = ?2").run(hourAgo, other.id);
    expect(card()?.sentence).toContain(`held on purpose: 1 by ${HOLD_WORDS.memory}.`);
  });
});
describe("the check queue holds a second check under pressure", () => {
  const f = checksForge("forge-memory-checks-");
  let share = 0.9;
  test("a second check stays queued with the log line while one is in flight, and starts once the pressure drops", async () => {
    await f.checks.stop();
    f.checks = startChecks(f.app, {
      forgeUrl: f.servers.url,
      concurrency: 2,
      tickMs: 20,
      pressure: () => share,
    });
    const first = await commitToBranch(f.app, f.alice, f.repo, {
      message: "Holds a slot",
      files: [{ path: "forge.json", content: checkJson(f.held("mem-first")) }],
    });
    if (!first.ok) throw new Error(JSON.stringify(first.error));
    await f.until(() => f.checksOf(first.value.commitOid)[0]?.state === "running", "first running");
    const second = await commitToBranch(f.app, f.alice, f.repo, {
      message: "Waits for memory",
      files: [{ path: "forge.json", content: checkJson("test -f OK.md") }],
    });
    if (!second.ok) throw new Error(JSON.stringify(second.error));
    await f.until(() => f.checksOf(second.value.commitOid).length > 0, "second queued");
    const id = f.checksOf(second.value.commitOid)[0]!.id;
    await f.until(() => readRunLog(f.dataDir, id).lines.join("\n").includes("this forge holds 90%"), "the hold said");
    await Bun.sleep(100); // real-timer: a hold is a start that never comes; let several ticks pass
    expect(getRun(f.app.db, id)?.state).toBe("queued");
    share = 0.1;
    const done = await f.settled((r) => r.id === id);
    expect(done.state).toBe("ok");
    f.release("mem-first");
  });
});
		@@ -0,0 +1,10 @@
	1	# Memory held, not hit (#11980)
	2
	3	A child forge in a cell has one memory limit for itself, its crew run, its checks and its apps. At or over 75% of that limit (anonymous memory, the page cache not counted) and with one of its own runs already going, the child starts nothing new:
	4
	5	- the dispatcher keeps the queued agent run queued, with one log line, and vitals counts it as "held on purpose: … by the forge's memory near its limit", not as waiting;
	6	- the check queue keeps checks queued, one log line each, so checks run one at a time.
	7
	8	With nothing running a start is always allowed, so the hold never lasts. A forge that is not a cell child reads no pressure and is never held. The rule is `src/runs/memory_hold.ts`; the cgroup reading is reused for a second; tests in `test/memory-hold.test.ts` drive the dispatcher, the check queue and vitals.
	9
	10	Not done here: a smaller heap for the child's Bun processes, reclaiming idle app cells, stopping the largest process instead of the forge.
		@@ -92,6 +92,7 @@ import { transact } from "../kernel/db.ts";
92	92	import { createLog } from "../kernel/log.ts";
93	93	import { holdLease, STALE_AFTER_MS } from "../runs/lease.ts";
94	94	import { appendRunLog } from "../runs/log.ts";
	95	import { memoryHold } from "../runs/memory_hold.ts";
95	96	import { cancelledLine } from "../runs/cancelled.ts";
96	97	import { trackRun, untrackRun } from "../runs/stoppable.ts";
97	98	import { runningLimitSentence, runningRuns } from "../settings/quotas.ts";
		@@ -145,6 +146,8 @@ const INTERRUPTED = "this forge is stopping";
145	146	export interface ChecksOptions {
146	147	/** How many checks may run at once. @default 1 */
147	148	readonly concurrency?: number \| undefined;
	149	/** How full this forge's memory is, 0 to 1: its cgroup's unless a test says otherwise. */
	150	readonly pressure?: (() => number \| undefined) \| undefined;
148	151	/** The forge's public URL, told to the check as `FORGE_URL`. */
149	152	readonly forgeUrl: string;
150	153	/** The parent's gate, for a hosted child: its checks take their turn in the parent's lane. */
		@@ -489,6 +492,7 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
489	492	const waiting = new Set<string>();
490	493	/** Queued checks a full disk holds that have said so once (#11799); apart from `waiting`, so neither line hides the other. */
491	494	const diskWaiting = new Set<string>();
	495	const memoryWaiting = new Set<string>();
492	496	/** The verification-lane run in flight, which holds no check slot. */
493	497	let verifying: string \| undefined;
494	498	const slotted = (): number => inFlight.size - (verifying === undefined ? 0 : 1);
		@@ -582,6 +586,7 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
582	586	if (inFlight.has(run.id)) return;
583	587	waiting.delete(run.id);
584	588	diskWaiting.delete(run.id);
	589	memoryWaiting.delete(run.id);
585	590	const abort = new AbortController();
586	591	const low = inVerifyLane(run.env);
587	592	if (low) verifying = run.id;
		@@ -642,6 +647,16 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
642	647	}
643	648	return undefined;
644	649	}
	650	// A child near its memory limit runs one check at a time (`runs/memory_hold.ts`).
	651	const tight = memoryHold(slotted(), options.pressure);
		junipercrew 1d ago outdated inFlight.size includes the verification-lane run, which holds no check slot (see slotted() above). Pass slotted() so a verification does not hold back a real check. Add a test that drives the queue under pressure.
	652	if (tight !== undefined) {
	653	for (const run of queued)
	654	if (!memoryWaiting.has(run.id)) {
	655	memoryWaiting.add(run.id);
	656	appendRunLog(app.config.dataDir, run.id, `queued, ${tight}`);
	657	}
	658	return undefined;
	659	}
645	660	let rest = queued;
646	661	const isApproved = approvedOnce(app.db);
647	662	let next = pickNext(rest, Date.now(), isApproved);
		@@ -46,6 +46,7 @@ import { resolveOpen } from "../core/decisions.ts";
46	46	import { accountActive } from "../core/accounts.ts";
47	47	import { runningLimitSentence, runningRuns } from "../settings/quotas.ts";
48	48	import { appendRunLog } from "../runs/log.ts";
	49	import { memoryHold } from "../runs/memory_hold.ts";
49	50	import { noteHeld, type HoldCause } from "../runs/held.ts";
50	51	import { decideAndRecord } from "./ledger.ts";
51	52	import { bugsFirst } from "../flow/bugs.ts";
		@@ -1702,6 +1703,17 @@ export function pump(
1702	1703	}
1703	1704	continue;
1704	1705	}
	1706	// A child near its memory limit starts nothing new while a run is going
	1707	// (`runs/memory_hold.ts`).
	1708	const tight = memoryHold(inFlight.size + started.length, options.pressure);
	1709	if (tight !== undefined) {
	1710	held.set(fresh.id, "memory");
	1711	if (!noted.has(fresh.id)) {
	1712	noted.add(fresh.id);
	1713	appendRunLog(app.config.dataDir, fresh.id, `queued, not started: ${tight}`);
	1714	}
	1715	continue;
	1716	}
1705	1717	// Out of slots: nothing more starts. The runs past this point are still
1706	1718	// read, so those a policy above holds are named for vitals, and only the
1707	1719	// rest count as waiting for a slot (#11518).
		@@ -211,6 +211,8 @@ export interface CrewOptions {
211	211	readonly localUrl: string;
212	212	readonly binary: string;
213	213	readonly concurrency: number;
	214	/** How full this forge's memory is, 0 to 1: its cgroup's unless a test says otherwise. */
	215	readonly pressure?: (() => number \| undefined) \| undefined;
214	216	/** Spend per run when the agent's own budget is 0. */
215	217	readonly budgetUsd: number;
216	218	readonly runTimeoutMs: number;
		@@ -17,7 +17,7 @@ import type { Database } from "bun:sqlite";
17	17	export const HELD_FRESH_MS = 2 * 60_000;
18	18
19	19	/** Why the dispatcher held a run. */
20		export type HoldCause = "valve" \| "pace" \| "limit" \| "maintenance" \| "issue";
	20	export type HoldCause = "valve" \| "pace" \| "limit" \| "maintenance" \| "issue" \| "memory";
21	21
22	22	/** How vitals names each cause: "2 by a shut valve". */
23	23	export const HOLD_WORDS: Readonly<Record<HoldCause, string>> = {
		@@ -26,6 +26,7 @@ export const HOLD_WORDS: Readonly<Record<HoldCause, string>> = {
26	26	limit: "an account limit",
27	27	maintenance: "planned maintenance",
28	28	issue: "its own issue's run still going",
	29	memory: "the forge's memory near its limit",
29	30	};
30	31
31	32	interface Pass {
		@@ -0,0 +1,69 @@
	1	/**
	2	* Memory is held, not hit (#11980). A child forge in a cell has one memory
	3	* limit for itself, its crew run, its checks and its apps. Near that limit a
	4	* new process is the straw: the host stops the whole forge for it. So while
	5	* the child's anonymous memory is at or over `HOLD_SHARE` of its limit and
	6	* something of its own is already running, nothing new starts (the dispatcher
	7	* keeps an agent run queued, the check queue keeps a check queued) until the
	8	* running work gives memory back. With nothing running a start is always
	9	* allowed, so a hold can never be permanent.
	10	*
	11	* Only a cell child reads its pressure (`FORGE_CELL_SIZE` is in its
	12	* environment); on any other forge the answer is always "not held".
	13	*/
	14
	15	import { readFileSync } from "node:fs";
	16	import { join } from "node:path";
	17	import { memoryPressure } from "../looks/service.ts";
	18	import { isSize } from "../spawn/sizes.ts";
	19	import { CGROUP_DIR_ENV, SIZE_ENV } from "../spawn/usage.ts";
	20
	21	/** At this share of the limit, in anonymous memory, new work waits. */
	22	export const HOLD_SHARE = 0.75;
	23
	24	export type Pressure = () => number \| undefined;
	25
	26	/** A reading is reused this long: the pump and the check tick ask often, the files are on disk. */
	27	export const READING_MS = 1_000;
	28
	29	let last: { readonly base: string; readonly at: number; readonly share: number \| undefined } \| undefined;
	30
	31	/** How much of its limit this child's cgroup holds in anonymous memory; none off a cell child. */
	32	export function childPressure(
	33	env: Readonly<Record<string, string \| undefined>> = process.env,
	34	now: number = Date.now(),
	35	): number \| undefined {
	36	if (!isSize(env[SIZE_ENV])) return undefined;
	37	const dir = env[CGROUP_DIR_ENV];
	38	const base = dir !== undefined && dir.startsWith("/") ? dir : "/sys/fs/cgroup";
	39	if (last !== undefined && last.base === base && now >= last.at && now - last.at < READING_MS) return last.share;
	40	const share = readShare(base);
	41	last = { base, at: now, share };
	42	return share;
	43	}
	44
	45	function readShare(base: string): number \| undefined {
	46	const read = (name: string): string => readFileSync(join(base, name), "utf8");
	47	try {
	48	let stat = "";
	49	try {
	50	stat = read("memory.stat");
	51	} catch {
	52	// The total will have to do.
	53	}
	54	return memoryPressure(stat, read("memory.max"), read("memory.current"));
	55	} catch {
	56	return undefined;
	57	}
	58	}
	59
	60	/**
	61	* The line saying why nothing new starts, or undefined when something may.
	62	* `running` is how many of this forge's own runs are going now.
	63	*/
	64	export function memoryHold(running: number, pressure: Pressure = childPressure): string \| undefined {
	65	if (running <= 0) return undefined;
	66	const share = pressure();
	67	if (share === undefined \|\| share < HOLD_SHARE) return undefined;
	68	return `this forge holds ${Math.round(share * 100)}% of its memory; new work waits for a running one to finish`;
	69	}
		@@ -0,0 +1,157 @@
	1	import { afterAll, beforeAll, describe, expect, test } from "bun:test";
	2	import { chmodSync, writeFileSync } from "node:fs";
	3	import { mkdtemp, rm } from "node:fs/promises";
	4	import { tmpdir } from "node:os";
	5	import { join } from "node:path";
	6	import { createApp, type App } from "../src/app.ts";
	7	import { commitToBranch } from "../src/actions/repos.ts";
	8	import { startChecks } from "../src/checks/service.ts";
	9	import { createRun, getRun } from "../src/core/runs.ts";
	10	import { createUser } from "../src/core/users.ts";
	11	import { pump } from "../src/crew/dispatch.ts";
	12	import type { CrewOptions } from "../src/crew/runner.ts";
	13	import { QUEUE_WAIT_MAX_MS, vitals } from "../src/kernel/vitals.ts";
	14	import { heldRuns, HOLD_WORDS } from "../src/runs/held.ts";
	15	import { readRunLog } from "../src/runs/log.ts";
	16	import {
	17	childPressure,
	18	HOLD_SHARE,
	19	memoryHold,
	20	READING_MS,
	21	} from "../src/runs/memory_hold.ts";
	22	import { CGROUP_DIR_ENV, SIZE_ENV } from "../src/spawn/usage.ts";
		junipercrew 1d ago outdated Add tests that drive pump() and the check queue with a pressure stub. As written, deleting the hold from dispatch.ts and service.ts leaves every test green.
	23	import { checkJson, checksForge } from "./helpers/checks-forge.ts";
	24
	25	describe("memory is held, not hit: the rule", () => {
	26	test("nothing running: a start is always allowed, however full", () => {
	27	expect(memoryHold(0, () => 0.99)).toBeUndefined();
	28	});
	29	test("work running and memory near the limit: new work waits, with a sentence", () => {
	30	expect(memoryHold(1, () => HOLD_SHARE)).toContain("75%");
	31	});
	32	test("work running and memory below the share, or unknown: starts", () => {
	33	expect(memoryHold(2, () => 0.5)).toBeUndefined();
	34	expect(memoryHold(2, () => undefined)).toBeUndefined();
	35	});
	36	test("a forge that is no cell child reads no pressure", () => {
	37	expect(childPressure({})).toBeUndefined();
	38	});
	39	test("the cgroup is read once a second, not on every pump", async () => {
	40	const dir = await mkdtemp(join(tmpdir(), "forge-cgroup-"));
	41	try {
	42	const env = { [SIZE_ENV]: "small", [CGROUP_DIR_ENV]: dir };
	43	writeFileSync(join(dir, "memory.max"), "1000\n");
	44	writeFileSync(join(dir, "memory.current"), "800\n");
	45	const at = 5_000_000_000_000;
	46	expect(childPressure(env, at)).toBe(0.8);
	47	writeFileSync(join(dir, "memory.current"), "100\n");
	48	expect(childPressure(env, at + READING_MS - 1)).toBe(0.8);
	49	expect(childPressure(env, at + READING_MS)).toBe(0.1);
	50	} finally {
	51	await rm(dir, { recursive: true, force: true });
	52	}
	53	});
	54	});
	55
	56	describe("the dispatcher holds a queued run under pressure", () => {
	57	let dataDir: string;
	58	let app: App;
	59	let repoId = "";
	60	const roomy = () => ({ freeBytes: 100, totalBytes: 100 });
	61
	62	beforeAll(async () => {
	63	dataDir = await mkdtemp(join(tmpdir(), "forge-memory-hold-"));
	64	app = createApp({ dataDir, port: 0, disk: roomy });
	65	const owner = await createUser(app.db, {
	66	origin: "seed",
	67	trusted: true,
	68	username: "holder",
	69	email: "[email protected]",
	70	password: "a-good-passphrase",
	71	});
	72	if (!owner.ok) throw new Error(String(owner.error));
	73	repoId = crypto.randomUUID();
	74	app.db
	75	.query("INSERT INTO repo (id, owner_id, name, visibility, created_at) VALUES (?1, ?2, 'mem', 'private', ?3)")
	76	.run(repoId, owner.value.id, new Date().toISOString());
	77	});
	78	afterAll(async () => {
	79	app.close();
	80	await rm(dataDir, { recursive: true, force: true });
	81	});
	82
	83	test("a queued run stays queued and held as memory, is named in vitals, and the pump says so once", () => {
	84	const fake = join(dataDir, "never-run");
	85	writeFileSync(fake, "#!/bin/sh\nexit 1\n");
	86	chmodSync(fake, 0o755);
	87	const options = {
	88	forgeUrl: "http://127.0.0.1:1",
	89	localUrl: "http://127.0.0.1:1",
	90	binary: fake,
	91	concurrency: 3,
	92	budgetUsd: 1,
	93	runTimeoutMs: 60_000,
	94	pressure: () => 0.9,
	95	} satisfies Partial<CrewOptions> as unknown as CrewOptions;
	96	const queued = createRun(app.db, { repoId, kind: "agent" });
	97	const hourAgo = new Date(Date.now() - 2 * QUEUE_WAIT_MAX_MS).toISOString();
	98	app.db.query("UPDATE run SET created_at = ?1 WHERE id = ?2").run(hourAgo, queued.id);
	99	// One of this forge's own runs is going.
	100	const going = new Map<string, Promise<void>>([["going", new Promise<void>(() => {})]]);
	101
	102	pump(app, options, going);
	103	pump(app, options, going);
	104	expect(getRun(app.db, queued.id)?.state).toBe("queued");
	105	expect(heldRuns(app.db).get(queued.id)).toBe("memory");
	106	const log = readRunLog(dataDir, queued.id).lines.join("\n");
	107	expect(log).toContain("this forge holds 90% of its memory");
	108	expect(log.split("this forge holds").length - 1).toBe(1);
	109
	110	// Vitals counts it as held on purpose, not as waiting: all held is no stalled queue ...
	111	const now = Date.now() + 1_000;
	112	const card = () =>
	113	vitals({ db: app.db, signals: app.signals, dataDir, disk: roomy, now }).degraded.find(
	114	(d) => d.cause === "queue:agent",
	115	);
	116	expect(card()).toBeUndefined();
	117	// ... and with another run waiting beside it, the card names the hold.
	118	const other = createRun(app.db, { repoId, kind: "agent" });
	119	app.db.query("UPDATE run SET created_at = ?1 WHERE id = ?2").run(hourAgo, other.id);
	120	expect(card()?.sentence).toContain(`held on purpose: 1 by ${HOLD_WORDS.memory}.`);
	121	});
	122	});
	123
	124	describe("the check queue holds a second check under pressure", () => {
	125	const f = checksForge("forge-memory-checks-");
	126	let share = 0.9;
	127
	128	test("a second check stays queued with the log line while one is in flight, and starts once the pressure drops", async () => {
	129	await f.checks.stop();
	130	f.checks = startChecks(f.app, {
	131	forgeUrl: f.servers.url,
	132	concurrency: 2,
	133	tickMs: 20,
	134	pressure: () => share,
	135	});
	136	const first = await commitToBranch(f.app, f.alice, f.repo, {
	137	message: "Holds a slot",
	138	files: [{ path: "forge.json", content: checkJson(f.held("mem-first")) }],
	139	});
	140	if (!first.ok) throw new Error(JSON.stringify(first.error));
	141	await f.until(() => f.checksOf(first.value.commitOid)[0]?.state === "running", "first running");
	142	const second = await commitToBranch(f.app, f.alice, f.repo, {
	143	message: "Waits for memory",
	144	files: [{ path: "forge.json", content: checkJson("test -f OK.md") }],
	145	});
	146	if (!second.ok) throw new Error(JSON.stringify(second.error));
	147	await f.until(() => f.checksOf(second.value.commitOid).length > 0, "second queued");
	148	const id = f.checksOf(second.value.commitOid)[0]!.id;
	149	await f.until(() => readRunLog(f.dataDir, id).lines.join("\n").includes("this forge holds 90%"), "the hold said");
	150	await Bun.sleep(100); // real-timer: a hold is a start that never comes; let several ticks pass
	151	expect(getRun(f.app.db, id)?.state).toBe("queued");
	152	share = 0.1;
	153	const done = await f.settled((r) => r.id === id);
	154	expect(done.state).toBe("ok");
	155	f.release("mem-first");
	156	});
	157	});