30e7b7c · 8065e11 · fd29865
@@ -32,10 +32,11 @@
spec:
      annotations:
        plat.sh/user-env-rev: {{ .Values.userEnvRev | default "init" | quote }}
    spec:
      # The crew drains for three minutes, then runs are cut and resume on
      # the next pod. Checks and merge-queue gates under way drain for
      # FORGE_CHECKS_DRAIN_S (660 s, a full gate's ten minutes and more), so
      # a roll does not cut a gate off; a minute more than that here (#6722).
      # The crew, checks and merge-queue gates under way get ten seconds
      # (FORGE_CREW_DRAIN_S, FORGE_CHECKS_DRAIN_S), then are handed back
      # queued to resume on the next pod: the old pod stops writing at once,
      # so the new one's queue work waits seconds, not minutes (#7799). The
      # long grace stays for SSH sessions that are still connected.
      terminationGracePeriodSeconds: 720
      {{- with .Values.imagePullSecrets }}
      imagePullSecrets:
@@ -1500,7 +1500,8 @@
a link to the milestone (`docs/PLANNING.md`); its progress never does.
| `FORGE_CREW_CONCURRENCY`  | 1 per 3 GiB above 2 GiB, 2–16   | agent runs at once (live: Settings → Capacity)    |
| `FORGE_CHECKS_CONCURRENCY` | `1`                            | checks at once (live: Settings → Capacity)        |
| `FORGE_CHECKS_PLACEMENT`  | `process`                       | `job` runs a check's commands as a cluster Job (see Placement above) |
| `FORGE_CHECKS_DRAIN_S`       | `660`                          | seconds a roll lets running checks finish first   |
| `FORGE_CHECKS_DRAIN_S`       | `10`                           | seconds a roll lets running checks finish first   |
| `FORGE_CREW_DRAIN_S`         | `10`                           | seconds a roll lets crew runs finish before they are queued again |
| `FORGE_CHECK_SHARDS`      | the machine's cores             | test shards per check (live: Settings → Capacity) |
| `FORGE_CELLS_EGRESS`      | `enforce`                       | `report` lets undeclared egress through on 80/443 (live: Settings → Capacity) |
| `FORGE_CELL_CGROUP`       | `/sys/fs/cgroup` at a namespace root | the delegated cgroup v2 whose children cap each cell's memory; `off` watches only |
@@ -155,12 +155,27 @@
this, frozen.
## Gates in flight across a roll (#6722)
A roll does not turn a merge-queue gate red. On SIGTERM the old forge
starts no new gate and lets the gates under way finish for
`FORGE_CHECKS_DRAIN_S` (660 s: a full gate takes about ten minutes). The
chart's `terminationGracePeriodSeconds` is 720 for that reason, a minute
past the drain; lower one and lower the other with it, or the kubelet kills
the pod while its gates still run. The old and new pods share the node for
up to those 12 minutes, which section 1's rule already sizes for.
starts no new gate and gives the gates under way `FORGE_CHECKS_DRAIN_S`
(10 s) to finish, the crew's runs `FORGE_CREW_DRAIN_S` (10 s); what is left
is handed back queued. The old forge stops writing within seconds, so the
new forge's `waitForPeer` (`src/runs/peer.ts`) returns at once and the
release's write pause stays under 30 s (#7799). Before that the drain was
660 s and the pause as long. Both budgets are read in `src/drain.ts`; junk
falls back to 10 s.
A gate cut off at the drain is `interrupted`, never red, and the new forge
gates the same candidate again from the persisted queue
(`test/merge-queue-restart.test.ts`: "the run gone, the same candidate is
gated again" and the #6722 red case). A gate that outlives the drain on a
roll is therefore redone, not lost; set `FORGE_CHECKS_DRAIN_S` higher to
trade a longer write pause for fewer redone gates.
If a crew run ignores the abort at the end of its 10 s, the old forge's
leases stay renewed and the new forge logs "another forge still works this
database" and keeps serving pages but holds its queue work. It starts when
the run's process finally ends, or after `PEER_WAIT_MAX_MS` (720 s, the
pod's grace, after which the platform has stopped the old pod). Pages answer
throughout; only queue work is late.
What the drain does not finish is no verdict, never red:
@@ -0,0 +1,29 @@
/**
 * How long a SIGTERM'd forge lets its work finish before it hands it back
 * queued (#7799). The new forge holds its own queue work until the old one
 * stops writing, so these budgets are the write pause of a release: seconds,
 * not the ten minutes a full gate takes. What is cut off is queued again and
 * resumes on the new forge.
 */
/** Seconds, for checks, merge-queue gates and crew runs alike, unless the environment says. */
export const DRAIN_DEFAULT_S = 10;
export interface DrainBudgets {
  readonly checksMs: number;
  readonly crewMs: number;
}
/** A positive number of seconds, else the default: junk never lengthens a roll. */
function seconds(value: string | undefined): number {
  const parsed = Number(value);
  return Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) || DRAIN_DEFAULT_S : DRAIN_DEFAULT_S;
}
/** `FORGE_CHECKS_DRAIN_S` and `FORGE_CREW_DRAIN_S`, in milliseconds. */
export function drainBudgets(env: Readonly<Record<string, string | undefined>>): DrainBudgets {
  return {
    checksMs: seconds(env["FORGE_CHECKS_DRAIN_S"]) * 1000,
    crewMs: seconds(env["FORGE_CREW_DRAIN_S"]) * 1000,
  };
}
@@ -25,6 +25,7 @@
import { startUpstreamWatch } from "./spawn/upstream.ts";
import { pruneSessions } from "./core/sessions.ts";
import { runDelivery } from "./core/webhooks.ts";
import { isHookName, runHookClient } from "./git/hooks.ts";
import { drainBudgets } from "./drain.ts";
import { drainStreams } from "./http/sse.ts";
import { serve } from "./http/server.ts";
import { checkDue } from "./actions/upstream_follow.ts";
@@ -405,9 +406,11 @@
const checks =
        redHoldMs: RED_HOLD_MS,
        landSettleMs: LAND_SETTLE_MS,
        // A roll lets running checks and merge-queue gates finish this long
        // before cutting them off: a full gate takes about ten minutes, and
        // the pod's grace (charts/app, 720 s) is longer still (#6722).
        drainMs: positive(process.env.FORGE_CHECKS_DRAIN_S, 660) * 1000,
        // before cutting them off, and no longer: the new forge holds its
        // own queue work until this one stops writing, so the write pause
        // of a release is this budget (#7799). A gate cut off is queued
        // again and resumes on the new forge.
        drainMs: drainBudgets(process.env).checksMs,
        // Main is verified after landing at most this often (#7786).
        verifyEveryMs: VERIFY_EVERY_MS,
      });
@@ -730,7 +733,7 @@
function drain(signal: NodeJS.Signals): Promise<unknown> {
    linger,
    prompts.stop(),
    notifier.stop(),
    crew?.stop({ drain: true }),
    crew?.stop({ drain: true, drainMs: drainBudgets(process.env).crewMs }),
    // Routers still send this process requests for the linger's length;
    // its apps answer them, and stop only after (#4496).
    (linger ?? Promise.resolve()).then(() => apps?.stop()).finally(() => presence?.stop()),
@@ -21,7 +21,7 @@
import { HEARTBEAT_MS, heldRuns } from "./lease.ts";
/** Two renewals missed: the lease is not being kept. */
export const PEER_FRESH_MS = 2 * HEARTBEAT_MS;
/** The pod's grace (charts/app, 720 s): an old pod is gone by then. */
/** The pod's grace (charts/app, 720 s): an old pod is gone by then. A drain is 10 s (`src/drain.ts`); this is the ceiling for a run that ignores it. */
export const PEER_WAIT_MAX_MS = 720_000;
const POLL_MS = 5_000;
@@ -238,6 +238,65 @@
sleep 120
    expect(again[0]!.summary).toContain("carries on");
  }, 40_000);
  test("a draining crew hands a long run back queued within its budget (#7799)", async () => {
    await crew?.stop();
    crew = undefined;
    // The run the test before put back on the queue would take the one slot.
    for (const left of listQueued(app.db, "agent")) cancelRun(app.db, left.id, "cleared for the next test");
    const created = await createRepo(app, alice, { name: "thicket", visibility: "private" });
    if (!created.ok) throw new Error(String(created.error));
    const repo = created.value;
    const seeded = await commitToBranch(app, alice, repo, {
      message: "Start",
      files: [{ path: "README.md", content: "# thicket\n" }],
    });
    if (!seeded.ok) throw new Error(JSON.stringify(seeded.error));
    expect(addToCrew(app, alice, repo, "bot").ok).toBe(true);
    const issued = openIssue(app, alice, repo, { title: "Clear", body: "" });
    if (!issued.ok) throw new Error(String(issued.error));
    const issue = issued.value;
    const long = join(dataDir, "long-claude");
    writeFileSync(
      long,
      `#!/bin/sh
if [ "$1" = "--version" ]; then echo "2.1.0 (fake)"; exit 0; fi
cat > /dev/null &
echo '{"type":"system","subtype":"init","session_id":"long-1","model":"fake"}'
sleep 120
`,
    );
    chmodSync(long, 0o755);
    const queuedRun = createRun(app.db, {
      repoId: repo.id,
      kind: "agent",
      issueId: issue.id,
      actorId: bot.id,
      reason: "assigned",
    });
    crew = await startCrew(app, {
      forgeUrl: servers.url,
      localUrl: servers.url,
      binary: long,
      concurrency: 1,
      budgetUsd: 1,
      runTimeoutMs: 60_000,
    });
    const deadline = Date.now() + 20_000;
    while (getRun(app.db, queuedRun.id)?.state !== "running" && Date.now() < deadline) await Bun.sleep(100);
    expect(getRun(app.db, queuedRun.id)?.state).toBe("running");
    const started = Date.now();
    await crew.stop({ drain: true, drainMs: 1_000 });
    crew = undefined;
    expect(Date.now() - started).toBeLessThan(15_000);
    const ended = getRun(app.db, queuedRun.id)!;
    expect(ended.state === "cancelled" || ended.state === "failed").toBe(true);
    const again = listQueued(app.db, "agent").filter((r) => r.issueId === issue.id && r.state === "queued");
    expect(again).toHaveLength(1);
    expect(again[0]!.summary).toContain("carries on");
  }, 40_000);
  test("a draining crew lets a run in flight finish and land; nothing is queued again", async () => {
    await crew?.stop();
    crew = undefined;
@@ -0,0 +1,19 @@
import { expect, test } from "bun:test";
import { drainBudgets } from "../src/drain.ts";
test("a roll's drain defaults to ten seconds for checks and for the crew", () => {
  expect(drainBudgets({})).toEqual({ checksMs: 10_000, crewMs: 10_000 });
});
test("each variable overrides its own budget", () => {
  expect(drainBudgets({ FORGE_CHECKS_DRAIN_S: "660" })).toEqual({ checksMs: 660_000, crewMs: 10_000 });
  expect(drainBudgets({ FORGE_CREW_DRAIN_S: "30" })).toEqual({ checksMs: 10_000, crewMs: 30_000 });
});
test("junk falls back to the default", () => {
  for (const junk of ["", "abc", "-5", "0", "0.4", "NaN", "Infinity"])
    expect(drainBudgets({ FORGE_CHECKS_DRAIN_S: junk, FORGE_CREW_DRAIN_S: junk })).toEqual({
      checksMs: 10_000,
      crewMs: 10_000,
    });
});
@@ -17,6 +17,7 @@
import { getPull } from "../src/core/pulls.ts";
import type { Repo } from "../src/core/repos.ts";
import { createUser, type User } from "../src/core/users.ts";
import { repoDir } from "../src/git/repos.ts";
import { peerRuns } from "../src/runs/peer.ts";
describe("a gate under way when the forge rolls", () => {
  let dataDir: string;
@@ -93,3 +94,61 @@
describe("a gate under way when the forge rolls", () => {
    expect(checks.adopt?.(pull.headOid)).toBeUndefined();
  }, 60_000);
});
describe("the drain is seconds, not minutes (#7799)", () => {
  test("a gate still running at the drain's end is handed back queued within it, and no run is left for a peer to wait on", async () => {
    const dataDir = await mkdtemp(join(tmpdir(), "forge-gate-short-"));
    const cells = process.env["FORGE_CELLS"];
    process.env["FORGE_CELLS"] = "off";
    const app = createApp({ dataDir, port: 0 });
    const checks = startChecks(app, { forgeUrl: "http://127.0.0.1:1", concurrency: 1, drainMs: 1_000 });
    try {
      const a = await createUser(app.db, {
        origin: "seed",
        trusted: true,
        username: "alice",
        email: "[email protected]",
        password: "a-good-passphrase",
        adminIfFirst: true,
      });
      if (!a.ok) throw new Error(a.error);
      const made = await createRepo(app, a.value, { name: "short", visibility: "public" });
      if (!made.ok) throw new Error(String(made.error));
      const repo = made.value;
      const seeded = await commitToBranch(app, a.value, repo, {
        message: "Start",
        files: [{ path: "forge.json", content: JSON.stringify({ check: { run: "true", fast: "true", full: "sleep 30" } }) }],
      });
      if (!seeded.ok) throw new Error(JSON.stringify(seeded.error));
      const branched = await commitToBranch(app, a.value, repo, {
        branch: "feature",
        message: "Touch b",
        files: [{ path: "b.txt", content: "b\n" }],
      });
      if (!branched.ok) throw new Error(JSON.stringify(branched.error));
      const opened = await openPull(app, a.value, repo, { title: "Feature", baseBranch: "main", headBranch: "feature" });
      if (!opened.ok) throw new Error(String(opened.error));
      const pull = getPull(app.db, repo.id, opened.value.number)!;
      const gating = gateCandidate(app, repo, pull, repoDir(app.reposRoot, repo.ownerName, repo.name), pull.headOid);
      let adopted = checks.adopt?.(pull.headOid);
      for (let i = 0; i < 400 && adopted === undefined; i++) {
        await Bun.sleep(5);
        adopted = checks.adopt?.(pull.headOid);
      }
      expect(adopted).toBeDefined();
      const started = Date.now();
      await checks.stop({ drain: true });
      expect(Date.now() - started).toBeLessThan(15_000);
      const verdict = await gating;
      expect(verdict.lands).toBe(false);
      expect("interrupted" in verdict).toBe(true);
      expect((await adopted!).interrupted).toBe(true);
      expect(peerRuns(app.db)).toBe(0);
    } finally {
      app.close();
      if (cells === undefined) delete process.env["FORGE_CELLS"];
      else process.env["FORGE_CELLS"] = cells;
      await rm(dataDir, { recursive: true, force: true });
    }
  }, 60_000);
});
@@ -0,0 +1,120 @@
/**
 * Two forges on one data directory across a roll (#7799). The old one has a
 * check running and a merge-queue gate in flight; on SIGTERM it gives them
 * its drain budget, then hands them back queued and lets go. The new one,
 * booted beside it, holds its own queue work only until then: `waitForPeer`
 * returns well inside 30 s, and the check is queued again for the new forge.
 */
import { expect, test } from "bun:test";
import { mkdtemp, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join, resolve } from "node:path";
import { createApp } from "../src/app.ts";
const SRC = resolve(import.meta.dir, "../src");
/** The old forge in miniature: the real checks service and gate lane on the shared data, nothing else. */
const oldForge = `
const { createApp } = require(${JSON.stringify(SRC + "/app.ts")});
const { openPull } = require(${JSON.stringify(SRC + "/actions/pulls.ts")});
const { commitToBranch, createRepo } = require(${JSON.stringify(SRC + "/actions/repos.ts")});
const { gateCandidate } = require(${JSON.stringify(SRC + "/checks/landing.ts")});
const { startChecks } = require(${JSON.stringify(SRC + "/checks/service.ts")});
const { getPull } = require(${JSON.stringify(SRC + "/core/pulls.ts")});
const { createUser } = require(${JSON.stringify(SRC + "/core/users.ts")});
const { repoDir } = require(${JSON.stringify(SRC + "/git/repos.ts")});
const { drainBudgets } = require(${JSON.stringify(SRC + "/drain.ts")});
(async () => {
  const app = createApp({ dataDir: process.argv[2], port: 0 });
  const checks = startChecks(app, { forgeUrl: "http://127.0.0.1:1", concurrency: 2, drainMs: drainBudgets(process.env).checksMs });
  const a = await createUser(app.db, { origin: "seed", trusted: true, username: "alice", email: "[email protected]", password: "a-good-passphrase", adminIfFirst: true });
  const repo = (await createRepo(app, a.value, { name: "roll", visibility: "public" })).value;
  await commitToBranch(app, a.value, repo, { message: "Start", files: [{ path: "forge.json", content: JSON.stringify({ check: { run: "sleep 120", fast: "true", full: "sleep 120" } }) }] });
  await commitToBranch(app, a.value, repo, { branch: "feature", message: "Touch", files: [{ path: "b.txt", content: "b\\n" }] });
  const opened = await openPull(app, a.value, repo, { title: "Feature", baseBranch: "main", headBranch: "feature" });
  const pull = getPull(app.db, repo.id, opened.value.number);
  void gateCandidate(app, repo, pull, repoDir(app.reposRoot, repo.ownerName, repo.name), pull.headOid);
  process.on("SIGTERM", async () => {
    await checks.stop({ drain: true });
    app.close();
    console.log("old forge let go");
    process.exit(0);
  });
  const t = setInterval(() => {
    const running = app.db.query("SELECT COUNT(*) AS n FROM run WHERE state = 'running'").get().n;
    const gating = checks.adopt(pull.headOid) !== undefined;
    if (running > 0 && gating) { clearInterval(t); console.log("old forge up"); }
  }, 100);
})();
`;
test("the new forge starts its queue work within 30 s of SIGTERM to the old, with a gate and a check still in flight", async () => {
  const dataDir = await mkdtemp(join(tmpdir(), "forge-rollpause-"));
  const script = join(dataDir, "..", `${dataDir.split("/").pop()}-old.js`);
  await writeFile(script, oldForge);
  const old = Bun.spawn(["bun", script, dataDir], {
    env: { ...(process.env as Record<string, string>), FORGE_CELLS: "off", FORGE_CHECKS_DRAIN_S: "3" },
    stdout: "pipe",
    stderr: "inherit",
  });
  const oldOut = old.stdout.getReader();
  let new_ = undefined as ReturnType<typeof Bun.spawn> | undefined;
  try {
    let said = "";
    const upDeadline = Date.now() + 60_000;
    while (!said.includes("old forge up") && Date.now() < upDeadline) {
      const { value, done } = await oldOut.read();
      if (done) break;
      said += new TextDecoder().decode(value);
    }
    expect(said).toContain("old forge up");
    const web = 20000 + Math.floor(Math.random() * 20000);
    const env: Record<string, string> = {
      ...(process.env as Record<string, string>),
      FORGE_DATA: dataDir,
      PORT: String(web),
      FORGE_SSH_PORT: "0",
      FORGE_APPS: "0",
      FORGE_CELLS: "off",
    };
    delete env.FORGE_URL;
    const forge = Bun.spawn(["bun", "run", "src/main.ts"], { env, stdout: "pipe", stderr: "pipe" });
    new_ = forge;
    let out = "";
    const collect = async (stream: ReadableStream<Uint8Array>) => {
      for await (const chunk of stream) out += new TextDecoder().decode(chunk);
    };
    void collect(forge.stdout as ReadableStream<Uint8Array>);
    void collect(forge.stderr as ReadableStream<Uint8Array>);
    for (let i = 0; i < 300 && !out.includes("another forge still works this database"); i++) await Bun.sleep(100);
    expect(out).toContain("another forge still works this database");
    const signalled = Date.now();
    old.kill("SIGTERM");
    for (let i = 0; i < 300 && !out.includes("the other forge let go of this database"); i++) await Bun.sleep(100);
    const pause = Date.now() - signalled;
    if (pause >= 30_000) console.log(out);
    expect(out).toContain("the other forge let go of this database");
    expect(pause).toBeLessThan(30_000);
    await old.exited;
    // The check the old forge cut off is queued again, for the new one to take.
    forge.kill();
    await forge.exited;
    const app = createApp({ dataDir, port: 0 });
    try {
      const states = app.db.query<{ state: string }, []>("SELECT state FROM run WHERE kind = 'check'").all().map((r) => r.state);
      expect(states.length).toBeGreaterThan(0);
      expect(states).not.toContain("failed");
    } finally {
      app.close();
    }
  } finally {
    old.kill("SIGKILL");
    new_?.kill("SIGKILL");
    await rm(dataDir, { recursive: true, force: true });
    await rm(script, { force: true });
  }
}, 120_000);
		@@ -32,10 +32,11 @@ spec:
32	32	annotations:
33	33	plat.sh/user-env-rev: {{ .Values.userEnvRev \| default "init" \| quote }}
34	34	spec:
35		# The crew drains for three minutes, then runs are cut and resume on
36		# the next pod. Checks and merge-queue gates under way drain for
37		# FORGE_CHECKS_DRAIN_S (660 s, a full gate's ten minutes and more), so
38		# a roll does not cut a gate off; a minute more than that here (#6722).
	35	# The crew, checks and merge-queue gates under way get ten seconds
	36	# (FORGE_CREW_DRAIN_S, FORGE_CHECKS_DRAIN_S), then are handed back
	37	# queued to resume on the next pod: the old pod stops writing at once,
	38	# so the new one's queue work waits seconds, not minutes (#7799). The
	39	# long grace stays for SSH sessions that are still connected.
39	40	terminationGracePeriodSeconds: 720
40	41	{{- with .Values.imagePullSecrets }}
41	42	imagePullSecrets:
		@@ -1500,7 +1500,8 @@ a link to the milestone (`docs/PLANNING.md`); its progress never does.
1500	1500	\| `FORGE_CREW_CONCURRENCY` \| 1 per 3 GiB above 2 GiB, 2–16 \| agent runs at once (live: Settings → Capacity) \|
1501	1501	\| `FORGE_CHECKS_CONCURRENCY` \| `1` \| checks at once (live: Settings → Capacity) \|
1502	1502	\| `FORGE_CHECKS_PLACEMENT` \| `process` \| `job` runs a check's commands as a cluster Job (see Placement above) \|
1503		\| `FORGE_CHECKS_DRAIN_S` \| `660` \| seconds a roll lets running checks finish first \|
	1503	\| `FORGE_CHECKS_DRAIN_S` \| `10` \| seconds a roll lets running checks finish first \|
	1504	\| `FORGE_CREW_DRAIN_S` \| `10` \| seconds a roll lets crew runs finish before they are queued again \|
1504	1505	\| `FORGE_CHECK_SHARDS` \| the machine's cores \| test shards per check (live: Settings → Capacity) \|
1505	1506	\| `FORGE_CELLS_EGRESS` \| `enforce` \| `report` lets undeclared egress through on 80/443 (live: Settings → Capacity) \|
1506	1507	\| `FORGE_CELL_CGROUP` \| `/sys/fs/cgroup` at a namespace root \| the delegated cgroup v2 whose children cap each cell's memory; `off` watches only \|
		@@ -155,12 +155,27 @@ this, frozen.
155	155	## Gates in flight across a roll (#6722)
156	156
157	157	A roll does not turn a merge-queue gate red. On SIGTERM the old forge
158		starts no new gate and lets the gates under way finish for
159		`FORGE_CHECKS_DRAIN_S` (660 s: a full gate takes about ten minutes). The
160		chart's `terminationGracePeriodSeconds` is 720 for that reason, a minute
161		past the drain; lower one and lower the other with it, or the kubelet kills
162		the pod while its gates still run. The old and new pods share the node for
163		up to those 12 minutes, which section 1's rule already sizes for.
	158	starts no new gate and gives the gates under way `FORGE_CHECKS_DRAIN_S`
	159	(10 s) to finish, the crew's runs `FORGE_CREW_DRAIN_S` (10 s); what is left
	160	is handed back queued. The old forge stops writing within seconds, so the
	161	new forge's `waitForPeer` (`src/runs/peer.ts`) returns at once and the
	162	release's write pause stays under 30 s (#7799). Before that the drain was
	163	660 s and the pause as long. Both budgets are read in `src/drain.ts`; junk
	164	falls back to 10 s.
	165
	166	A gate cut off at the drain is `interrupted`, never red, and the new forge
	167	gates the same candidate again from the persisted queue
	168	(`test/merge-queue-restart.test.ts`: "the run gone, the same candidate is
	169	gated again" and the #6722 red case). A gate that outlives the drain on a
	170	roll is therefore redone, not lost; set `FORGE_CHECKS_DRAIN_S` higher to
	171	trade a longer write pause for fewer redone gates.
	172
	173	If a crew run ignores the abort at the end of its 10 s, the old forge's
	174	leases stay renewed and the new forge logs "another forge still works this
	175	database" and keeps serving pages but holds its queue work. It starts when
	176	the run's process finally ends, or after `PEER_WAIT_MAX_MS` (720 s, the
	177	pod's grace, after which the platform has stopped the old pod). Pages answer
	178	throughout; only queue work is late.
164	179
165	180	What the drain does not finish is no verdict, never red:
166	181
		@@ -0,0 +1,29 @@
	1	/**
	2	* How long a SIGTERM'd forge lets its work finish before it hands it back
	3	* queued (#7799). The new forge holds its own queue work until the old one
	4	* stops writing, so these budgets are the write pause of a release: seconds,
	5	* not the ten minutes a full gate takes. What is cut off is queued again and
	6	* resumes on the new forge.
	7	*/
	8
	9	/** Seconds, for checks, merge-queue gates and crew runs alike, unless the environment says. */
	10	export const DRAIN_DEFAULT_S = 10;
	11
	12	export interface DrainBudgets {
	13	readonly checksMs: number;
	14	readonly crewMs: number;
	15	}
	16
	17	/** A positive number of seconds, else the default: junk never lengthens a roll. */
	18	function seconds(value: string \| undefined): number {
	19	const parsed = Number(value);
	20	return Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) \|\| DRAIN_DEFAULT_S : DRAIN_DEFAULT_S;
	21	}
	22
	23	/** `FORGE_CHECKS_DRAIN_S` and `FORGE_CREW_DRAIN_S`, in milliseconds. */
	24	export function drainBudgets(env: Readonly<Record<string, string \| undefined>>): DrainBudgets {
	25	return {
	26	checksMs: seconds(env["FORGE_CHECKS_DRAIN_S"]) * 1000,
	27	crewMs: seconds(env["FORGE_CREW_DRAIN_S"]) * 1000,
	28	};
	29	}
		@@ -25,6 +25,7 @@ import { startUpstreamWatch } from "./spawn/upstream.ts";
25	25	import { pruneSessions } from "./core/sessions.ts";
26	26	import { runDelivery } from "./core/webhooks.ts";
27	27	import { isHookName, runHookClient } from "./git/hooks.ts";
	28	import { drainBudgets } from "./drain.ts";
28	29	import { drainStreams } from "./http/sse.ts";
29	30	import { serve } from "./http/server.ts";
30	31	import { checkDue } from "./actions/upstream_follow.ts";
		@@ -405,9 +406,11 @@ const checks =
405	406	redHoldMs: RED_HOLD_MS,
406	407	landSettleMs: LAND_SETTLE_MS,
407	408	// A roll lets running checks and merge-queue gates finish this long
408		// before cutting them off: a full gate takes about ten minutes, and
409		// the pod's grace (charts/app, 720 s) is longer still (#6722).
410		drainMs: positive(process.env.FORGE_CHECKS_DRAIN_S, 660) * 1000,
	409	// before cutting them off, and no longer: the new forge holds its
	410	// own queue work until this one stops writing, so the write pause
	411	// of a release is this budget (#7799). A gate cut off is queued
	412	// again and resumes on the new forge.
	413	drainMs: drainBudgets(process.env).checksMs,
411	414	// Main is verified after landing at most this often (#7786).
412	415	verifyEveryMs: VERIFY_EVERY_MS,
413	416	});
		@@ -730,7 +733,7 @@ function drain(signal: NodeJS.Signals): Promise<unknown> {
730	733	linger,
731	734	prompts.stop(),
732	735	notifier.stop(),
733		crew?.stop({ drain: true }),
	736	crew?.stop({ drain: true, drainMs: drainBudgets(process.env).crewMs }),
734	737	// Routers still send this process requests for the linger's length;
735	738	// its apps answer them, and stop only after (#4496).
736	739	(linger ?? Promise.resolve()).then(() => apps?.stop()).finally(() => presence?.stop()),
		@@ -21,7 +21,7 @@ import { HEARTBEAT_MS, heldRuns } from "./lease.ts";
21	21
22	22	/** Two renewals missed: the lease is not being kept. */
23	23	export const PEER_FRESH_MS = 2 * HEARTBEAT_MS;
24		/** The pod's grace (charts/app, 720 s): an old pod is gone by then. */
	24	/** The pod's grace (charts/app, 720 s): an old pod is gone by then. A drain is 10 s (`src/drain.ts`); this is the ceiling for a run that ignores it. */
25	25	export const PEER_WAIT_MAX_MS = 720_000;
26	26	const POLL_MS = 5_000;
27	27
		@@ -238,6 +238,65 @@ sleep 120
238	238	expect(again[0]!.summary).toContain("carries on");
239	239	}, 40_000);
240	240
	241	test("a draining crew hands a long run back queued within its budget (#7799)", async () => {
	242	await crew?.stop();
	243	crew = undefined;
	244	// The run the test before put back on the queue would take the one slot.
	245	for (const left of listQueued(app.db, "agent")) cancelRun(app.db, left.id, "cleared for the next test");
	246	const created = await createRepo(app, alice, { name: "thicket", visibility: "private" });
	247	if (!created.ok) throw new Error(String(created.error));
	248	const repo = created.value;
	249	const seeded = await commitToBranch(app, alice, repo, {
	250	message: "Start",
	251	files: [{ path: "README.md", content: "# thicket\n" }],
	252	});
	253	if (!seeded.ok) throw new Error(JSON.stringify(seeded.error));
	254	expect(addToCrew(app, alice, repo, "bot").ok).toBe(true);
	255	const issued = openIssue(app, alice, repo, { title: "Clear", body: "" });
	256	if (!issued.ok) throw new Error(String(issued.error));
	257	const issue = issued.value;
	258	const long = join(dataDir, "long-claude");
	259	writeFileSync(
	260	long,
	261	`#!/bin/sh
	262	if [ "$1" = "--version" ]; then echo "2.1.0 (fake)"; exit 0; fi
	263	cat > /dev/null &
	264	echo '{"type":"system","subtype":"init","session_id":"long-1","model":"fake"}'
	265	sleep 120
	266	`,
	267	);
	268	chmodSync(long, 0o755);
	269	const queuedRun = createRun(app.db, {
	270	repoId: repo.id,
	271	kind: "agent",
	272	issueId: issue.id,
	273	actorId: bot.id,
	274	reason: "assigned",
	275	});
	276	crew = await startCrew(app, {
	277	forgeUrl: servers.url,
	278	localUrl: servers.url,
	279	binary: long,
	280	concurrency: 1,
	281	budgetUsd: 1,
	282	runTimeoutMs: 60_000,
	283	});
	284	const deadline = Date.now() + 20_000;
	285	while (getRun(app.db, queuedRun.id)?.state !== "running" && Date.now() < deadline) await Bun.sleep(100);
	286	expect(getRun(app.db, queuedRun.id)?.state).toBe("running");
	287
	288	const started = Date.now();
	289	await crew.stop({ drain: true, drainMs: 1_000 });
	290	crew = undefined;
	291	expect(Date.now() - started).toBeLessThan(15_000);
	292
	293	const ended = getRun(app.db, queuedRun.id)!;
	294	expect(ended.state === "cancelled" \|\| ended.state === "failed").toBe(true);
	295	const again = listQueued(app.db, "agent").filter((r) => r.issueId === issue.id && r.state === "queued");
	296	expect(again).toHaveLength(1);
	297	expect(again[0]!.summary).toContain("carries on");
	298	}, 40_000);
	299
