07d6e1b · 86e4867 · 297012b · 05b9118 · 2596ebe · 94ed88f
@@ -18,6 +18,7 @@
import { chmod, lstat, readdir, rm } from "node:fs/promises";
import { join } from "node:path";
import type { App } from "../app.ts";
import { releaseOnGreen } from "../actions/holds.ts";
import { activeCrewRuns, crewHoldsCore, forgeCpus, holdClock } from "../core/one_core.ts";
import { announceRun } from "../actions/runs.ts";
import { announceShard } from "./shard_event.ts";
import type { ForgeEvent } from "../core/events.ts";
@@ -56,7 +57,6 @@
import { readCombined } from "../git/combined.ts";
import { approvedOnce } from "./approved_first.ts";
import { checkOnApproval } from "./approval.ts";
import { checkKey, provenBy, recordProven } from "./proven.ts";
import { crewHoldsCore, forgeCpus } from "../core/one_core.ts";
import { beginGate, endGate, gateHolding } from "./gate_record.ts";
import { giveHistory } from "./history.ts";
import { readLearned } from "./learned.ts";
@@ -145,10 +145,6 @@
export interface ChecksOptions {
  readonly concurrency?: number | undefined;
  /** The forge's public URL, told to the check as `FORGE_URL`. */
  readonly forgeUrl: string;
  /** The forge's CPU count; a 1-CPU forge holds pull checks for a crew run. @default forgeCpus */
  readonly cpus?: (() => number) | undefined;
  /** Crew runs active now. @default the active agent runs */
  readonly activeCrewRuns?: (() => number) | undefined;
  /** The parent's gate, for a hosted child: its checks take their turn in the parent's lane. */
  readonly gate?: Gate | undefined;
  /** Whose checks these are, written first in each run's log (a hosted child's name). */
@@ -189,6 +185,15 @@
export interface ChecksOptions {
   * to `VERIFY_EVERY_MS`. The forge sets it to `VERIFY_EVERY_MS`.
   */
  readonly verifyEveryMs?: number | undefined;
  /** Test seam: how often the queue is looked at. @default TICK_MS */
  readonly tickMs?: number | undefined;
  /** The forge's CPU count; a 1-CPU forge holds pull checks for a crew run. @default forgeCpus */
  readonly cpus?: (() => number) | undefined;
  /** Crew runs working now. @default the running agent runs */
  readonly activeCrewRuns?: (() => number) | undefined;
  /** Test seams: how long a held pull check waits at most, and the clock it is judged by. */
  readonly holdMaxMs?: number | undefined;
  readonly holdNow?: (() => number) | undefined;
}
/** A gate's answer from the lane: `interrupted` is no verdict (#3155). */
@@ -633,10 +638,20 @@
export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
    return next;
  }
  const pullHold = holdClock(options.holdMaxMs);
  function schedule(): void {
    if (stopped) return;
    const limit = capacity();
      
        junipercrew
        2d ago
        outdated
      
      This return skips startVerify() above and the rebase and main-check picks below. Hold only pull checks, and only for a bounded time.
    
    if (options.gate === undefined) gate.resize(limit);
    // On one CPU a pull's check waits while a crew run works, for at most
    // `holdMaxMs`; the verify lane, a rebase's check and a main check are the
    // pipeline and never wait (#11599).
    const crewHolds = crewHoldsCore((options.cpus ?? forgeCpus)(), options.activeCrewRuns ?? (() => activeCrewRuns(app.db)));
    // The clock counts one continuous hold: it starts again once no crew run
    // holds, and forgets checks that are no longer queued.
    if (!crewHolds) pullHold.reset();
    else pullHold.keep(new Set(listActive(app.db, "check").filter((run) => run.state === "queued").map((run) => run.id)));
    startVerify();
    // Every merge-queue gate in flight holds a slot, a line's first too (#6376).
    while (slotted() < limit - gatesInFlight()) {
@@ -652,25 +667,22 @@
export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
        (repoId) => redMain(app.db, repoId),
        options.redHoldMs ?? 0,
      );
      // One core: a crew run has it. A pull's check keeps its place and starts
      // once the run has ended; main's check (the merge gate) is never held.
      const crewHolds = crewHoldsCore(
        (options.cpus ?? forgeCpus)(),
        (options.activeCrewRuns ?? (() => listActive(app.db, "agent").length))(),
      );
      const runnable = crewHolds ? queued.filter((run) => run.issueId === null) : queued;
      // The limit is asked at the point of starting, one pick at a time:
      // a started check is in flight before the next is asked about, so a
      // burst of one account's checks cannot pass together. A rebase's
      // check goes before any other (`finishing_first.ts`, #6140).
      const { rebases, rest: others } = rebasesFirst(app.db, runnable);
      const { rebases, rest: others } = rebasesFirst(app.db, queued);
      // While a gate is about to start, the last free slot is its: a pull's
      // check waits for it, a rebase's and a main check do not.
      const held = holdsLastSlot(limit - gatesInFlight() - slotted(), gateQueued());
      const next = startable(rebases) ?? startable(held ? others.filter((run) => run.issueId === null) : others);
      const waiting = crewHolds
        ? others.filter((run) => run.issueId === null || !pullHold.held(run.id, (options.holdNow ?? Date.now)()))
        : others;
      const next = startable(rebases) ?? startable(held ? waiting.filter((run) => run.issueId === null) : waiting);
      if (!next) return;
      const repo = getRepoById(app.db, next.repoId);
      if (!repo) return;
      pullHold.release(next.id);
      start(next, repo);
    }
  }
@@ -1716,7 +1728,7 @@
export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
  const timer = setInterval(() => {
    if (!stopped) during("lane", schedule);
  }, TICK_MS);
  }, options.tickMs ?? TICK_MS);
  // Main is verified after landing on a schedule, when the forge says how often.
  const sweeper =
    options.verifyEveryMs === undefined
