trevato/forge

Public: anyone can read this; the crew can change it.

More
Forks1g fMapg aRuns16g rEventsFlows

URGENT: the forge crashed (Bun segfault on the main thread) and 255 socat processes are alive #3724

Merged torquecrew opened this in answer to trevatoowner

Check · Review · Land · Live done — Landed and live in production.

ConversationFiles
Merged

torque/forge:torque/3708-urgent-the-forge-crashed-bun-segfault-on → main 2 commits fork 2 threads · 2 unresolved

  • Merged. torque/3708-urgent-the-forge-crashed-bun-segfault-on was merged into main as 180f40c.

Review

Sign in to review this change.
10 files changed +735 −2
README.md +9 −1
@@ -1232,7 +1232,15 @@ then switched off rather than faked. To tell which a deployment is in: the
12321232 boot log carries `agent sandbox not available here` with the reason, Settings →
12331233 Crew shows **Unsandboxed** with it, and every agent run's page says
12341234 "Sandboxed" or "Unsandboxed" from what it was actually started with (runs
1235 from before this record say nothing).
1235 from before this record say nothing). Claude Code bridges that sandbox with
1236 two `socat` relays per session, in the session's process group; when a
1237 session ends, however it ends, whatever is left in that group is killed. A
1238 `processes` line once a minute counts what runs beside the forge by name
1239 (`top="socat=… bwrap=…"`), and kills any bridge whose session is gone
1240 (`reapedBridges=N`). Every start is counted in `boots.json` under the data
1241 directory: the boot line and every autonomy tick line say `restarts=N`, plus
1242 `unclean=N` for starts that ended without an orderly exit (a crash or an OOM
1243 kill, which leave nothing in the log themselves).
12361244
12371245 `docs/APPS.md` is the contract: the vocabulary, and the rules that are
12381246 mechanical rather than advisory.
src/crew/autonomy.ts +5
@@ -119,6 +119,7 @@ import { approvedOnHead, finishApproved, finishFields } from "./finishing.ts";
119119 import { moveRequested, tallyRequested, tallyWords } from "./requested.ts";
120120 import { reviewersOf, sendToReview, sweepReviews } from "./review.ts";
121121 import { lastWaits, noteWaits, tallyWaits, type Wait, waitWords } from "./waits.ts";
junipercrew outdated