241	300	test("a draining crew lets a run in flight finish and land; nothing is queued again", async () => {
242	301	await crew?.stop();
243	302	crew = undefined;
		@@ -0,0 +1,19 @@
	1	import { expect, test } from "bun:test";
	2	import { drainBudgets } from "../src/drain.ts";
	3
	4	test("a roll's drain defaults to ten seconds for checks and for the crew", () => {
	5	expect(drainBudgets({})).toEqual({ checksMs: 10_000, crewMs: 10_000 });
	6	});
	7
	8	test("each variable overrides its own budget", () => {
	9	expect(drainBudgets({ FORGE_CHECKS_DRAIN_S: "660" })).toEqual({ checksMs: 660_000, crewMs: 10_000 });
	10	expect(drainBudgets({ FORGE_CREW_DRAIN_S: "30" })).toEqual({ checksMs: 10_000, crewMs: 30_000 });
	11	});
	12
	13	test("junk falls back to the default", () => {
	14	for (const junk of ["", "abc", "-5", "0", "0.4", "NaN", "Infinity"])
	15	expect(drainBudgets({ FORGE_CHECKS_DRAIN_S: junk, FORGE_CREW_DRAIN_S: junk })).toEqual({
	16	checksMs: 10_000,
	17	crewMs: 10_000,
	18	});
	19	});
		@@ -17,6 +17,7 @@ import { getPull } from "../src/core/pulls.ts";
17	17	import type { Repo } from "../src/core/repos.ts";
18	18	import { createUser, type User } from "../src/core/users.ts";
19	19	import { repoDir } from "../src/git/repos.ts";
	20	import { peerRuns } from "../src/runs/peer.ts";
20	21
21	22	describe("a gate under way when the forge rolls", () => {
22	23	let dataDir: string;
		@@ -93,3 +94,61 @@ describe("a gate under way when the forge rolls", () => {
93	94	expect(checks.adopt?.(pull.headOid)).toBeUndefined();
94	95	}, 60_000);
95	96	});
	97
	98	describe("the drain is seconds, not minutes (#7799)", () => {
	99	test("a gate still running at the drain's end is handed back queued within it, and no run is left for a peer to wait on", async () => {
	100	const dataDir = await mkdtemp(join(tmpdir(), "forge-gate-short-"));
	101	const cells = process.env["FORGE_CELLS"];
	102	process.env["FORGE_CELLS"] = "off";
	103	const app = createApp({ dataDir, port: 0 });
	104	const checks = startChecks(app, { forgeUrl: "http://127.0.0.1:1", concurrency: 1, drainMs: 1_000 });
	105	try {
	106	const a = await createUser(app.db, {
	107	origin: "seed",
	108	trusted: true,
	109	username: "alice",
	110	email: "[email protected]",
	111	password: "a-good-passphrase",
	112	adminIfFirst: true,
	113	});
	114	if (!a.ok) throw new Error(a.error);
	115	const made = await createRepo(app, a.value, { name: "short", visibility: "public" });
	116	if (!made.ok) throw new Error(String(made.error));
	117	const repo = made.value;
	118	const seeded = await commitToBranch(app, a.value, repo, {
	119	message: "Start",
	120	files: [{ path: "forge.json", content: JSON.stringify({ check: { run: "true", fast: "true", full: "sleep 30" } }) }],
	121	});
	122	if (!seeded.ok) throw new Error(JSON.stringify(seeded.error));
	123	const branched = await commitToBranch(app, a.value, repo, {
	124	branch: "feature",
	125	message: "Touch b",
	126	files: [{ path: "b.txt", content: "b\n" }],
	127	});
	128	if (!branched.ok) throw new Error(JSON.stringify(branched.error));
	129	const opened = await openPull(app, a.value, repo, { title: "Feature", baseBranch: "main", headBranch: "feature" });
	130	if (!opened.ok) throw new Error(String(opened.error));
	131	const pull = getPull(app.db, repo.id, opened.value.number)!;
	132	const gating = gateCandidate(app, repo, pull, repoDir(app.reposRoot, repo.ownerName, repo.name), pull.headOid);
	133	let adopted = checks.adopt?.(pull.headOid);
	134	for (let i = 0; i < 400 && adopted === undefined; i++) {
	135	await Bun.sleep(5);
	136	adopted = checks.adopt?.(pull.headOid);
	137	}
	138	expect(adopted).toBeDefined();
	139	const started = Date.now();
	140	await checks.stop({ drain: true });
	141	expect(Date.now() - started).toBeLessThan(15_000);
	142	const verdict = await gating;
	143	expect(verdict.lands).toBe(false);
	144	expect("interrupted" in verdict).toBe(true);
	145	expect((await adopted!).interrupted).toBe(true);
	146	expect(peerRuns(app.db)).toBe(0);
	147	} finally {
	148	app.close();
	149	if (cells === undefined) delete process.env["FORGE_CELLS"];
	150	else process.env["FORGE_CELLS"] = cells;
	151	await rm(dataDir, { recursive: true, force: true });
	152	}
	153	}, 60_000);
	154	});
		@@ -0,0 +1,120 @@
	1	/**
	2	* Two forges on one data directory across a roll (#7799). The old one has a
	3	* check running and a merge-queue gate in flight; on SIGTERM it gives them
	4	* its drain budget, then hands them back queued and lets go. The new one,
	5	* booted beside it, holds its own queue work only until then: `waitForPeer`
	6	* returns well inside 30 s, and the check is queued again for the new forge.
	7	*/
	8
	9	import { expect, test } from "bun:test";
	10	import { mkdtemp, rm, writeFile } from "node:fs/promises";
	11	import { tmpdir } from "node:os";
	12	import { join, resolve } from "node:path";
	13	import { createApp } from "../src/app.ts";
	14
	15	const SRC = resolve(import.meta.dir, "../src");
	16
	17	/** The old forge in miniature: the real checks service and gate lane on the shared data, nothing else. */
	18	const oldForge = `
	19	const { createApp } = require(${JSON.stringify(SRC + "/app.ts")});
	20	const { openPull } = require(${JSON.stringify(SRC + "/actions/pulls.ts")});
	21	const { commitToBranch, createRepo } = require(${JSON.stringify(SRC + "/actions/repos.ts")});
	22	const { gateCandidate } = require(${JSON.stringify(SRC + "/checks/landing.ts")});
	23	const { startChecks } = require(${JSON.stringify(SRC + "/checks/service.ts")});
	24	const { getPull } = require(${JSON.stringify(SRC + "/core/pulls.ts")});
	25	const { createUser } = require(${JSON.stringify(SRC + "/core/users.ts")});
	26	const { repoDir } = require(${JSON.stringify(SRC + "/git/repos.ts")});
	27	const { drainBudgets } = require(${JSON.stringify(SRC + "/drain.ts")});
	28	(async () => {
	29	const app = createApp({ dataDir: process.argv[2], port: 0 });
	30	const checks = startChecks(app, { forgeUrl: "http://127.0.0.1:1", concurrency: 2, drainMs: drainBudgets(process.env).checksMs });
	31	const a = await createUser(app.db, { origin: "seed", trusted: true, username: "alice", email: "[email protected]", password: "a-good-passphrase", adminIfFirst: true });
	32	const repo = (await createRepo(app, a.value, { name: "roll", visibility: "public" })).value;
	33	await commitToBranch(app, a.value, repo, { message: "Start", files: [{ path: "forge.json", content: JSON.stringify({ check: { run: "sleep 120", fast: "true", full: "sleep 120" } }) }] });
	34	await commitToBranch(app, a.value, repo, { branch: "feature", message: "Touch", files: [{ path: "b.txt", content: "b\\n" }] });
	35	const opened = await openPull(app, a.value, repo, { title: "Feature", baseBranch: "main", headBranch: "feature" });
	36	const pull = getPull(app.db, repo.id, opened.value.number);
	37	void gateCandidate(app, repo, pull, repoDir(app.reposRoot, repo.ownerName, repo.name), pull.headOid);
	38	process.on("SIGTERM", async () => {
	39	await checks.stop({ drain: true });
	40	app.close();
	41	console.log("old forge let go");
	42	process.exit(0);
	43	});
	44	const t = setInterval(() => {
	45	const running = app.db.query("SELECT COUNT(*) AS n FROM run WHERE state = 'running'").get().n;
	46	const gating = checks.adopt(pull.headOid) !== undefined;
	47	if (running > 0 && gating) { clearInterval(t); console.log("old forge up"); }
	48	}, 100);
	49	})();
	50	`;
	51
	52	test("the new forge starts its queue work within 30 s of SIGTERM to the old, with a gate and a check still in flight", async () => {
	53	const dataDir = await mkdtemp(join(tmpdir(), "forge-rollpause-"));
	54	const script = join(dataDir, "..", `${dataDir.split("/").pop()}-old.js`);
	55	await writeFile(script, oldForge);
	56	const old = Bun.spawn(["bun", script, dataDir], {
	57	env: { ...(process.env as Record<string, string>), FORGE_CELLS: "off", FORGE_CHECKS_DRAIN_S: "3" },
	58	stdout: "pipe",
	59	stderr: "inherit",
	60	});
	61	const oldOut = old.stdout.getReader();
	62	let new_ = undefined as ReturnType<typeof Bun.spawn> \| undefined;
	63	try {
	64	let said = "";
	65	const upDeadline = Date.now() + 60_000;
	66	while (!said.includes("old forge up") && Date.now() < upDeadline) {
	67	const { value, done } = await oldOut.read();
	68	if (done) break;
	69	said += new TextDecoder().decode(value);
	70	}
	71	expect(said).toContain("old forge up");
	72
	73	const web = 20000 + Math.floor(Math.random() * 20000);
	74	const env: Record<string, string> = {
	75	...(process.env as Record<string, string>),
	76	FORGE_DATA: dataDir,
	77	PORT: String(web),
	78	FORGE_SSH_PORT: "0",
	79	FORGE_APPS: "0",
	80	FORGE_CELLS: "off",
	81	};
	82	delete env.FORGE_URL;
	83	const forge = Bun.spawn(["bun", "run", "src/main.ts"], { env, stdout: "pipe", stderr: "pipe" });
	84	new_ = forge;
	85	let out = "";
	86	const collect = async (stream: ReadableStream<Uint8Array>) => {
	87	for await (const chunk of stream) out += new TextDecoder().decode(chunk);
	88	};
	89	void collect(forge.stdout as ReadableStream<Uint8Array>);
	90	void collect(forge.stderr as ReadableStream<Uint8Array>);
	91	for (let i = 0; i < 300 && !out.includes("another forge still works this database"); i++) await Bun.sleep(100);
	92	expect(out).toContain("another forge still works this database");
	93
	94	const signalled = Date.now();
	95	old.kill("SIGTERM");
	96	for (let i = 0; i < 300 && !out.includes("the other forge let go of this database"); i++) await Bun.sleep(100);
	97	const pause = Date.now() - signalled;
	98	if (pause >= 30_000) console.log(out);
	99	expect(out).toContain("the other forge let go of this database");
	100	expect(pause).toBeLessThan(30_000);
	101	await old.exited;
	102
	103	// The check the old forge cut off is queued again, for the new one to take.
	104	forge.kill();
	105	await forge.exited;
	106	const app = createApp({ dataDir, port: 0 });
	107	try {
	108	const states = app.db.query<{ state: string }, []>("SELECT state FROM run WHERE kind = 'check'").all().map((r) => r.state);
	109	expect(states.length).toBeGreaterThan(0);
	110	expect(states).not.toContain("failed");
	111	} finally {
	112	app.close();
	113	}
	114	} finally {
	115	old.kill("SIGKILL");
	116	new_?.kill("SIGKILL");
	117	await rm(dataDir, { recursive: true, force: true });
	118	await rm(script, { force: true });
	119	}
	120	}, 120_000);