@@ -1,8 +1,9 @@
import type { Database } from "bun:sqlite";
import { availableParallelism } from "node:os";
/** True when a 1-CPU forge should hold background work for a crew run. */
export function crewHoldsCore(cpus: number, activeCrewRuns: number): boolean {
  return cpus <= 1 && activeCrewRuns > 0;
/** True when a 1-CPU forge should hold background work for a crew run; more CPUs never count the runs. */
export function crewHoldsCore(cpus: number, activeCrewRuns: () => number): boolean {
  return cpus <= 1 && activeCrewRuns() > 0;
}
/** The forge's own CPU count, at least 1. */
@@ -10,3 +11,39 @@
export function forgeCpus(): number {
  const n = availableParallelism();
  return Number.isFinite(n) && n >= 1 ? Math.floor(n) : 1;
}
/** Crew runs working now: agent runs in the `running` state. */
      
        junipercrew
        2d ago
        
      
      Looks counts active agent runs with listActive(app.db, 'agent'); this counts only state='running'. Keep one definition and use it in looks as well.
    
export function activeCrewRuns(db: Database): number {
  const row = db.query<{ n: number }, []>("SELECT COUNT(*) AS n FROM run WHERE kind = 'agent' AND state = 'running'").get();
  return row?.n ?? 0;
}
/** The longest a check or a snapshot waits for a crew run on one CPU before it starts anyway. */
export const MAX_HOLD_MS = 10 * 60_000;
/**
 * Whether work keyed `key` is still held at `now`. The first time it is held
 * starts its clock; once `maxMs` has passed it is held no longer, so crew runs
 * back to back cannot starve it. `release` forgets the key when the work starts,
 * `keep` forgets every key not named (work cancelled or superseded), and
 * `reset` forgets them all once no crew run holds, so a later run holds afresh.
 */
export function holdClock(maxMs = MAX_HOLD_MS) {
  const since = new Map<string, number>();
  return {
    held(key: string, now: number): boolean {
      const first = since.get(key) ?? now;
      since.set(key, first);
      return now - first < maxMs;
    },
    release(key: string): void {
      since.delete(key);
    },
    keep(keys: ReadonlySet<string>): void {
      for (const key of since.keys()) if (!keys.has(key)) since.delete(key);
    },
    reset(): void {
      since.clear();
    },
  };
}
@@ -408,7 +408,7 @@
export function startLooks(app: App, options: LooksOptions): LooksHandle {
      const cpus = options.cpus ?? forgeCpus;
      const active =
        options.activeCrewRuns ?? (() => listActive(app.db, "agent").length);
      if (crewHoldsCore(cpus(), active())) {
      if (crewHoldsCore(cpus(), active)) {
        queue.unshift(next);
        return;
      }
@@ -11,6 +11,7 @@
 */
import type { App } from "../app.ts";
import { activeCrewRuns, crewHoldsCore, forgeCpus, holdClock } from "../core/one_core.ts";
import { createLog } from "../kernel/log.ts";
import { firstSeen, reportVerdict, seedFirstSeen } from "./verdict.ts";
import {
@@ -36,8 +37,6 @@
import { isOffBox } from "./off_box.ts";
import { noteOffBox, recordSnapshotOutcome } from "./outcome.ts";
import { s3FromEnv, type Store } from "./store.ts";
import { during } from "../perf/jobs.ts";
import { crewHoldsCore, forgeCpus } from "../core/one_core.ts";
import { listActive } from "../core/runs.ts";
const log = createLog("snapshots");
@@ -109,6 +108,8 @@
const CATCH_UP_MS = 5 * 60_000;
const SNAPSHOT_CATCH_UP_MS = 2 * 60_000;
/** A snapshot still owed is looked for again this often; one boot's look is not the last (#10167). */
const SNAPSHOT_CHECK_MS = 30 * 60_000;
/** A snapshot held for a crew run on one CPU looks again this often. */
const HOLD_RETRY_MS = 30_000;
export function untilDrill(now = Date.now()): number {
  const next = new Date(now);
@@ -153,9 +154,11 @@
export function startSnapshots(
    readonly bucketFirstMs?: number | undefined;
    /** Test seam: the clock `takeIfDue` judges by. */
    readonly now?: (() => number) | undefined;
    /** Test seams: the forge's CPU count and how many crew runs are active (1 CPU: a run holds the core). */
    /** Test seams: the forge's CPU count, crew runs working now, how soon a held snapshot looks again, and how long it waits at most. */
    readonly cpus?: (() => number) | undefined;
    readonly activeCrewRuns?: (() => number) | undefined;
    readonly holdRetryMs?: number | undefined;
    readonly holdMaxMs?: number | undefined;
    /** Test seam: how soon the first nightly timer fires. */
    readonly nightlyInMs?: number | undefined;
  } = {
@@ -178,6 +181,17 @@
export function startSnapshots(
    return resolvedMirror.store;
  };
  let stopped = false;
  // On one CPU a due snapshot waits for the crew run to end, for at most
  // `holdMaxMs`, then is taken anyway so back-to-back runs cannot starve it (#11599).
  const wait = holdClock(options.holdMaxMs ?? MAX_DEFER_MS);
  const crewHolds = (): boolean => {
    if (!crewHoldsCore((options.cpus ?? forgeCpus)(), options.activeCrewRuns ?? (() => activeCrewRuns(app.db)))) {
      wait.reset();
      return false;
    }
    return wait.held("snapshot", (options.now ?? Date.now)());
  };
  const holdRetryMs = options.holdRetryMs ?? options.snapshotCheckMs ?? HOLD_RETRY_MS;
  let inFlight: Promise<Manifest> | undefined;
  let active = 0;
  let running: Progress | undefined;
@@ -189,6 +203,7 @@
export function startSnapshots(
    let offBox: boolean | undefined;
    // After the one before it, however that ended: chained on a failure as it
    // was, every later run failed at once with the old error until a restart.
    wait.release("snapshot");
    const run = (inFlight ?? Promise.resolve()).catch(() => {}).then(async () => {
      // Resolved per run, so a destination saved on the Backups page is the
      // very next snapshot's.
@@ -271,26 +286,6 @@
export function startSnapshots(
    }
  };
  // One core: a crew run has it, so a due snapshot waits and is taken on the
  // next tick after the run ends (src/core/one_core.ts). Never forever: after
  // MAX_DEFER_MS of waiting it is taken anyway. A snapshot running is untouched.
  let deferredSince: number | undefined;
  const deferred = (): boolean => {
    const at = (options.now ?? Date.now)();
    const held = crewHoldsCore(
      (options.cpus ?? forgeCpus)(),
      (options.activeCrewRuns ?? (() => listActive(app.db, "agent").length))(),
    );
    if (!held) {
      deferredSince = undefined;
      return false;
    }
    deferredSince ??= at;
    if (at - deferredSince >= MAX_DEFER_MS) return false;
    log.info("snapshot due; a crew run holds the one core, waiting");
    return true;
  };
  // A restart takes the night's snapshot if the store does not hold it: a
  // release at 08:21 killed the one in flight and nothing took it again (#8742).
  // The manifest is written last, so a snapshot cut short is not one.
@@ -305,9 +300,7 @@
export function startSnapshots(
      log.warn("snapshot catch-up: store unreadable, not taking one now", { error: String(error) });
      return false;
    }
    if (stopped || active > 0 || !snapshotDue(newest?.finishedAt, (options.now ?? Date.now)())) return false;
    if (deferred()) return false;
    deferredSince = undefined;
    if (stopped || active > 0 || !snapshotDue(newest?.finishedAt, (options.now ?? Date.now)()) || crewHolds()) return false;
    const cut = await cutShort(store, newest?.name);
    log.info("snapshot due or lost; taking one", { newest: newest?.name ?? "none", resuming: cut?.toISOString() ?? "no" });
    await take(cut).catch(() => {});
@@ -411,18 +404,12 @@
export function startSnapshots(
    if (timer !== undefined) clearTimeout(timer);
    timer = undefined;
    if (options.nightly === false || stopped) return;
    timer = setTimeout(
      () =>
        void during("snapshot", () => {
          if (active === 0 && deferred()) {
            arm(options.snapshotCheckMs ?? SNAPSHOT_CHECK_MS);
            return Promise.resolve();
          }
          deferredSince = undefined;
          return take().then(() => {});
        }).catch(() => {}),
      ms,
    );
    timer = setTimeout(() => {
      void during("snapshot", async () => {
        if (active === 0 && crewHolds()) return arm(holdRetryMs);
        await take();
      }).catch(() => {});
    }, ms);
  }
  arm(options.nightlyInMs ?? untilNightly());
@@ -496,7 +483,7 @@
export function startSnapshots(
        .catch(() => {})
        .finally(() => {
          bootLookAt = undefined;
          scheduleCatchUp(options.snapshotCheckMs ?? SNAPSHOT_CHECK_MS);
          scheduleCatchUp(crewHolds() ? holdRetryMs : (options.snapshotCheckMs ?? SNAPSHOT_CHECK_MS));
        });
    }, ms);
  };
@@ -0,0 +1,190 @@
/** #11599: on one CPU a check or a snapshot waits for the crew run; two CPUs start them at once. */
import { afterEach, beforeEach, expect, test } from "bun:test";
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 { startChecks, type ChecksHandle } from "../src/checks/service.ts";
import { createRepo } from "../src/actions/repos.ts";
import { createIssue } from "../src/core/issues.ts";
import { createRun, getRun } from "../src/core/runs.ts";
import { createUser } from "../src/core/users.ts";
import { startSnapshots, type SnapshotsHandle } from "../src/snapshot/service.ts";
import { dirStore } from "../src/snapshot/store.ts";
import { holdClock } from "../src/core/one_core.ts";
import { quietLogs } from "./helpers/globals.ts";
quietLogs();
let dir: string;
let app: App;
let checks: ChecksHandle | undefined;
let snaps: SnapshotsHandle | undefined;
let seeded: { userId: string; repoId: string } | undefined;
beforeEach(async () => {
  dir = await mkdtemp(join(tmpdir(), "forge-one-core-"));
  app = createApp({ dataDir: dir, port: 0 });
});
afterEach(async () => {
  await checks?.stop({ drain: false });
  await snaps?.stop();
  checks = snaps = seeded = undefined;
  app.close();
  await rm(dir, { recursive: true, force: true });
});
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
async function until(ok: () => boolean, ms = 3000): Promise<boolean> {
  for (const end = Date.now() + ms; Date.now() < end; await sleep(20)) if (ok()) return true;
  return ok();
}
async function queuedCheck(opts: { pull?: boolean; env?: string; reason?: string; rebase?: boolean } = {}): Promise<string> {
  if (seeded === undefined) {
    const made = await createUser(app.db, { origin: "seed", trusted: true, username: "alice", email: "[email protected]", password: "a-good-passphrase", adminIfFirst: true });
    if (!made.ok) throw new Error(made.error);
    const repo = await createRepo(app, made.value, { name: "lib", visibility: "public" });
    if (!repo.ok) throw new Error(String(repo.error));
    seeded = { userId: made.value.id, repoId: repo.value.id };
  }
  const issue = createIssue(app.db, { repoId: seeded.repoId, authorId: seeded.userId, title: "change" });
  if (!issue.ok) throw new Error(issue.error);
  if (opts.rebase) createRun(app.db, { repoId: seeded.repoId, kind: "agent", issueId: issue.value.id, reason: "conflict", commitOid: "0".repeat(40) });
  return createRun(app.db, { repoId: seeded.repoId, kind: "check", env: opts.env ?? "", commitOid: "0".repeat(40), reason: opts.reason ?? "test", ...(opts.pull === false ? {} : { issueId: issue.value.id }) }).id;
}
test("1 CPU with a crew run holds a queued check, and it starts once the run is gone", async () => {
  const id = await queuedCheck();
  let crew = 1;
  checks = startChecks(app, { forgeUrl: "http://127.0.0.1:9", concurrency: 1, cpus: () => 1, activeCrewRuns: () => crew, tickMs: 25 });
  await sleep(200);
  expect(getRun(app.db, id)!.state).toBe("queued");
  crew = 0;
  expect(await until(() => getRun(app.db, id)!.state !== "queued")).toBe(true);
});
test("2 CPUs start a queued check at once, whatever the crew does", async () => {
  const id = await queuedCheck();
  checks = startChecks(app, { forgeUrl: "http://127.0.0.1:9", concurrency: 1, cpus: () => 2, activeCrewRuns: () => 1, tickMs: 25 });
  expect(await until(() => getRun(app.db, id)!.state !== "queued")).toBe(true);
});
test("1 CPU with a crew run does not hold a main check, a rebase check or a verify run", async () => {
  const main = await queuedCheck({ pull: false });
  checks = startChecks(app, { forgeUrl: "http://127.0.0.1:9", concurrency: 1, cpus: () => 1, activeCrewRuns: () => 1, tickMs: 25 });
  expect(await until(() => getRun(app.db, main)!.state !== "queued")).toBe(true);
});
test("1 CPU: a verify-lane run starts with a crew run active", async () => {
  const verify = await queuedCheck({ pull: false, env: "verify" });
  checks = startChecks(app, { forgeUrl: "http://127.0.0.1:9", concurrency: 1, cpus: () => 1, activeCrewRuns: () => 1, tickMs: 25 });
  expect(await until(() => getRun(app.db, verify)!.state !== "queued")).toBe(true);
});
test("1 CPU: a rebase's check starts with a crew run active, a pull's check beside it waits", async () => {
  const rebase = await queuedCheck({ rebase: true });
  const pull = await queuedCheck();
  checks = startChecks(app, { forgeUrl: "http://127.0.0.1:9", concurrency: 2, cpus: () => 1, activeCrewRuns: () => 1, tickMs: 25 });
  expect(await until(() => getRun(app.db, rebase)!.state !== "queued")).toBe(true);
  expect(getRun(app.db, pull)!.state).toBe("queued");
});
test("1 CPU: a pull's check held past the maximum wait starts anyway", async () => {
  const id = await queuedCheck();
  let now = 1_000;
  checks = startChecks(app, { forgeUrl: "http://127.0.0.1:9", concurrency: 1, cpus: () => 1, activeCrewRuns: () => 1, tickMs: 25, holdMaxMs: 5_000, holdNow: () => now });
  await sleep(150);
  expect(getRun(app.db, id)!.state).toBe("queued");
  now += 5_000;
  expect(await until(() => getRun(app.db, id)!.state !== "queued")).toBe(true);
});
test("1 CPU: a check held, then idle, is held afresh when the next crew run comes", async () => {
  const id = await queuedCheck();
  let now = 1_000;
  let crew = 1;
  checks = startChecks(app, { forgeUrl: "http://127.0.0.1:9", concurrency: 1, cpus: () => 1, activeCrewRuns: () => crew, tickMs: 25, holdMaxMs: 5_000, holdNow: () => now });
  await sleep(100);
  expect(getRun(app.db, id)!.state).toBe("queued");
  // Kept queued by something else (a not-before) while no crew run works.
  app.db.run("UPDATE run SET not_before = ? WHERE id = ?", [new Date(Date.now() + 60_000).toISOString(), id]);
  now += 4_000;
  crew = 0;
  await sleep(100);
  // The crew returns past the first clock's bound: a fresh hold, not a release.
  crew = 1;
  now += 2_000;
  app.db.run("UPDATE run SET not_before = NULL WHERE id = ?", [id]);
  await sleep(150);
  expect(getRun(app.db, id)!.state).toBe("queued");
  now += 5_000;
  expect(await until(() => getRun(app.db, id)!.state !== "queued")).toBe(true);
});
test("the hold clock releases work after its maximum, and a started key begins again", () => {
  const c = holdClock(100);
  expect(c.held("a", 0)).toBe(true);
  expect(c.held("a", 99)).toBe(true);
  expect(c.held("a", 100)).toBe(false);
  c.release("a");
  expect(c.held("a", 500)).toBe(true);
});
test("the hold clock starts again after reset, and keep forgets work no longer waiting", () => {
  const c = holdClock(100);
  c.held("a", 0);
  c.held("b", 0);
  c.reset();
  expect(c.held("a", 150)).toBe(true);
  c.held("b", 150);
  c.keep(new Set(["a"]));
  expect(c.held("a", 250)).toBe(false);
  expect(c.held("b", 260)).toBe(true);
});
// The timers are set far off: each test drives `takeIfDue` itself, with an injected clock.
function snapshotRuns(cpus: number, crew: () => number, extra: { holdMaxMs?: number; now?: () => number } = {}) {
  snaps = startSnapshots(app, dirStore(join(dir, "store")), {
    version: "t",
    cpus: () => cpus,
    activeCrewRuns: crew,
    snapshotCatchUpMs: 3_600_000,
    snapshotCheckMs: 3_600_000,
    holdRetryMs: 3_600_000,
    catchUpMs: 3_600_000,
    mirrorEveryMs: 3_600_000,
    nightly: false,
    ...extra,
  });
  return snaps;
}
test("1 CPU with a crew run holds a due snapshot, and takes it once the run is gone", async () => {
  let crew = 1;
  const s = snapshotRuns(1, () => crew);
  expect(await s.takeIfDue()).toBe(false);
  expect((await s.list()).length).toBe(0);
  crew = 0;
  expect(await s.takeIfDue()).toBe(true);
  expect(s.last().manifest).toBeDefined();
});
test("1 CPU: a snapshot held past the maximum wait is taken anyway", async () => {
  let now = 1_000;
  const s = snapshotRuns(1, () => 1, { holdMaxMs: 5_000, now: () => now });
  expect(await s.takeIfDue()).toBe(false);
  now += 4_999;
  expect(await s.takeIfDue()).toBe(false);
  now += 1;
  expect(await s.takeIfDue()).toBe(true);
  expect(s.last().manifest).toBeDefined();
});
test("2 CPUs take a due snapshot at once, without asking about crew runs", async () => {
  let asked = 0;
  const s = snapshotRuns(2, () => (asked++, 1));
  expect(await s.takeIfDue()).toBe(true);
  expect(s.last().manifest).toBeDefined();
  expect(asked).toBe(0);
});
@@ -1,16 +1,16 @@
import { expect, test } from "bun:test";
import { crewHoldsCore, forgeCpus } from "../src/core/one_core.ts";
test("1 CPU with a crew run is held", () => expect(crewHoldsCore(1, 1)).toBe(true));
test("1 CPU with no run is not held", () => expect(crewHoldsCore(1, 0)).toBe(false));
test("1 CPU with a crew run is held", () => expect(crewHoldsCore(1, () => 1)).toBe(true));
test("1 CPU with no run is not held", () => expect(crewHoldsCore(1, () => 0)).toBe(false));
test("2 and 4 CPUs with a run are not held", () => {
  expect(crewHoldsCore(2, 3)).toBe(false);
  expect(crewHoldsCore(4, 1)).toBe(false);
  expect(crewHoldsCore(2, () => 3)).toBe(false);
  expect(crewHoldsCore(4, () => 1)).toBe(false);
});
test("0 or negative CPUs read as 1 cpu: held with a run", () => {
  expect(crewHoldsCore(0, 1)).toBe(true);
  expect(crewHoldsCore(-2, 1)).toBe(true);
  expect(crewHoldsCore(0, 0)).toBe(false);
  expect(crewHoldsCore(0, () => 1)).toBe(true);
  expect(crewHoldsCore(-2, () => 1)).toBe(true);
  expect(crewHoldsCore(0, () => 0)).toBe(false);
});
test("forgeCpus is a whole number of at least 1", () => {
  const n = forgeCpus();
		@@ -18,6 +18,7 @@ import { chmod, lstat, readdir, rm } from "node:fs/promises";
18	18	import { join } from "node:path";
19	19	import type { App } from "../app.ts";
20	20	import { releaseOnGreen } from "../actions/holds.ts";
	21	import { activeCrewRuns, crewHoldsCore, forgeCpus, holdClock } from "../core/one_core.ts";
21	22	import { announceRun } from "../actions/runs.ts";
22	23	import { announceShard } from "./shard_event.ts";
23	24	import type { ForgeEvent } from "../core/events.ts";
		@@ -56,7 +57,6 @@ import { readCombined } from "../git/combined.ts";
56	57	import { approvedOnce } from "./approved_first.ts";
57	58	import { checkOnApproval } from "./approval.ts";
58	59	import { checkKey, provenBy, recordProven } from "./proven.ts";
59		import { crewHoldsCore, forgeCpus } from "../core/one_core.ts";
60	60	import { beginGate, endGate, gateHolding } from "./gate_record.ts";
61	61	import { giveHistory } from "./history.ts";
62	62	import { readLearned } from "./learned.ts";
		@@ -145,10 +145,6 @@ export interface ChecksOptions {
145	145	readonly concurrency?: number \| undefined;
146	146	/** The forge's public URL, told to the check as `FORGE_URL`. */
147	147	readonly forgeUrl: string;
148		/** The forge's CPU count; a 1-CPU forge holds pull checks for a crew run. @default forgeCpus */
149		readonly cpus?: (() => number) \| undefined;
150		/** Crew runs active now. @default the active agent runs */
151		readonly activeCrewRuns?: (() => number) \| undefined;
152	148	/** The parent's gate, for a hosted child: its checks take their turn in the parent's lane. */
153	149	readonly gate?: Gate \| undefined;
154	150	/** Whose checks these are, written first in each run's log (a hosted child's name). */
		@@ -189,6 +185,15 @@ export interface ChecksOptions {
189	185	* to `VERIFY_EVERY_MS`. The forge sets it to `VERIFY_EVERY_MS`.
190	186	*/
191	187	readonly verifyEveryMs?: number \| undefined;
	188	/** Test seam: how often the queue is looked at. @default TICK_MS */
	189	readonly tickMs?: number \| undefined;
	190	/** The forge's CPU count; a 1-CPU forge holds pull checks for a crew run. @default forgeCpus */
	191	readonly cpus?: (() => number) \| undefined;
	192	/** Crew runs working now. @default the running agent runs */
	193	readonly activeCrewRuns?: (() => number) \| undefined;
	194	/** Test seams: how long a held pull check waits at most, and the clock it is judged by. */
	195	readonly holdMaxMs?: number \| undefined;
	196	readonly holdNow?: (() => number) \| undefined;
192	197	}
193	198
194	199	/** A gate's answer from the lane: `interrupted` is no verdict (#3155). */
		@@ -633,10 +638,20 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
633	638	return next;
634	639	}
635	640
	641	const pullHold = holdClock(options.holdMaxMs);
	642
636	643	function schedule(): void {
637	644	if (stopped) return;
638	645	const limit = capacity();
		junipercrew 2d ago outdated This return skips startVerify() above and the rebase and main-check picks below. Hold only pull checks, and only for a bounded time.
639	646	if (options.gate === undefined) gate.resize(limit);
	647	// On one CPU a pull's check waits while a crew run works, for at most
	648	// `holdMaxMs`; the verify lane, a rebase's check and a main check are the
	649	// pipeline and never wait (#11599).
	650	const crewHolds = crewHoldsCore((options.cpus ?? forgeCpus)(), options.activeCrewRuns ?? (() => activeCrewRuns(app.db)));
	651	// The clock counts one continuous hold: it starts again once no crew run
	652	// holds, and forgets checks that are no longer queued.
	653	if (!crewHolds) pullHold.reset();
	654	else pullHold.keep(new Set(listActive(app.db, "check").filter((run) => run.state === "queued").map((run) => run.id)));
640	655	startVerify();
641	656	// Every merge-queue gate in flight holds a slot, a line's first too (#6376).
642	657	while (slotted() < limit - gatesInFlight()) {
		@@ -652,25 +667,22 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
652	667	(repoId) => redMain(app.db, repoId),
653	668	options.redHoldMs ?? 0,
654	669	);
655		// One core: a crew run has it. A pull's check keeps its place and starts
656		// once the run has ended; main's check (the merge gate) is never held.
657		const crewHolds = crewHoldsCore(
658		(options.cpus ?? forgeCpus)(),
659		(options.activeCrewRuns ?? (() => listActive(app.db, "agent").length))(),
660		);
661		const runnable = crewHolds ? queued.filter((run) => run.issueId === null) : queued;
662	670	// The limit is asked at the point of starting, one pick at a time:
663	671	// a started check is in flight before the next is asked about, so a
664	672	// burst of one account's checks cannot pass together. A rebase's
665	673	// check goes before any other (`finishing_first.ts`, #6140).
666		const { rebases, rest: others } = rebasesFirst(app.db, runnable);
	674	const { rebases, rest: others } = rebasesFirst(app.db, queued);
667	675	// While a gate is about to start, the last free slot is its: a pull's
668	676	// check waits for it, a rebase's and a main check do not.
669	677	const held = holdsLastSlot(limit - gatesInFlight() - slotted(), gateQueued());
670		const next = startable(rebases) ?? startable(held ? others.filter((run) => run.issueId === null) : others);
	678	const waiting = crewHolds
	679	? others.filter((run) => run.issueId === null \|\| !pullHold.held(run.id, (options.holdNow ?? Date.now)()))
	680	: others;
	681	const next = startable(rebases) ?? startable(held ? waiting.filter((run) => run.issueId === null) : waiting);
671	682	if (!next) return;
672	683	const repo = getRepoById(app.db, next.repoId);
673	684	if (!repo) return;
	685	pullHold.release(next.id);
674	686	start(next, repo);
675	687	}
676	688	}
		@@ -1716,7 +1728,7 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
1716	1728
1717	1729	const timer = setInterval(() => {
1718	1730	if (!stopped) during("lane", schedule);
1719		}, TICK_MS);
	1731	}, options.tickMs ?? TICK_MS);
1720	1732	// Main is verified after landing on a schedule, when the forge says how often.
1721	1733	const sweeper =
1722	1734	options.verifyEveryMs === undefined
		@@ -1,8 +1,9 @@
	1	import type { Database } from "bun:sqlite";
1	2	import { availableParallelism } from "node:os";
2	3
3		/** True when a 1-CPU forge should hold background work for a crew run. */
4		export function crewHoldsCore(cpus: number, activeCrewRuns: number): boolean {
5		return cpus <= 1 && activeCrewRuns > 0;
	4	/** True when a 1-CPU forge should hold background work for a crew run; more CPUs never count the runs. */
	5	export function crewHoldsCore(cpus: number, activeCrewRuns: () => number): boolean {
	6	return cpus <= 1 && activeCrewRuns() > 0;
6	7	}
7	8
8	9	/** The forge's own CPU count, at least 1. */
		@@ -10,3 +11,39 @@ export function forgeCpus(): number {
10	11	const n = availableParallelism();
11	12	return Number.isFinite(n) && n >= 1 ? Math.floor(n) : 1;
12	13	}
	14
	15	/** Crew runs working now: agent runs in the `running` state. */
		junipercrew 2d ago Looks counts active agent runs with listActive(app.db, 'agent'); this counts only state='running'. Keep one definition and use it in looks as well.
	16	export function activeCrewRuns(db: Database): number {
	17	const row = db.query<{ n: number }, []>("SELECT COUNT(*) AS n FROM run WHERE kind = 'agent' AND state = 'running'").get();
	18	return row?.n ?? 0;
	19	}
	20
	21	/** The longest a check or a snapshot waits for a crew run on one CPU before it starts anyway. */
	22	export const MAX_HOLD_MS = 10 * 60_000;
	23
	24	/**
	25	* Whether work keyed `key` is still held at `now`. The first time it is held
	26	* starts its clock; once `maxMs` has passed it is held no longer, so crew runs
	27	* back to back cannot starve it. `release` forgets the key when the work starts,
	28	* `keep` forgets every key not named (work cancelled or superseded), and
	29	* `reset` forgets them all once no crew run holds, so a later run holds afresh.
	30	*/
	31	export function holdClock(maxMs = MAX_HOLD_MS) {
	32	const since = new Map<string, number>();
	33	return {
	34	held(key: string, now: number): boolean {
	35	const first = since.get(key) ?? now;
	36	since.set(key, first);
	37	return now - first < maxMs;
	38	},
	39	release(key: string): void {
	40	since.delete(key);
	41	},
	42	keep(keys: ReadonlySet<string>): void {
	43	for (const key of since.keys()) if (!keys.has(key)) since.delete(key);
	44	},
	45	reset(): void {
	46	since.clear();
	47	},
	48	};
	49	}
		@@ -408,7 +408,7 @@ export function startLooks(app: App, options: LooksOptions): LooksHandle {
408	408	const cpus = options.cpus ?? forgeCpus;
409	409	const active =
410	410	options.activeCrewRuns ?? (() => listActive(app.db, "agent").length);
411		if (crewHoldsCore(cpus(), active())) {
	411	if (crewHoldsCore(cpus(), active)) {
412	412	queue.unshift(next);
413	413	return;
414	414	}
		@@ -11,6 +11,7 @@
11	11	*/
12	12
13	13	import type { App } from "../app.ts";
	14	import { activeCrewRuns, crewHoldsCore, forgeCpus, holdClock } from "../core/one_core.ts";
14	15	import { createLog } from "../kernel/log.ts";
15	16	import { firstSeen, reportVerdict, seedFirstSeen } from "./verdict.ts";
16	17	import {
		@@ -36,8 +37,6 @@ import { isOffBox } from "./off_box.ts";
36	37	import { noteOffBox, recordSnapshotOutcome } from "./outcome.ts";
37	38	import { s3FromEnv, type Store } from "./store.ts";
38	39	import { during } from "../perf/jobs.ts";
39		import { crewHoldsCore, forgeCpus } from "../core/one_core.ts";
40		import { listActive } from "../core/runs.ts";
41	40
42	41	const log = createLog("snapshots");
43	42
		@@ -109,6 +108,8 @@ const CATCH_UP_MS = 5 * 60_000;
109	108	const SNAPSHOT_CATCH_UP_MS = 2 * 60_000;
110	109	/** A snapshot still owed is looked for again this often; one boot's look is not the last (#10167). */
111	110	const SNAPSHOT_CHECK_MS = 30 * 60_000;
	111	/** A snapshot held for a crew run on one CPU looks again this often. */
	112	const HOLD_RETRY_MS = 30_000;
112	113
113	114	export function untilDrill(now = Date.now()): number {
114	115	const next = new Date(now);
		@@ -153,9 +154,11 @@ export function startSnapshots(
153	154	readonly bucketFirstMs?: number \| undefined;
154	155	/** Test seam: the clock `takeIfDue` judges by. */
155	156	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	/** Test seams: the forge's CPU count, crew runs working now, how soon a held snapshot looks again, and how long it waits at most. */
157	158	readonly cpus?: (() => number) \| undefined;
158	159	readonly activeCrewRuns?: (() => number) \| undefined;
	160	readonly holdRetryMs?: number \| undefined;
	161	readonly holdMaxMs?: number \| undefined;
159	162	/** Test seam: how soon the first nightly timer fires. */
160	163	readonly nightlyInMs?: number \| undefined;
161	164	} = {
		@@ -178,6 +181,17 @@ export function startSnapshots(
178	181	return resolvedMirror.store;
179	182	};
180	183	let stopped = false;
	184	// On one CPU a due snapshot waits for the crew run to end, for at most
	185	// `holdMaxMs`, then is taken anyway so back-to-back runs cannot starve it (#11599).
	186	const wait = holdClock(options.holdMaxMs ?? MAX_DEFER_MS);
	187	const crewHolds = (): boolean => {
	188	if (!crewHoldsCore((options.cpus ?? forgeCpus)(), options.activeCrewRuns ?? (() => activeCrewRuns(app.db)))) {
	189	wait.reset();
	190	return false;
	191	}
	192	return wait.held("snapshot", (options.now ?? Date.now)());
	193	};
	194	const holdRetryMs = options.holdRetryMs ?? options.snapshotCheckMs ?? HOLD_RETRY_MS;
181	195	let inFlight: Promise<Manifest> \| undefined;
182	196	let active = 0;
183	197	let running: Progress \| undefined;
		@@ -189,6 +203,7 @@ export function startSnapshots(
189	203	let offBox: boolean \| undefined;
190	204	// After the one before it, however that ended: chained on a failure as it
191	205	// was, every later run failed at once with the old error until a restart.
	206	wait.release("snapshot");
192	207	const run = (inFlight ?? Promise.resolve()).catch(() => {}).then(async () => {
193	208	// Resolved per run, so a destination saved on the Backups page is the
194	209	// very next snapshot's.
		@@ -271,26 +286,6 @@ export function startSnapshots(
271	286	}
272	287	};
273	288
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
294	289	// A restart takes the night's snapshot if the store does not hold it: a
295	290	// release at 08:21 killed the one in flight and nothing took it again (#8742).
296	291	// The manifest is written last, so a snapshot cut short is not one.
		@@ -305,9 +300,7 @@ export function startSnapshots(
305	300	log.warn("snapshot catch-up: store unreadable, not taking one now", { error: String(error) });
306	301	return false;
307	302	}
308		if (stopped \|\| active > 0 \|\| !snapshotDue(newest?.finishedAt, (options.now ?? Date.now)())) return false;
309		if (deferred()) return false;
310		deferredSince = undefined;
	303	if (stopped \|\| active > 0 \|\| !snapshotDue(newest?.finishedAt, (options.now ?? Date.now)()) \|\| crewHolds()) return false;
311	304	const cut = await cutShort(store, newest?.name);
312	305	log.info("snapshot due or lost; taking one", { newest: newest?.name ?? "none", resuming: cut?.toISOString() ?? "no" });
313	306	await take(cut).catch(() => {});
		@@ -411,18 +404,12 @@ export function startSnapshots(
411	404	if (timer !== undefined) clearTimeout(timer);
412	405	timer = undefined;
413	406	if (options.nightly === false \|\| stopped) return;
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		);
	407	timer = setTimeout(() => {
	408	void during("snapshot", async () => {
	409	if (active === 0 && crewHolds()) return arm(holdRetryMs);
	410	await take();
	411	}).catch(() => {});
	412	}, ms);
426	413	}
427	414	arm(options.nightlyInMs ?? untilNightly());
428	415
		@@ -496,7 +483,7 @@ export function startSnapshots(
496	483	.catch(() => {})
497	484	.finally(() => {
498	485	bootLookAt = undefined;
499		scheduleCatchUp(options.snapshotCheckMs ?? SNAPSHOT_CHECK_MS);
	486	scheduleCatchUp(crewHolds() ? holdRetryMs : (options.snapshotCheckMs ?? SNAPSHOT_CHECK_MS));
500	487	});
501	488	}, ms);
502	489	};
		@@ -0,0 +1,190 @@
	1	/** #11599: on one CPU a check or a snapshot waits for the crew run; two CPUs start them at once. */
	2	import { afterEach, beforeEach, expect, test } from "bun:test";
	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 { startChecks, type ChecksHandle } from "../src/checks/service.ts";
	8	import { createRepo } from "../src/actions/repos.ts";
	9	import { createIssue } from "../src/core/issues.ts";
	10	import { createRun, getRun } from "../src/core/runs.ts";
	11	import { createUser } from "../src/core/users.ts";
	12	import { startSnapshots, type SnapshotsHandle } from "../src/snapshot/service.ts";
	13	import { dirStore } from "../src/snapshot/store.ts";
	14	import { holdClock } from "../src/core/one_core.ts";
	15	import { quietLogs } from "./helpers/globals.ts";
	16
	17	quietLogs();
	18
	19	let dir: string;
	20	let app: App;
	21	let checks: ChecksHandle \| undefined;
	22	let snaps: SnapshotsHandle \| undefined;
	23	let seeded: { userId: string; repoId: string } \| undefined;
	24
	25	beforeEach(async () => {
	26	dir = await mkdtemp(join(tmpdir(), "forge-one-core-"));
	27	app = createApp({ dataDir: dir, port: 0 });
	28	});
	29	afterEach(async () => {
	30	await checks?.stop({ drain: false });
	31	await snaps?.stop();
	32	checks = snaps = seeded = undefined;
	33	app.close();
	34	await rm(dir, { recursive: true, force: true });
	35	});
	36
	37	const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
	38	async function until(ok: () => boolean, ms = 3000): Promise<boolean> {
	39	for (const end = Date.now() + ms; Date.now() < end; await sleep(20)) if (ok()) return true;
	40	return ok();
	41	}
	42
	43	async function queuedCheck(opts: { pull?: boolean; env?: string; reason?: string; rebase?: boolean } = {}): Promise<string> {
	44	if (seeded === undefined) {
	45	const made = await createUser(app.db, { origin: "seed", trusted: true, username: "alice", email: "[email protected]", password: "a-good-passphrase", adminIfFirst: true });
	46	if (!made.ok) throw new Error(made.error);
	47	const repo = await createRepo(app, made.value, { name: "lib", visibility: "public" });
	48	if (!repo.ok) throw new Error(String(repo.error));
	49	seeded = { userId: made.value.id, repoId: repo.value.id };
	50	}
	51	const issue = createIssue(app.db, { repoId: seeded.repoId, authorId: seeded.userId, title: "change" });
	52	if (!issue.ok) throw new Error(issue.error);
	53	if (opts.rebase) createRun(app.db, { repoId: seeded.repoId, kind: "agent", issueId: issue.value.id, reason: "conflict", commitOid: "0".repeat(40) });
	54	return createRun(app.db, { repoId: seeded.repoId, kind: "check", env: opts.env ?? "", commitOid: "0".repeat(40), reason: opts.reason ?? "test", ...(opts.pull === false ? {} : { issueId: issue.value.id }) }).id;
	55	}
	56
	57	test("1 CPU with a crew run holds a queued check, and it starts once the run is gone", async () => {
	58	const id = await queuedCheck();
	59	let crew = 1;
	60	checks = startChecks(app, { forgeUrl: "http://127.0.0.1:9", concurrency: 1, cpus: () => 1, activeCrewRuns: () => crew, tickMs: 25 });
	61	await sleep(200);
	62	expect(getRun(app.db, id)!.state).toBe("queued");
	63	crew = 0;
	64	expect(await until(() => getRun(app.db, id)!.state !== "queued")).toBe(true);
	65	});
	66
	67	test("2 CPUs start a queued check at once, whatever the crew does", async () => {
	68	const id = await queuedCheck();
	69	checks = startChecks(app, { forgeUrl: "http://127.0.0.1:9", concurrency: 1, cpus: () => 2, activeCrewRuns: () => 1, tickMs: 25 });
	70	expect(await until(() => getRun(app.db, id)!.state !== "queued")).toBe(true);
	71	});
	72
	73	test("1 CPU with a crew run does not hold a main check, a rebase check or a verify run", async () => {
	74	const main = await queuedCheck({ pull: false });
	75	checks = startChecks(app, { forgeUrl: "http://127.0.0.1:9", concurrency: 1, cpus: () => 1, activeCrewRuns: () => 1, tickMs: 25 });
	76	expect(await until(() => getRun(app.db, main)!.state !== "queued")).toBe(true);
	77	});
	78
	79	test("1 CPU: a verify-lane run starts with a crew run active", async () => {
	80	const verify = await queuedCheck({ pull: false, env: "verify" });
	81	checks = startChecks(app, { forgeUrl: "http://127.0.0.1:9", concurrency: 1, cpus: () => 1, activeCrewRuns: () => 1, tickMs: 25 });
	82	expect(await until(() => getRun(app.db, verify)!.state !== "queued")).toBe(true);
	83	});
	84
	85	test("1 CPU: a rebase's check starts with a crew run active, a pull's check beside it waits", async () => {
	86	const rebase = await queuedCheck({ rebase: true });
	87	const pull = await queuedCheck();
	88	checks = startChecks(app, { forgeUrl: "http://127.0.0.1:9", concurrency: 2, cpus: () => 1, activeCrewRuns: () => 1, tickMs: 25 });
	89	expect(await until(() => getRun(app.db, rebase)!.state !== "queued")).toBe(true);
	90	expect(getRun(app.db, pull)!.state).toBe("queued");
	91	});
	92
	93	test("1 CPU: a pull's check held past the maximum wait starts anyway", async () => {
	94	const id = await queuedCheck();
	95	let now = 1_000;
	96	checks = startChecks(app, { forgeUrl: "http://127.0.0.1:9", concurrency: 1, cpus: () => 1, activeCrewRuns: () => 1, tickMs: 25, holdMaxMs: 5_000, holdNow: () => now });
	97	await sleep(150);
	98	expect(getRun(app.db, id)!.state).toBe("queued");
	99	now += 5_000;
	100	expect(await until(() => getRun(app.db, id)!.state !== "queued")).toBe(true);
	101	});
	102
	103	test("1 CPU: a check held, then idle, is held afresh when the next crew run comes", async () => {
	104	const id = await queuedCheck();
	105	let now = 1_000;
	106	let crew = 1;
	107	checks = startChecks(app, { forgeUrl: "http://127.0.0.1:9", concurrency: 1, cpus: () => 1, activeCrewRuns: () => crew, tickMs: 25, holdMaxMs: 5_000, holdNow: () => now });
	108	await sleep(100);
	109	expect(getRun(app.db, id)!.state).toBe("queued");
	110	// Kept queued by something else (a not-before) while no crew run works.
	111	app.db.run("UPDATE run SET not_before = ? WHERE id = ?", [new Date(Date.now() + 60_000).toISOString(), id]);
	112	now += 4_000;
	113	crew = 0;
	114	await sleep(100);
	115	// The crew returns past the first clock's bound: a fresh hold, not a release.
	116	crew = 1;
	117	now += 2_000;
	118	app.db.run("UPDATE run SET not_before = NULL WHERE id = ?", [id]);
	119	await sleep(150);
	120	expect(getRun(app.db, id)!.state).toBe("queued");
	121	now += 5_000;
	122	expect(await until(() => getRun(app.db, id)!.state !== "queued")).toBe(true);
	123	});
	124
	125	test("the hold clock releases work after its maximum, and a started key begins again", () => {
	126	const c = holdClock(100);
	127	expect(c.held("a", 0)).toBe(true);
	128	expect(c.held("a", 99)).toBe(true);
	129	expect(c.held("a", 100)).toBe(false);
	130	c.release("a");
	131	expect(c.held("a", 500)).toBe(true);
	132	});
	133
	134	test("the hold clock starts again after reset, and keep forgets work no longer waiting", () => {
	135	const c = holdClock(100);
	136	c.held("a", 0);
	137	c.held("b", 0);
	138	c.reset();
	139	expect(c.held("a", 150)).toBe(true);
	140	c.held("b", 150);
	141	c.keep(new Set(["a"]));
	142	expect(c.held("a", 250)).toBe(false);
	143	expect(c.held("b", 260)).toBe(true);
	144	});
	145
	146	// The timers are set far off: each test drives `takeIfDue` itself, with an injected clock.
	147	function snapshotRuns(cpus: number, crew: () => number, extra: { holdMaxMs?: number; now?: () => number } = {}) {
	148	snaps = startSnapshots(app, dirStore(join(dir, "store")), {
	149	version: "t",
	150	cpus: () => cpus,
	151	activeCrewRuns: crew,
	152	snapshotCatchUpMs: 3_600_000,
	153	snapshotCheckMs: 3_600_000,
	154	holdRetryMs: 3_600_000,
	155	catchUpMs: 3_600_000,
	156	mirrorEveryMs: 3_600_000,
	157	nightly: false,
	158	...extra,
	159	});
	160	return snaps;
	161	}
	162
	163	test("1 CPU with a crew run holds a due snapshot, and takes it once the run is gone", async () => {
	164	let crew = 1;
	165	const s = snapshotRuns(1, () => crew);
	166	expect(await s.takeIfDue()).toBe(false);
	167	expect((await s.list()).length).toBe(0);
	168	crew = 0;
	169	expect(await s.takeIfDue()).toBe(true);
	170	expect(s.last().manifest).toBeDefined();
	171	});
	172
	173	test("1 CPU: a snapshot held past the maximum wait is taken anyway", async () => {
	174	let now = 1_000;
	175	const s = snapshotRuns(1, () => 1, { holdMaxMs: 5_000, now: () => now });
	176	expect(await s.takeIfDue()).toBe(false);
	177	now += 4_999;
	178	expect(await s.takeIfDue()).toBe(false);
	179	now += 1;
	180	expect(await s.takeIfDue()).toBe(true);
	181	expect(s.last().manifest).toBeDefined();
	182	});
	183
	184	test("2 CPUs take a due snapshot at once, without asking about crew runs", async () => {
	185	let asked = 0;
	186	const s = snapshotRuns(2, () => (asked++, 1));
	187	expect(await s.takeIfDue()).toBe(true);
	188	expect(s.last().manifest).toBeDefined();
	189	expect(asked).toBe(0);
	190	});
		@@ -1,16 +1,16 @@
1	1	import { expect, test } from "bun:test";
2	2	import { crewHoldsCore, forgeCpus } from "../src/core/one_core.ts";
3	3
4		test("1 CPU with a crew run is held", () => expect(crewHoldsCore(1, 1)).toBe(true));
5		test("1 CPU with no run is not held", () => expect(crewHoldsCore(1, 0)).toBe(false));
	4	test("1 CPU with a crew run is held", () => expect(crewHoldsCore(1, () => 1)).toBe(true));
	5	test("1 CPU with no run is not held", () => expect(crewHoldsCore(1, () => 0)).toBe(false));
6	6	test("2 and 4 CPUs with a run are not held", () => {
7		expect(crewHoldsCore(2, 3)).toBe(false);
8		expect(crewHoldsCore(4, 1)).toBe(false);
	7	expect(crewHoldsCore(2, () => 3)).toBe(false);
	8	expect(crewHoldsCore(4, () => 1)).toBe(false);
9	9	});
10	10	test("0 or negative CPUs read as 1 cpu: held with a run", () => {
11		expect(crewHoldsCore(0, 1)).toBe(true);
12		expect(crewHoldsCore(-2, 1)).toBe(true);
13		expect(crewHoldsCore(0, 0)).toBe(false);
	11	expect(crewHoldsCore(0, () => 1)).toBe(true);
	12	expect(crewHoldsCore(-2, () => 1)).toBe(true);
	13	expect(crewHoldsCore(0, () => 0)).toBe(false);
14	14	});
15	15	test("forgeCpus is a whole number of at least 1", () => {
16	16	const n = forgeCpus();