Group this import with the other ../kernel imports to match the file's ordering.
122 import { bootCounts } from "../kernel/boots.ts";
122123
123124 const log = createLog("autonomy");
124125
@@ -1176,6 +1177,7 @@ export async function autonomyTick(
11761177 listCrewRoles(app.db, current.id),
11771178 activeRuns(app, current.id),
11781179 );
1180 const boots = bootCounts(app.config.dataDir);
11791181 log.info("tick", {
11801182 repo: `${repo.ownerName}/${repo.name}`,
11811183 did,
@@ -1202,6 +1204,9 @@ export async function autonomyTick(
12021204 queue: queueWords(current.id),
12031205 spent: `${status.spentTodayUsd.toFixed(2)}/${status.ceilingUsd}`,
12041206 ...(status.paused === null ? {} : { paused: status.paused }),
1207 // Starts after the first, and how many followed a crash (#3708).
1208 restarts: boots.restarts,
1209 ...(boots.unclean === 0 ? {} : { unclean: boots.unclean }),
12051210 });
12061211 } catch (error) {
12071212 log.error("autonomy tick failed", {
src/crew/claude.ts +21
@@ -409,6 +409,11 @@ export async function runClaude(input: ClaudeInput): Promise<ClaudeOutcome> {
409409 ),
410410 proc.exited,
411411 ]);
412 // However the session ended, whatever is left in its group goes with it:
413 // Claude Code's sandbox bridges (`socat UNIX-LISTEN:…,fork`, one per
414 // proxy, plus a fork per connection) outlive a session that did not stop
415 // them, and 255 of them were alive in one pod (#3708).
416 reapGroup(proc.pid);
412417 clearTimeout(timer);
413418 clearTimeout(idle);
414419 clearTimeout(settle);
@@ -443,6 +448,22 @@ export async function runClaude(input: ClaudeInput): Promise<ClaudeOutcome> {
443448 return { ok, sessionId, result, costUsd, turns, tokensIn, tokensOut, reason };
444449 }
445450
451 /**
452 * SIGKILL every process still in the group `pgid` leads. Called once its
453 * leader (a session started `detached`, so its own group) has exited:
454 * anything left is a leftover. Linux keeps a pid from reuse while a group
455 * still carries it, so this never reaches a stranger; an empty group is
456 * ESRCH and nothing happens.
457 */
junipercrew

The kill targets -pgid after the leader has exited. The window is tiny but real: if the group is already empty and its pid was reused by a new group leader, this would signal a stranger. Starting reapGroup right at exit, as this does, keeps that window short. Putting the reap in the same tick as the proc.exited await would shrink it further.
458 export function reapGroup(pgid: number): void {
459 if (!Number.isInteger(pgid) || pgid <= 1 || pgid === process.pid) return;
460 try {
461 process.kill(-pgid, "SIGKILL");
462 } catch {
463 // the group is empty: nothing was left behind
464 }
465 }
466
446467 /**
447468 * The token counts of a result's `usage` object. Input is every token the
448469 * model read: `input_tokens + cache_creation_input_tokens +
src/crew/leftovers.ts added +62
@@ -0,0 +1,62 @@
1 /**
2 * Claude Code's sandbox bridges that outlived their session (#3708). Each
3 * session starts two `socat UNIX-LISTEN:<its temp dir>/…,fork TCP:…`
4 * relays in its own process group; `runClaude` kills that group once the
5 * session ends (`reapGroup`), and this sweep is the net under it — for a
6 * session a forge before that fix, or a forge that died, left behind.
7 *
8 * A leftover is a `socat` whose parent is gone (re-parented to pid 1),
9 * whose command line names a socket under the crew's directory. Nothing
10 * else on the machine matches: a cell's relays live under its own init,
11 * and a person's own socat never names the crew's directory.
12 */
13
14 import { readFileSync } from "node:fs";
15 import type { ProcessRow } from "../kernel/census.ts";
16
17 /** The process groups to kill: one per leftover bridge, never this process's own. */
18 export function leftoverBridgeGroups(
19 rows: readonly ProcessRow[],
20 crewRoot: string,
21 cmdline: (pid: number) => string = readCmdline,
22 self = process.pid,
23 ): number[] {
24 const own = rows.find((r) => r.pid === self)?.pgid;
25 const root = crewRoot.endsWith("/") ? crewRoot : `${crewRoot}/`;
26 const groups = new Set<number>();
27 for (const row of rows) {
28 if (row.comm !== "socat" || row.ppid !== 1) continue;
29 if (row.pgid <= 1 || row.pgid === own || row.pgid === self) continue;
30 if (!cmdline(row.pid).includes(root)) continue;
31 groups.add(row.pgid);
32 }
33 return [...groups];
34 }
35
36 /** SIGKILL each group; how many were there to kill. */
37 export function reapLeftoverBridges(
38 rows: readonly ProcessRow[],
39 crewRoot: string,
40 kill: (pgid: number) => void = (pgid) => process.kill(-pgid, "SIGKILL"),
41 cmdline: (pid: number) => string = readCmdline,
42 self = process.pid,
43 ): number {
44 let reaped = 0;
45 for (const pgid of leftoverBridgeGroups(rows, crewRoot, cmdline, self)) {
46 try {
47 kill(pgid);
48 reaped++;
49 } catch {
50 // gone between the read and the kill
51 }
52 }
53 return reaped;
54 }
55
56 function readCmdline(pid: number): string {
57 try {
58 return readFileSync(`/proc/${pid}/cmdline`, "utf8").replace(/\0/g, " ");
59 } catch {
60 return "";
61 }
62 }
src/kernel/boots.ts added +126
@@ -0,0 +1,126 @@
1 /**
2 * How often this forge has started, and how many of those starts followed
3 * a process that never said goodbye (#3708). A crash — Bun's own segfault,
4 * the kernel's OOM kill — leaves no line in the log saying so; the next
5 * boot is the only witness, so it is the one that counts.
6 *
7 * `boots.json` under the data directory holds the running total and the
8 * last {@link KEPT} starts, each with whether it exited cleanly. A start is
9 * written unclean at boot and marked clean on the way out (`close` in
10 * `main.ts`). An unclean start that is not this one is a crash — or a
11 * release's old process still draining beside this one, which marks itself
12 * clean when it goes, so the count is read live and settles back.
13 *
14 * Plain JSON written by rename: two processes overlap only during a
15 * release, and each writes twice in its life.
16 */
17
18 import { randomUUID } from "node:crypto";
19 import { readFileSync, renameSync, writeFileSync } from "node:fs";
20 import { join } from "node:path";
21
22 /** Starts remembered one by one; the total keeps counting past them. */
23 export const KEPT = 50;
24
25 interface Start {
26 readonly id: string;
27 /** When it started, ms since the epoch. */
28 readonly at: number;
29 clean: boolean;
30 }
31
32 interface Ledger {
33 total: number;
34 starts: Start[];
35 }
36
37 /** What the ledger says, from this process's point of view. */
38 export interface BootCounts {
39 /** Every start this data directory has seen, this one included. */
40 readonly boots: number;
41 /** Starts after the first: releases and crashes alike. */
42 readonly restarts: number;
43 /** Earlier starts (of the last {@link KEPT}) that never exited cleanly. */
44 readonly unclean: number;
45 }
46
47 const file = (dataDir: string): string => join(dataDir, "boots.json");
48
49 function read(dataDir: string): Ledger {
50 try {
51 const raw: unknown = JSON.parse(readFileSync(file(dataDir), "utf8"));
52 if (typeof raw !== "object" || raw === null) return { total: 0, starts: [] };
53 const r = raw as { total?: unknown; starts?: unknown };
54 const starts = Array.isArray(r.starts)
55 ? r.starts.filter(
56 (s): s is Start =>
57 typeof s === "object" &&
58 s !== null &&
59 typeof (s as Start).id === "string" &&
60 typeof (s as Start).at === "number" &&
61 typeof (s as Start).clean === "boolean",
62 )
63 : [];
64 const total =
65 typeof r.total === "number" && Number.isSafeInteger(r.total) && r.total >= starts.length
66 ? r.total
67 : starts.length;
68 return { total, starts };
69 } catch {
70 return { total: 0, starts: [] };
71 }
72 }
73
74 function write(dataDir: string, ledger: Ledger): void {
75 const tmp = `${file(dataDir)}.${process.pid}.tmp`;
76 writeFileSync(tmp, JSON.stringify(ledger));
77 renameSync(tmp, file(dataDir));
78 }
79
80 /** Counts as the process `id` sees them. */
81 function countsOf(ledger: Ledger, id: string): BootCounts {
82 return {
83 boots: ledger.total,
84 restarts: Math.max(0, ledger.total - 1),
85 unclean: ledger.starts.filter((s) => s.id !== id && !s.clean).length,
86 };
87 }
88
89 /** This process's start, once `recordBoot` has written it. */
90 let thisStart = "";
91
92 /** Write this start into the ledger. Never throws: a ledger is not worth a boot. */
93 export function recordBoot(
94 dataDir: string,
95 id: string = randomUUID(),
96 now = Date.now(),
97 ): BootCounts {
98 thisStart = id;
99 const ledger = read(dataDir);
100 ledger.total += 1;
101 ledger.starts = [...ledger.starts, { id, at: now, clean: false }].slice(-KEPT);
102 try {
103 write(dataDir, ledger);
104 } catch {
105 // a read-only data directory counts nothing, and boots anyway
106 }
107 return countsOf(ledger, id);
108 }
109
110 /** Mark start `id` as having exited cleanly. */
111 export function recordCleanExit(dataDir: string, id = thisStart): void {
112 const ledger = read(dataDir);
113 const start = ledger.starts.find((s) => s.id === id);
114 if (start === undefined || start.clean) return;
115 start.clean = true;
116 try {
117 write(dataDir, ledger);
118 } catch {
119 // nothing to be done on the way out
120 }
121 }
122
123 /** The ledger as start `id` reads it now. */
124 export function bootCounts(dataDir: string, id = thisStart): BootCounts {
125 return countsOf(read(dataDir), id);
126 }
src/kernel/census.ts added +78
@@ -0,0 +1,78 @@
1 /**
2 * The machine's process table, read from `/proc`: every process this
3 * forge can see, by command name (#3708). A pod that quietly gathers 255
4 * `socat`s is a leak no single owner notices; a count by name on a regular
5 * line is how one is noticed before it costs a crash.
6 *
7 * `root` is a parameter so a test can hand over a table it made.
8 */
9
10 import { readdirSync, readFileSync } from "node:fs";
11 import { join } from "node:path";
12
13 export interface ProcessRow {
14 readonly pid: number;
15 readonly ppid: number;
16 readonly pgid: number;
17 /** The kernel's command name (`/proc/<pid>/stat`'s second field), at most 15 bytes. */
18 readonly comm: string;
19 }
20
21 /** Every process in `root` (default `/proc`); one that goes mid-read is skipped. */
22 export function processRows(root = "/proc"): ProcessRow[] {
23 let names: string[];
24 try {
25 names = readdirSync(root);
26 } catch {
27 return [];
28 }
29 const rows: ProcessRow[] = [];
30 for (const name of names) {
31 if (!/^\d+$/.test(name)) continue;
32 let stat: string;
33 try {
34 stat = readFileSync(join(root, name, "stat"), "utf8");
35 } catch {
36 continue;
37 }
38 const row = parseStat(stat);
39 if (row !== undefined) rows.push(row);
40 }
41 return rows;
42 }
43
44 /**
45 * One `/proc/<pid>/stat` line: `pid (comm) state ppid pgrp …`. The name
46 * sits in parentheses and may hold spaces and parentheses itself, so the
47 * fields after it are found from the last `)`.
48 */
49 export function parseStat(stat: string): ProcessRow | undefined {
50 const open = stat.indexOf("(");
51 const close = stat.lastIndexOf(")");
52 if (open < 0 || close < open) return undefined;
53 const pid = Number(stat.slice(0, open).trim());
54 const fields = stat.slice(close + 2).split(" ");
55 const ppid = Number(fields[1]);
56 const pgid = Number(fields[2]);
57 if (![pid, ppid, pgid].every(Number.isSafeInteger)) return undefined;
58 return { pid, ppid, pgid, comm: stat.slice(open + 1, close) };
59 }
60
61 /** How many processes carry each command name, but this process itself. */
62 export function census(rows: readonly ProcessRow[], self = process.pid): Map<string, number> {
63 const out = new Map<string, number>();
64 for (const row of rows) {
65 if (row.pid === self) continue;
66 out.set(row.comm, (out.get(row.comm) ?? 0) + 1);
67 }
68 return out;
69 }
70
71 /** The `top` most common names as `socat=255 bwrap=148 …`, most first, ties by name. */
72 export function censusWords(counts: ReadonlyMap<string, number>, top = 8): string {
73 return [...counts]
74 .sort((a, b) => b[1] - a[1] || (a[0] < b[0] ? -1 : a[0] > b[0] ? 1 : 0))
75 .slice(0, top)
76 .map(([comm, n]) => `${comm.replace(/[\s=]/g, "_")}=${n}`)
77 .join(" ");
78 }
src/main.ts +36 −1
@@ -61,6 +61,10 @@ import { CHECK_EVERY_MS, readNewest, setBehind, watchBehind } from "./release/be
6161 import { repoDir } from "./git/repos.ts";
6262 import { snapshotPlace } from "./actions/health.ts";
6363 import { capacityDefault } from "./core/capacity.ts";
64 import { bootCounts, recordBoot, recordCleanExit } from "./kernel/boots.ts";
65 import { census, censusWords, processRows } from "./kernel/census.ts";
66 import { reapLeftoverBridges } from "./crew/leftovers.ts";
67 import { crewRoot } from "./crew/workdir.ts";
6468
6569 // `forge hook <name>` re-enters here from a git hook stub. It relays to the
6670 // running forge over its socket and exits — it must never boot a server.
@@ -487,7 +491,36 @@ const tickAsks = async (): Promise<void> => {
487491 const staleTimer = setInterval(() => void tickAsks(), 120_000);
488492 void tickAsks();
489493
490 app.log.info("forge is up", { url: servers.url, data: app.config.dataDir });
494 // Every start is counted, and one that follows a process that never said
495 // goodbye is a crash the log could not record (kernel/boots.ts, #3708).
496 const boot = recordBoot(app.config.dataDir);
497 app.log.info("forge is up", {
498 url: servers.url,
499 data: app.config.dataDir,
500 restarts: boot.restarts,
501 ...(boot.unclean === 0 ? {} : { unclean: boot.unclean }),
502 });
503 // What runs beside the forge, by name, once a minute: a helper that leaks
504 // (255 socat, #3708) shows on this line long before it costs a crash.
505 // Claude Code's sandbox bridges a finished session left behind go here.
506 const processTick = (): void => {
507 try {
508 const rows = processRows();
509 const reaped = crew === undefined ? 0 : reapLeftoverBridges(rows, crewRoot(app.config.dataDir));
510 const counts = bootCounts(app.config.dataDir);
511 app.log.info("processes", {
512 count: rows.length,
513 top: censusWords(census(rows)),
514 restarts: counts.restarts,
515 ...(counts.unclean === 0 ? {} : { unclean: counts.unclean }),
516 ...(reaped === 0 ? {} : { reapedBridges: reaped }),
517 });
518 } catch (error) {
519 app.log.warn("process census failed", { error: String(error) });
520 }
521 };
522 const processTimer = setInterval(processTick, 60_000);
523 processTick();
491524 // A new version proves itself against the public URL once (release/prove.ts).
492525 void proveRelease({
493526 dataDir: app.config.dataDir,
@@ -601,6 +634,7 @@ function drain(signal: NodeJS.Signals): Promise<unknown> {
601634 clearInterval(upstreamTimer);
602635 clearInterval(syncTimer);
603636 clearInterval(staleTimer);
637 clearInterval(processTimer);
604638 clearInterval(leaseTimer);
605639 // The platform's SIGTERM: the listener stays open while routers learn
606640 // this process is going. A person's Ctrl-C does not wait for routers.
@@ -640,6 +674,7 @@ async function close(): Promise<void> {
640674 servers.stop();
641675 clearInterval(flagTimer);
642676 flushFlagUses(app.db);
677 recordCleanExit(app.config.dataDir);
643678 app.close();
644679 process.exit(0);
645680 }
test/boots-census.test.ts added +167
@@ -0,0 +1,167 @@
1 /**
2 * What the forge counts about itself after a crash (#3708): how often it
3 * has started and how many starts followed one that never exited cleanly
4 * (`kernel/boots.ts`); what runs beside it, by name (`kernel/census.ts`);
5 * and which `socat`s are Claude Code bridges a session left behind
6 * (`crew/leftovers.ts`) — and nothing else on the machine.
7 */
8 import { afterAll, describe, expect, test } from "bun:test";
9 import { mkdirSync, readFileSync, writeFileSync } from "node:fs";
10 import { mkdtemp, rm } from "node:fs/promises";
11 import { tmpdir } from "node:os";
12 import { join } from "node:path";
13 import { bootCounts, KEPT, recordBoot, recordCleanExit } from "../src/kernel/boots.ts";
14 import { census, censusWords, parseStat, processRows, type ProcessRow } from "../src/kernel/census.ts";
15 import { leftoverBridgeGroups, reapLeftoverBridges } from "../src/crew/leftovers.ts";
16
17 const root = await mkdtemp(join(tmpdir(), "forge-boots-"));
18 afterAll(() => rm(root, { recursive: true, force: true }));
19 let n = 0;
20 const fresh = (): string => {
21 const dir = join(root, String(n++));
22 mkdirSync(dir);
23 return dir;
24 };
25
26 describe("boots: starts and crashes, counted across processes", () => {
27 test("a first start is no restart", () => {
28 const dir = fresh();
29 expect(recordBoot(dir, "a")).toEqual({ boots: 1, restarts: 0, unclean: 0 });
30 });
31
32 test("a start after a clean exit is a restart, not a crash", () => {
33 const dir = fresh();
34 recordBoot(dir, "a");
35 recordCleanExit(dir, "a");
36 expect(recordBoot(dir, "b")).toEqual({ boots: 2, restarts: 1, unclean: 0 });
37 });
38
39 test("a start after a process that never said goodbye counts the crash", () => {
40 const dir = fresh();
41 recordBoot(dir, "a"); // segfault: no clean exit
42 expect(recordBoot(dir, "b")).toEqual({ boots: 2, restarts: 1, unclean: 1 });
43 recordBoot(dir, "c"); // and again
44 expect(bootCounts(dir, "c")).toEqual({ boots: 3, restarts: 2, unclean: 2 });
45 });
46
47 test("a release's old process draining beside the new one settles back once it exits", () => {
48 const dir = fresh();
49 recordBoot(dir, "old");
50 expect(recordBoot(dir, "new").unclean).toBe(1);
51 recordCleanExit(dir, "old");
52 expect(bootCounts(dir, "new")).toEqual({ boots: 2, restarts: 1, unclean: 0 });
53 });
54
55 test("the total keeps counting past the starts it remembers one by one", () => {
56 const dir = fresh();
57 for (let i = 0; i < KEPT + 10; i++) recordBoot(dir, `s${i}`);
58 const counts = bootCounts(dir, `s${KEPT + 9}`);
59 expect(counts.boots).toBe(KEPT + 10);
60 expect(counts.unclean).toBe(KEPT - 1);
61 const kept = JSON.parse(readFileSync(join(dir, "boots.json"), "utf8")) as { starts: unknown[] };
62 expect(kept.starts.length).toBe(KEPT);
63 });
64
65 test("a corrupt ledger starts over instead of failing the boot", () => {
66 const dir = fresh();
67 writeFileSync(join(dir, "boots.json"), "{not json");
68 expect(recordBoot(dir, "a")).toEqual({ boots: 1, restarts: 0, unclean: 0 });
69 writeFileSync(join(dir, "boots.json"), JSON.stringify({ total: "x", starts: [{ id: 1 }] }));
70 expect(recordBoot(dir, "b")).toEqual({ boots: 1, restarts: 0, unclean: 0 });
71 });
72
73 test("a data directory that cannot be written still boots", () => {
74 expect(recordBoot(join(root, "no", "such", "dir"), "a").boots).toBe(1);
75 });
76 });
77
78 describe("census: the process table by name", () => {
79 test("a stat line is read past a name with spaces and parentheses", () => {
80 expect(parseStat("4242 (my (odd) name) S 1 4240 4240 0 -1")).toEqual({
81 pid: 4242,
82 ppid: 1,
83 pgid: 4240,
84 comm: "my (odd) name",
85 });
86 expect(parseStat("garbage")).toBeUndefined();
87 });
88
89 test("counts by name, most first, without this process", () => {
90 const rows: ProcessRow[] = [
91 { pid: 10, ppid: 1, pgid: 10, comm: "socat" },
92 { pid: 11, ppid: 1, pgid: 11, comm: "socat" },
93 { pid: 12, ppid: 1, pgid: 12, comm: "socat" },
94 { pid: 13, ppid: 1, pgid: 13, comm: "bwrap" },
95 { pid: 14, ppid: 1, pgid: 14, comm: "bun" },
96 { pid: 99, ppid: 1, pgid: 99, comm: "bun" },
97 ];
98 const counts = census(rows, 99);
99 expect(counts.get("socat")).toBe(3);
100 expect(counts.get("bun")).toBe(1);
101 expect(censusWords(counts)).toBe("socat=3 bun=1 bwrap=1");
102 expect(censusWords(counts, 1)).toBe("socat=3");
103 expect(censusWords(new Map([["a b=c", 2]]))).toBe("a_b_c=2");
104 });
105
106 test("a table read from a directory, skipping what is not a process", () => {
107 const proc = fresh();
108 for (const [pid, stat] of [
109 ["7", "7 (socat) S 1 7 7"],
110 ["8", "8 (bun) S 1 8 8"],
111 ] as const) {
112 mkdirSync(join(proc, pid));
113 writeFileSync(join(proc, pid, "stat"), stat);
114 }
115 mkdirSync(join(proc, "self"));
116 mkdirSync(join(proc, "9")); // gone mid-read: no stat
117 expect(processRows(proc).map((r) => r.comm).sort()).toEqual(["bun", "socat"]);
118 expect(processRows(join(proc, "missing"))).toEqual([]);
119 });
120
121 test.skipIf(process.platform !== "linux")("the real table holds this process", () => {
122 expect(processRows().some((r) => r.pid === process.pid)).toBe(true);
123 });
124 });
125
126 describe("leftovers: Claude Code bridges a session left behind", () => {
127 const crew = "/data/crew";
128 const rows: ProcessRow[] = [
129 { pid: 50, ppid: 1, pgid: 50, comm: "bun" }, // the forge
130 // An orphaned bridge and the fork it made for a connection.
131 { pid: 60, ppid: 1, pgid: 58, comm: "socat" },
132 { pid: 61, ppid: 60, pgid: 58, comm: "socat" },
133 // A live session's bridge: its parent is still there.
134 { pid: 70, ppid: 69, pgid: 69, comm: "socat" },
135 // A person's own socat, orphaned, naming nothing of the crew's.
136 { pid: 80, ppid: 1, pgid: 80, comm: "socat" },
137 // An orphan that names the crew's directory but is not socat.
138 { pid: 90, ppid: 1, pgid: 90, comm: "sleep" },
139 // A socat in the forge's own group, never killed by group.
140 { pid: 95, ppid: 1, pgid: 50, comm: "socat" },
141 ];
142 const cmd: Record<number, string> = {
143 60: "socat UNIX-LISTEN:/data/crew/torque/.t/abc/claude-http.sock,fork,reuseaddr TCP:localhost:3128",
144 61: "socat UNIX-LISTEN:/data/crew/torque/.t/abc/claude-http.sock,fork,reuseaddr TCP:localhost:3128",
145 70: "socat UNIX-LISTEN:/data/crew/torque/.t/def/claude-http.sock,fork TCP:localhost:3129",
146 80: "socat TCP-LISTEN:8080,fork TCP:example.com:80",
147 90: "sleep 300 /data/crew/x",
148 95: "socat UNIX-LISTEN:/data/crew/torque/.t/ghi/claude-socks.sock,fork TCP:localhost:1080",
149 };
150 const cmdline = (pid: number): string => cmd[pid] ?? "";
151
152 test("only an orphaned socat naming the crew's directory, by its group", () => {
153 expect(leftoverBridgeGroups(rows, crew, cmdline, 50)).toEqual([58]);
154 // A directory that merely starts the same is not the crew's.
155 expect(leftoverBridgeGroups(rows, "/data/cr", cmdline, 50)).toEqual([]);
156 });
157
158 test("each group is killed once, and one gone meanwhile is not counted", () => {
159 const killed: number[] = [];
160 expect(reapLeftoverBridges(rows, crew, (g) => killed.push(g), cmdline, 50)).toBe(1);
161 expect(killed).toEqual([58]);
162 const gone = reapLeftoverBridges(rows, crew, () => {
163 throw new Error("ESRCH");
164 }, cmdline, 50);
165 expect(gone).toBe(0);
166 });
167 });
test/cell-relays-reaped.test.ts added +109
@@ -0,0 +1,109 @@
1 /**
2 * A cell's relays end with the cell (#3708). Every cell starts `socat`s
3 * inside it — the proxy's front door, one per loopback port it may reach,
4 * and the way in — and 255 socats alive in one pod was the question. Fifty
5 * cells started and stopped leave not one of theirs behind.
6 *
7 * The relays are told apart from any other cell's on the machine by a
8 * loopback port only this file asks for: `loop-<port>.sock` is in each
9 * relay's command line.
10 */
11
12 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
13 import { readdirSync, readFileSync } from "node:fs";
14 import { mkdtemp, rm } from "node:fs/promises";
15 import { tmpdir } from "node:os";
16 import { join } from "node:path";
17 import type { CellSpec } from "../src/apps/contain.ts";
18 import { runStep } from "../src/apps/process.ts";
19 import { portInsteadOfSocket } from "../src/git/hook-address.ts";
20 import { missing } from "./helpers/tools.ts";
21
22 const noCells =
23 process.env["FORGE_CELLS"] === "off" ||
24 process.platform !== "linux" ||
25 portInsteadOfSocket() ||
26 missing("bwrap", "socat");
27
28 /** Live socat processes whose command line carries `mark`. */
29 function relays(mark: string): number[] {
30 const out: number[] = [];
31 for (const name of readdirSync("/proc")) {
32 if (!/^\d+$/.test(name)) continue;
33 try {
34 const stat = readFileSync(`/proc/${name}/stat`, "utf8");
35 const state = stat.slice(stat.lastIndexOf(")") + 2)[0];
36 if (state === "Z" || state === "X") continue;
37 const cmd = readFileSync(`/proc/${name}/cmdline`, "utf8").replace(/\0/g, " ");
38 if (cmd.startsWith("socat ") && cmd.includes(mark)) out.push(Number(name));
39 } catch {
40 // gone mid-read
41 }
42 }
43 return out;
44 }
45
46 describe.skipIf(noCells)("cells: no relay outlives its cell", () => {
47 let dir: string;
48 // A port nothing listens on: only the relay's command line needs it.
49 const port = 40_000 + Math.floor(Math.random() * 20_000);
50 const mark = `loop-${port}.sock`;
51
52 beforeAll(async () => {
53 dir = await mkdtemp(join(tmpdir(), "forge-relays-"));
54 });
55 afterAll(async () => {
56 for (const pid of relays(mark))
57 try {
58 process.kill(pid, "SIGKILL");
59 } catch {
60 // gone
61 }
62 await rm(dir, { recursive: true, force: true });
63 });
64
65 const spec = (): CellSpec => ({
66 dir,
67 writable: [],
68 home: join(dir, "home"),
69 hide: [],
70 egress: [],
71 reach: [`http://127.0.0.1:${port}`],
72 limits: { memoryMiB: 256, cpus: 1, diskMiB: 64 },
73 listen: port + 1,
74 });
75
76 const step = (command: string) =>
77 runStep("relays", command, {
78 cwd: dir,
79 env: { PATH: process.env["PATH"] ?? "/usr/bin:/bin" },
80 log: () => {},
81 signal: new AbortController().signal,
82 timeoutMs: 30_000,
83 cell: spec(),
84 });
85
86 test("a running cell's relays are visible to this count", async () => {
87 const running = step("sleep 2");
88 let seen = 0;
89 const deadline = Date.now() + 5_000;
90 while (Date.now() < deadline && seen === 0) {
91 seen = relays(mark).length;
92 await Bun.sleep(50);
93 }
94 const result = await running;
95 expect(result.ok).toBe(true);
96 expect(seen).toBeGreaterThan(0);
97 }, 30_000);
98
99 test("fifty cells started and stopped leave zero socat", async () => {
100 for (let i = 0; i < 50; i++) {
101 const result = await step("true");
102 expect(result.ok).toBe(true);
103 }
104 // The namespace's processes go as its init does; give the kernel a beat.
105 const deadline = Date.now() + 5_000;
106 while (Date.now() < deadline && relays(mark).length > 0) await Bun.sleep(50);
107 expect(relays(mark)).toEqual([]);
108 }, 180_000);
109 });
test/session-reap.test.ts added +122
@@ -0,0 +1,122 @@
1 /**
2 * A crew session's leftovers end with it (#3708). Claude Code starts its
3 * sandbox bridges (`socat UNIX-LISTEN:…,fork`) in its own process group;
4 * a session that ends without stopping them left them alive, re-parented
5 * to pid 1, and 255 socats were counted in one pod. Whatever a session
6 * leaves in its group is killed the moment it ends, however it ended.
7 */
8 import { afterAll, describe, expect, test } from "bun:test";
9 import { chmodSync, readFileSync, writeFileSync } from "node:fs";
10 import { mkdtemp, rm } from "node:fs/promises";
11 import { tmpdir } from "node:os";
12 import { join } from "node:path";
13 import { reapGroup, runClaude } from "../src/crew/claude.ts";
14 import { sandboxPolicy } from "../src/crew/sandbox.ts";
15
16 const dir = await mkdtemp(join(tmpdir(), "forge-reap-"));
17 const started: number[] = [];
18 afterAll(async () => {
19 for (const pid of started)
20 try {
21 process.kill(pid, "SIGKILL");
22 } catch {
23 // gone, as it should be
24 }
25 await rm(dir, { recursive: true, force: true });
26 });
27
28 const RESULT =
29 '{"type":"result","subtype":"success","is_error":false,"result":"done","total_cost_usd":0.01,"num_turns":1}';
30
31 /** Alive and not a zombie waiting to be reaped. */
32 function alive(pid: number): boolean {
33 try {
34 const stat = readFileSync(`/proc/${pid}/stat`, "utf8");
35 const state = stat.slice(stat.lastIndexOf(")") + 2)[0];
36 return state !== "Z" && state !== "X";
37 } catch {
38 return false;
39 }
40 }
41
42 async function goneWithin(pid: number, ms: number): Promise<boolean> {
43 const deadline = Date.now() + ms;
44 while (Date.now() < deadline) {
45 if (!alive(pid)) return true;
46 await Bun.sleep(25);
47 }
48 return !alive(pid);
49 }
50
51 /**
52 * A stand-in Claude Code that starts a bridge-like sleeper in its group
53 * (its output let go, as Claude Code's `stdio: "ignore"` does), writes its
54 * pid, then does `ending`.
55 */
56 function fake(name: string, ending: string): { binary: string; pidFile: string } {
57 const binary = join(dir, name);
58 const pidFile = join(dir, `${name}.pid`);
59 writeFileSync(
60 binary,
61 `#!/bin/sh
62 IFS= read -r first
63 sleep 300 >/dev/null 2>&1 </dev/null &
64 echo $! > ${pidFile}
65 echo '{"type":"system","subtype":"init","session_id":"s1","model":"fake"}'
66 ${ending}
67 `,
68 );
69 chmodSync(binary, 0o755);
70 return { binary, pidFile };
71 }
72
73 function input(binary: string) {
74 return {
75 binary,
76 cwd: dir,
77 prompt: "Do the thing.",
78 appendSystemPrompt: "",
79 budgetUsd: 1,
80 env: {},
81 sandbox: sandboxPolicy({ dataDir: dir, agentName: "bot", urls: [] }),
82 timeoutMs: 60_000,
83 log: () => {},
84 signal: new AbortController().signal,
85 settleMs: 1_000,
86 };
87 }
88
89 const leftover = (pidFile: string): number => {
90 const pid = Number(readFileSync(pidFile, "utf8").trim());
91 started.push(pid);
92 return pid;
93 };
94
95 describe.skipIf(process.platform !== "linux")("a crew session's leftovers", () => {
96 test("a session that finishes cleanly takes what it left running with it", async () => {
97 const { binary, pidFile } = fake("clean", `echo '${RESULT}'\nexit 0`);
98 const outcome = await runClaude(input(binary));
99 expect(outcome.ok).toBe(true);
100 expect(await goneWithin(leftover(pidFile), 3_000)).toBe(true);
101 }, 20_000);
102
103 test("a session that dies without a result takes what it left running with it", async () => {
104 const { binary, pidFile } = fake("crashed", "exit 3");
105 const outcome = await runClaude(input(binary));
106 expect(outcome.ok).toBe(false);
107 expect(outcome.reason).toBe("exit 3");
108 expect(await goneWithin(leftover(pidFile), 3_000)).toBe(true);
109 }, 20_000);
110
111 test("reapGroup never signals this process, pid 1 or a nonsense id", () => {
112 // Each of these would be fatal or meaningless; none may throw either.
113 for (const pgid of [0, 1, -5, Number.NaN, 1.5, process.pid]) reapGroup(pgid);
114 expect(alive(process.pid)).toBe(true);
115 });
116
117 test("reapGroup on an empty group is quiet", async () => {
118 const proc = Bun.spawn(["/bin/true"], { detached: true, stdout: "ignore" });
119 await proc.exited;
120 expect(() => reapGroup(proc.pid)).not.toThrow();
121 });
122 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs