fix: 完整回收 manager 外部命令子进程

This commit is contained in:
AgentRun Codex
2026-07-21 16:46:57 +02:00
parent 2d2139d4ee
commit 12623d5085
5 changed files with 50 additions and 21 deletions
+13 -2
View File
@@ -18,12 +18,15 @@ export function waitForChildProcess(child: ChildProcess, input: { signal?: Abort
return new Promise<ChildProcessExit>((resolve, reject) => {
let settled = false;
let aborted = false;
let spawnError: Error | null = null;
let exit: ChildProcessExit | null = null;
let killTimer: NodeJS.Timeout | null = null;
const cleanup = (): void => {
if (killTimer) clearTimeout(killTimer);
input.signal?.removeEventListener("abort", onAbort);
child.removeListener("error", onError);
child.removeListener("exit", onExit);
child.removeListener("close", onClose);
};
const finish = (callback: () => void): void => {
@@ -42,13 +45,21 @@ export function waitForChildProcess(child: ChildProcess, input: { signal?: Abort
killTimer.unref?.();
};
const onError = (error: Error): void => {
finish(() => reject(aborted ? abortedError(input.signal, input.label) : new AgentRunError("infra-failed", `failed to start ${input.label}: ${redactText(error.message)}`, { httpStatus: 503 })));
spawnError = error;
};
const onExit = (code: number | null, signal: NodeJS.Signals | null): void => {
exit = { code, signal };
};
const onClose = (code: number | null, signal: NodeJS.Signals | null): void => {
finish(() => aborted ? reject(abortedError(input.signal, input.label)) : resolve({ code, signal }));
finish(() => {
if (aborted) reject(abortedError(input.signal, input.label));
else if (spawnError) reject(new AgentRunError("infra-failed", `failed to start ${input.label}: ${redactText(spawnError.message)}`, { httpStatus: 503 }));
else resolve(exit ?? { code, signal });
});
};
child.once("error", onError);
child.once("exit", onExit);
child.once("close", onClose);
input.signal?.addEventListener("abort", onAbort, { once: true });
if (input.signal?.aborted) onAbort();
+3 -6
View File
@@ -7,6 +7,7 @@ import type { AgentRunStore } from "./store.js";
import type { BackendProfile, ExecutionPolicy, JsonRecord, JsonValue } from "../common/types.js";
import { asRecord, validateBackendProfile } from "../common/validation.js";
import { redactJson, redactText } from "../common/redaction.js";
import { waitForChildProcess } from "./abortable-child.js";
import { createKubernetesRunnerJob, type RunnerJobDefaults } from "./kubernetes-runner-job.js";
import { buildRunResult } from "./result.js";
import { runnerJobStatusSummary } from "./runner-job-status.js";
@@ -809,13 +810,9 @@ async function runKubectl(kubectlCommand: string, args: string[], stdin?: string
child.stderr.setEncoding("utf8");
child.stdout.on("data", (chunk) => { stdout += String(chunk); });
child.stderr.on("data", (chunk) => { stderr += String(chunk); });
const resultPromise = waitForChildProcess(child, { label: "kubectl" });
child.stdin.end(stdin ?? "");
const result = await new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => {
child.on("error", reject);
child.on("close", (code, signal) => resolve({ code, signal }));
}).catch((error: unknown) => {
throw new AgentRunError("infra-failed", `failed to start kubectl: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 503 });
});
const result = await resultPromise;
return { ...result, stdout, stderr };
}
+2 -6
View File
@@ -8,6 +8,7 @@ import { agentRunBusinessTraceId } from "../common/otel-trace.js";
import { nowIso, stableHash } from "../common/validation.js";
import type { AgentRunStore } from "./store.js";
import { isTerminalCommandState, isTerminalRunStatus } from "./store.js";
import { waitForChildProcess } from "./abortable-child.js";
import { reconcileStalePendingRunners, type RunnerRetentionOptions } from "./runner-retention.js";
export interface RunnerReconcilerOptions {
@@ -1020,12 +1021,7 @@ async function kubectlRun(kubectlCommand: string, args: string[]): Promise<{ cod
child.stderr.setEncoding("utf8");
child.stdout.on("data", (chunk) => { stdout += String(chunk); });
child.stderr.on("data", (chunk) => { stderr += String(chunk); });
const result = await new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => {
child.on("error", reject);
child.on("close", (code, signal) => resolve({ code, signal }));
}).catch((error: unknown) => {
throw new AgentRunError("infra-failed", `failed to start kubectl: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 503 });
});
const result = await waitForChildProcess(child, { label: "kubectl" });
return { ...result, stdout, stderr };
}
+3 -6
View File
@@ -4,6 +4,7 @@ import { AgentRunError } from "../common/errors.js";
import type { JsonRecord } from "../common/types.js";
import { asRecord } from "../common/validation.js";
import { redactJson, redactText } from "../common/redaction.js";
import { waitForChildProcess } from "./abortable-child.js";
const defaultNamespace = "agentrun-v01";
const annotationPrefix = "agentrun.pikastech.local/tool-credential";
@@ -251,13 +252,9 @@ async function runKubectl(kubectlCommand: string, args: string[], stdin?: string
child.stderr.setEncoding("utf8");
child.stdout.on("data", (chunk) => { stdout += String(chunk); });
child.stderr.on("data", (chunk) => { stderr += String(chunk); });
const resultPromise = waitForChildProcess(child, { label: "kubectl" });
child.stdin.end(stdin ?? "");
const result = await new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => {
child.on("error", reject);
child.on("close", (code, signal) => resolve({ code, signal }));
}).catch((error: unknown) => {
throw new AgentRunError("infra-failed", `failed to start kubectl: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 503 });
});
const result = await resultPromise;
return { ...result, stdout, stderr };
}
+29 -1
View File
@@ -1,5 +1,5 @@
import assert from "node:assert/strict";
import { chmod, writeFile } from "node:fs/promises";
import { chmod, readdir, readFile, writeFile } from "node:fs/promises";
import path from "node:path";
import type { JsonRecord } from "../../common/types.js";
import { reconcileRunnerJobsOnce, type RunnerStartupRecoveryOptions } from "../../mgr/runner-reconciler.js";
@@ -28,6 +28,7 @@ const selfTest: SelfTestCase = async (context) => {
await assertDeterministicFailureDoesNotRetry(context, fixturePath, kubectlCommand);
await assertFailedJobWithoutTerminalOutboxFails(context, fixturePath, kubectlCommand);
await assertSucceededJobWithoutTerminalOutboxFails(context, fixturePath, kubectlCommand);
await assertRepeatedReconcileReapsKubectl(fixturePath, kubectlCommand);
await assertRunnerBootObservationLifecycle();
} finally {
if (previousFixture === undefined) delete process.env.AGENTRUN_SELFTEST_STARTUP_FIXTURE_PATH;
@@ -41,12 +42,39 @@ const selfTest: SelfTestCase = async (context) => {
"image-not-found-fails-without-retry",
"failed-job-without-terminal-outbox-fails",
"succeeded-job-without-terminal-outbox-fails",
"repeated-manager-kubectl-commands-are-reaped",
"semantic-failure-events-carry-workbench-fields",
"runner-git-mirror-boot-is-visible-before-claim",
],
};
};
async function assertRepeatedReconcileReapsKubectl(fixturePath: string, kubectlCommand: string): Promise<void> {
if (process.platform !== "linux") return;
const fact = createRunnerFact("child-reaping");
await writeFixture(fixturePath, "", "", "running");
for (let index = 0; index < 32; index++) {
await reconcileRunnerJobsOnce({ store: fact.store, namespace: "agentrun-v02", kubectlCommand, startupRecovery: policy });
}
assert.deepEqual(await ownedZombieChildren(), []);
}
async function ownedZombieChildren(): Promise<Array<{ pid: number; command: string }>> {
const taskIds = await readdir(`/proc/${process.pid}/task`);
const childIds = new Set<string>();
for (const taskId of taskIds) {
const children = await readFile(`/proc/${process.pid}/task/${taskId}/children`, "utf8").catch(() => "");
for (const pid of children.trim().split(/\s+/u).filter(Boolean)) childIds.add(pid);
}
const zombies: Array<{ pid: number; command: string }> = [];
for (const pid of childIds) {
const stat = await readFile(`/proc/${pid}/stat`, "utf8").catch(() => "");
const match = stat.match(/^\d+ \((.*)\) ([A-Z])/u);
if (match?.[2] === "Z") zombies.push({ pid: Number(pid), command: match[1] ?? "" });
}
return zombies.sort((left, right) => left.pid - right.pid);
}
async function assertRunnerBootObservationLifecycle(): Promise<void> {
const fact = createRunnerFact("boot-observation");
const identity = { runId: fact.runId, commandId: fact.commandId, attemptId: "attempt-boot-observation", runnerId: "runner-boot-observation" };