Public: anyone can read this; the crew can change it.
A cancelled run ends every process it started, proved by a test that leaves none alive #10783
Check · Review · Land · Live done — Landed and live in production.
mason/forge:mason/10776-a-cancelled-run-ends-every-process-it-st → main 2 commits fork
5 files
changed +198
−3
AGENTS.md +1 −1
| @@ -15,7 +15,7 @@ A new read of a table that grows (`event`, `issue`, `run`, `comment`) names the | ||
| 15 | 15 | |
| 16 | 16 | A new `/settings…` page route goes in `src/http/settings_routes.ts` (`settingsRoutes`), not `src/http/server.ts`. |
| 17 | 17 | |
| 18 | State shared between modules through process memory, not the database, is named here. `src/perf/views_report.ts`: `readVisits` holds each scope and window's visits for 15 seconds (dropped by `flushViews`), so a page and its probe share one read; `src/perf/views_store.ts`: the view buffer, flushed every 10 seconds. The traffic sources view's terminal door is `src/tui/screens/trafficview.ts` (palette entry for admins, per-app command for a repository's admins). `src/runs/held.ts`: each dispatcher pass (`src/crew/dispatch.ts`, `pump`) records which queued agent runs it held and why (valve, pace, limit), and vitals (`src/kernel/vitals.ts`) leaves those out of `queue:agent`. A record older than `HELD_FRESH_MS` is ignored, so a stopped dispatcher cannot hide its queue. `src/apps/egress_refused.ts`: refused-host counts per repository, flushed to disk every 5 seconds (`flushRefused` forces it). | |
| 18 | State shared between modules through process memory, not the database, is named here. `src/perf/views_report.ts`: `readVisits` holds each scope and window's visits for 15 seconds (dropped by `flushViews`), so a page and its probe share one read; `src/perf/views_store.ts`: the view buffer, flushed every 10 seconds. The traffic sources view's terminal door is `src/tui/screens/trafficview.ts` (palette entry for admins, per-app command for a repository's admins). `src/runs/held.ts`: each dispatcher pass (`src/crew/dispatch.ts`, `pump`) records which queued agent runs it held and why (valve, pace, limit), and vitals (`src/kernel/vitals.ts`) leaves those out of `queue:agent`. A record older than `HELD_FRESH_MS` is ignored, so a stopped dispatcher cannot hide its queue. `src/runs/stoppable.ts`: the checks in flight in this process by run id (`src/checks/service.ts` `start` tracks each), so a person's cancel (`cancelRepoRun`) aborts the check and its process group ends with it; agent runs keep `src/crew/running.ts`. `src/apps/egress_refused.ts`: refused-host counts per repository, flushed to disk every 5 seconds (`flushRefused` forces it). | |
| 19 | 19 | |
| 20 | 20 | `src/perf/loop_worst.ts`: `loopWorst`, the worst event-loop lag of the last 10 s in ten one-second slots, fed by one unref'd 50 ms timer that the first `serve` starts (`startLoopWorst`), and read by every response's `lag;dur` and by `/healthz` asked for JSON (`loopLagMs`). Tests listing a request's phases read `requestTiming` (`test/helpers/request-timing.ts`), which leaves `lag` out. |
| 21 | 21 |
src/actions/runs.ts +7 −2
| @@ -33,6 +33,7 @@ import { | ||
| 33 | 33 | } from "../core/runs.ts"; |
| 34 | 34 | import { getUsersByIds, type User } from "../core/users.ts"; |
| 35 | 35 | import { abortAgentRun } from "../crew/running.ts"; |
| 36 | import { stopRun } from "../runs/stoppable.ts"; | |
| 36 | 37 | import { queueAgentRun, taskOf } from "../crew/dispatch.ts"; |
| 37 | 38 | import { err, ok, type Result } from "../kernel/result.ts"; |
| 38 | 39 | import { readRunLog, readRunLogTail, runLogPath } from "../runs/log.ts"; |
| @@ -332,8 +333,9 @@ export function runLog( | ||
| 332 | 333 | |
| 333 | 334 | /** |
| 334 | 335 | * Cancel a run that has not finished. A queued run never starts; a running |
| 335 | * agent run is stopped; a running deploy is abandoned by its environment | |
| 336 | * when its step ends. | |
| 336 | * agent run or check worked here is stopped, every process it started with | |
| 337 | * it (#10776); a running deploy is abandoned by its environment when its | |
| 338 | * step ends. | |
| 337 | 339 | */ |
| 338 | 340 | export function cancelRepoRun( |
| 339 | 341 | app: App, |
| @@ -351,7 +353,10 @@ export function cancelRepoRun( | ||
| 351 | 353 | `cancelled by ${writer.value.username}`, |
| 352 | 354 | ); |
| 353 | 355 | if (!cancelled) return err("finished"); |
| 356 | // The row says cancelled first, so the run's own end leaves it so; the | |
| 357 | // abort then ends the process group it started. | |
| 354 | 358 | if (run.kind === "agent") abortAgentRun(runId); |
| 359 | else stopRun(runId); | |
| 355 | 360 | announceRun(app, cancelled, writer.value.id); |
| 356 | 361 | return ok(cancelled); |
| 357 | 362 | } |
src/checks/service.ts +4
| @@ -88,6 +88,7 @@ import { transact } from "../kernel/db.ts"; | ||
| 88 | 88 | import { createLog } from "../kernel/log.ts"; |
| 89 | 89 | import { holdLease, STALE_AFTER_MS } from "../runs/lease.ts"; |
| 90 | 90 | import { appendRunLog } from "../runs/log.ts"; |
| 91 | import { trackRun, untrackRun } from "../runs/stoppable.ts"; | |
| 91 | 92 | import { runningLimitSentence, runningRuns } from "../settings/quotas.ts"; |
| 92 | 93 | import { exportTree, readFileAt } from "../apps/checkout.ts"; |
| 93 | 94 | import { CHECK_TIMEOUT_MAX, parseCheck } from "../apps/manifest.ts"; |
| @@ -561,6 +562,8 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle { | ||
| 561 | 562 | const abort = new AbortController(); |
| 562 | 563 | const low = inVerifyLane(run.env); |
| 563 | 564 | if (low) verifying = run.id; |
| 565 | // A person's cancel (`cancelRepoRun`) reaches it through this. | |
| 566 | trackRun(run.id, abort); | |
| 564 | 567 | const promise = execute(run, repo, abort.signal, !low) |
| 565 | 568 | .catch((error) => |
| 566 | 569 | log.error("check failed to execute", { |
| @@ -569,6 +572,7 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle { | ||
| 569 | 572 | }), |
| 570 | 573 | ) |
| 571 | 574 | .finally(() => { |
| 575 | untrackRun(run.id, abort); | |
| 572 | 576 | inFlight.delete(run.id); |
| 573 | 577 | if (verifying === run.id) verifying = undefined; |
| 574 | 578 | // The slot this run held is free again: give it to whichever queued |
src/runs/stoppable.ts added +34
| @@ -0,0 +1,34 @@ | ||
| 1 | /** | |
| 2 | * The runs in flight in this process that a cancel can stop, by run id | |
| 3 | * (#10776). Whatever works a run registers its controller here for as long | |
| 4 | * as it works it; a cancel from any surface (`actions/runs.ts`) marks the | |
| 5 | * row and then aborts the controller, and the worker's abort ends the | |
| 6 | * process group it started (`apps/process.ts` `runStep`). A run worked by | |
| 7 | * another forge process is not here: its row says cancelled, and that | |
| 8 | * process's own end leaves it so. | |
| 9 | * | |
| 10 | * Agent runs keep their own registry (`crew/running.ts`), which the crew | |
| 11 | * dispatcher counts and stops by; this one is for the rest. | |
| 12 | */ | |
| 13 | ||
| 14 | const running = new Map<string, AbortController>(); | |
| 15 | ||
| 16 | export function trackRun(runId: string, controller: AbortController): void { | |
| 17 | running.set(runId, controller); | |
| 18 | } | |
| 19 | ||
| 20 | /** Forget a run, but only if `controller` is still the one tracked for it. */ | |
| 21 | export function untrackRun(runId: string, controller: AbortController): void { | |
| 22 | if (running.get(runId) === controller) running.delete(runId); | |
| 23 | } | |
| 24 | ||
| 25 | /** Stop a run worked in this process. A run not in flight here is left alone. */ | |
| 26 | export function stopRun(runId: string): boolean { | |
| 27 | const controller = running.get(runId); | |
| 28 | if (controller === undefined) return false; | |
| 29 | controller.abort(); | |
| 30 | return true; | |
| 31 | } | |
| 32 | ||
| 33 | /** Whether a run is still being worked in this process. */ | |
| 34 | export const runningRun = (runId: string): boolean => running.has(runId); |
test/run-cancel-tree.test.ts added +152
| @@ -0,0 +1,152 @@ | ||
| 1 | /** | |
| 2 | * A cancelled run ends every process it started (#10776). A person cancels | |
| 3 | * a running check through the same action every surface calls | |
| 4 | * (`cancelRepoRun`); the check's command forked a child, and that child | |
| 5 | * forked its own. Within 2 s none of the three is alive, and the run's | |
| 6 | * record says `cancelled`, not `failed`, after the check has wound down. | |
| 7 | * | |
| 8 | * Before this, the cancel marked the row and nothing else: an agent run was | |
| 9 | * aborted, but a check ran on to its end, its tree with it. | |
| 10 | */ | |
| 11 | import { afterAll, beforeAll, describe, expect, test } from "bun:test"; | |
| 12 | import { existsSync, readFileSync } from "node:fs"; | |
| 13 | import { mkdtemp, rm } from "node:fs/promises"; | |
| 14 | import { tmpdir } from "node:os"; | |
| 15 | import { join } from "node:path"; | |
| 16 | import { createApp, type App } from "../src/app.ts"; | |
| 17 | import { commitToBranch, createRepo } from "../src/actions/repos.ts"; | |
| 18 | import { cancelRepoRun } from "../src/actions/runs.ts"; | |
| 19 | import { startChecks, type ChecksHandle } from "../src/checks/service.ts"; | |
| 20 | import { getRun, listRuns } from "../src/core/runs.ts"; | |
| 21 | import { createUser } from "../src/core/users.ts"; | |
| 22 | import { runningRun } from "../src/runs/stoppable.ts"; | |
| 23 | import { quietLogs } from "./helpers/globals.ts"; | |
| 24 | ||
| 25 | quietLogs(); | |
| 26 | process.env["FORGE_CELLS"] = "off"; | |
| 27 | ||
| 28 | let dataDir: string; | |
| 29 | let pidDir: string; | |
| 30 | let app: App; | |
| 31 | let checks: ChecksHandle; | |
| 32 | const started: number[] = []; | |
| 33 | ||
| 34 | beforeAll(async () => { | |
| 35 | dataDir = await mkdtemp(join(tmpdir(), "forge-run-cancel-tree-")); | |
| 36 | pidDir = await mkdtemp(join(tmpdir(), "forge-run-cancel-pids-")); | |
| 37 | app = createApp({ dataDir, port: 0 }); | |
| 38 | checks = startChecks(app, { forgeUrl: "http://forge.test", concurrency: 1 }); | |
| 39 | }); | |
| 40 | ||
| 41 | afterAll(async () => { | |
| 42 | // A red run must not leave its sleepers behind on the machine. | |
| 43 | for (const pid of started) | |
| 44 | try { | |
| 45 | process.kill(pid, "SIGKILL"); | |
| 46 | } catch { | |
| 47 | // gone, as it should be | |
| 48 | } | |
| 49 | await checks.stop(); | |
| 50 | app.close(); | |
| 51 | await rm(dataDir, { recursive: true, force: true }); | |
| 52 | await rm(pidDir, { recursive: true, force: true }); | |
| 53 | }); | |
| 54 | ||
| 55 | /** Alive and not a zombie a parent has yet to reap. */ | |
| 56 | function alive(pid: number): boolean { | |
| 57 | let stat: string; | |
| 58 | try { | |
| 59 | stat = readFileSync(`/proc/${pid}/stat`, "utf8"); | |
| 60 | } catch { | |
| 61 | return false; | |
| 62 | } | |
| 63 | const state = stat.slice(stat.lastIndexOf(")") + 2)[0]; | |
| 64 | return state !== "Z" && state !== "X"; | |
| 65 | } | |
| 66 | ||
| 67 | /** Polls, never a fixed sleep: true once every pid is gone, false at the deadline. */ | |
| 68 | async function allGoneWithin(pids: readonly number[], ms: number): Promise<boolean> { | |
| 69 | const deadline = Date.now() + ms; | |
| 70 | while (Date.now() < deadline) { | |
| 71 | if (pids.every((pid) => !alive(pid))) return true; | |
| 72 | await Bun.sleep(20); | |
| 73 | } | |
| 74 | return pids.every((pid) => !alive(pid)); | |
| 75 | } | |
| 76 | ||
| 77 | async function pidIn(file: string, ms: number): Promise<number> { | |
| 78 | const deadline = Date.now() + ms; | |
| 79 | while (Date.now() < deadline) { | |
| 80 | if (existsSync(file)) { | |
| 81 | const pid = Number(readFileSync(file, "utf8").trim()); | |
| 82 | if (Number.isInteger(pid) && pid > 1) return pid; | |
| 83 | } | |
| 84 | await Bun.sleep(20); | |
| 85 | } | |
| 86 | throw new Error(`${file} never named a pid`); | |
| 87 | } | |
| 88 | ||
| 89 | describe.skipIf(process.platform !== "linux")("a cancelled check", () => { | |
| 90 | test("ends its command, the child it forked and that child's child within 2 s, and says cancelled", async () => { | |
| 91 | const made = await createUser(app.db, { | |
| 92 | origin: "seed", | |
| 93 | trusted: true, | |
| 94 | username: "alice", | |
| 95 | email: "[email protected]", | |
| 96 | password: "a-good-passphrase", | |
| 97 | adminIfFirst: true, | |
| 98 | }); | |
| 99 | if (!made.ok) throw new Error(made.error); | |
| 100 | const alice = made.value; | |
| 101 | const created = await createRepo(app, alice, { name: "tree", visibility: "public" }); | |
| 102 | if (!created.ok) throw new Error(String(created.error)); | |
| 103 | const repo = created.value; | |
| 104 | ||
| 105 | // The command writes its own pid, forks a shell that forks a sleeper, | |
| 106 | // and waits. The grandchild's pid is written last (renamed into place), | |
| 107 | // so all three exist once it is there. | |
| 108 | const shell = join(pidDir, "shell.pid"); | |
| 109 | const child = join(pidDir, "child.pid"); | |
| 110 | const grand = join(pidDir, "grand.pid"); | |
| 111 | const command = | |
| 112 | `echo $$ > ${shell}; ` + | |
| 113 | `sh -c 'echo $$ > ${child}; sleep 300 & echo $! > ${grand}.tmp; mv ${grand}.tmp ${grand}; wait' & ` + | |
| 114 | `wait`; | |
| 115 | const pushed = await commitToBranch(app, alice, repo, { | |
| 116 | message: "A check that forks and never ends", | |
| 117 | files: [ | |
| 118 | { | |
| 119 | path: "forge.json", | |
| 120 | content: `${JSON.stringify({ check: { run: command, install: "" } })}\n`, | |
| 121 | }, | |
| 122 | ], | |
| 123 | }); | |
| 124 | if (!pushed.ok) throw new Error(JSON.stringify(pushed.error)); | |
| 125 | ||
| 126 | const pids = [ | |
| 127 | await pidIn(grand, 30_000), | |
| 128 | await pidIn(child, 1_000), | |
| 129 | await pidIn(shell, 1_000), | |
| 130 | ]; | |
| 131 | started.push(...pids); | |
| 132 | expect(pids.every(alive)).toBe(true); | |
| 133 | ||
| 134 | const run = listRuns(app.db, repo.id, { kind: "check", limit: 5 }).find( | |
| 135 | (r) => r.commitOid === pushed.value.commitOid, | |
| 136 | ); | |
| 137 | if (run === undefined) throw new Error("no check run"); | |
| 138 | expect(run.state).toBe("running"); | |
| 139 | ||
| 140 | const cancelled = cancelRepoRun(app, alice, repo, run.id); | |
| 141 | expect(cancelled.ok).toBe(true); | |
| 142 | expect(await allGoneWithin(pids, 2_000)).toBe(true); | |
| 143 | ||
| 144 | // The check winds down and lets the run go; the row keeps what the cancel wrote. | |
| 145 | const deadline = Date.now() + 5_000; | |
| 146 | while (runningRun(run.id) && Date.now() < deadline) await Bun.sleep(20); | |
| 147 | expect(runningRun(run.id)).toBe(false); | |
| 148 | const after = getRun(app.db, run.id); | |
| 149 | expect(after?.state).toBe("cancelled"); | |
| 150 | expect(after?.summary).toBe("cancelled by alice"); | |
| 151 | }, 45_000); | |
| 152 | }); |