From 12623d5085c1555f230077d5d8a7fb74d5fed2ee Mon Sep 17 00:00:00 2001 From: AgentRun Codex Date: Tue, 21 Jul 2026 16:46:57 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=AE=8C=E6=95=B4=E5=9B=9E=E6=94=B6=20m?= =?UTF-8?q?anager=20=E5=A4=96=E9=83=A8=E5=91=BD=E4=BB=A4=E5=AD=90=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mgr/abortable-child.ts | 15 ++++++++-- src/mgr/provider-profiles.ts | 9 ++---- src/mgr/runner-reconciler.ts | 8 ++--- src/mgr/tool-credentials.ts | 9 ++---- src/selftest/cases/24-runner-startup-retry.ts | 30 ++++++++++++++++++- 5 files changed, 50 insertions(+), 21 deletions(-) diff --git a/src/mgr/abortable-child.ts b/src/mgr/abortable-child.ts index 85f05d6..98fb872 100644 --- a/src/mgr/abortable-child.ts +++ b/src/mgr/abortable-child.ts @@ -18,12 +18,15 @@ export function waitForChildProcess(child: ChildProcess, input: { signal?: Abort return new Promise((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(); diff --git a/src/mgr/provider-profiles.ts b/src/mgr/provider-profiles.ts index dd9a254..f9a6715 100644 --- a/src/mgr/provider-profiles.ts +++ b/src/mgr/provider-profiles.ts @@ -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 }; } diff --git a/src/mgr/runner-reconciler.ts b/src/mgr/runner-reconciler.ts index b97a734..a914e07 100644 --- a/src/mgr/runner-reconciler.ts +++ b/src/mgr/runner-reconciler.ts @@ -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 }; } diff --git a/src/mgr/tool-credentials.ts b/src/mgr/tool-credentials.ts index bc8e8e9..0130151 100644 --- a/src/mgr/tool-credentials.ts +++ b/src/mgr/tool-credentials.ts @@ -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 }; } diff --git a/src/selftest/cases/24-runner-startup-retry.ts b/src/selftest/cases/24-runner-startup-retry.ts index 256b03f..962d251 100644 --- a/src/selftest/cases/24-runner-startup-retry.ts +++ b/src/selftest/cases/24-runner-startup-retry.ts @@ -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 { + 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> { + const taskIds = await readdir(`/proc/${process.pid}/task`); + const childIds = new Set(); + 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 { const fact = createRunnerFact("boot-observation"); const identity = { runId: fact.runId, commandId: fact.commandId, attemptId: "attempt-boot-observation", runnerId: "runner-boot-observation" };