mirror of
https://github.com/morluto/rea.git
synced 2026-11-08 07:46:28 +00:00
fix: report provider admission failures as typed codes (#8066)
* fix: report provider admission failures as typed codes * fix(evm): keep limiter path on admission diagnostics * fix: assert WABT and CBOR admission codes * fix: keep admission details on returned failures * fix(evm): detect repeated CBOR keys in the decoder --------- Co-authored-by: Cursor Agent <cursoragent@cursor.com>
This commit is contained in:
1 parent
3b647132c6
commit
317bf3fe23
63 files changed
+1021
-432
No files matched your search
@@ -1,3 +1,4 @@
|
||||
import { ProcessAdmissionError } from "./ProcessAdmissionError.js";
|
||||
import { mkdir, mkdtemp, rm } from "node:fs/promises";
|
||||
import { rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
@@ -579,10 +580,10 @@ export const createDarwinProcessRunTokenReader = (
|
||||
};
|
||||
|
||||
const compile = async (signal?: AbortSignal): Promise<string> => {
|
||||
if (closed) throw new Error("Darwin process token reader is closed");
|
||||
if (closed) throw new ProcessAdmissionError("reader_closed");
|
||||
signal?.throwIfAborted();
|
||||
while (true) {
|
||||
if (closed) throw new Error("Darwin process token reader is closed");
|
||||
if (closed) throw new ProcessAdmissionError("reader_closed");
|
||||
signal?.throwIfAborted();
|
||||
if (executable !== undefined) return executable;
|
||||
if (compilation === undefined && root !== undefined) {
|
||||
@@ -642,7 +643,7 @@ export const createDarwinProcessRunTokenReader = (
|
||||
signal?: AbortSignal,
|
||||
): Promise<string> => {
|
||||
if (closed)
|
||||
return Promise.reject(new Error("Darwin process token reader is closed"));
|
||||
return Promise.reject(new ProcessAdmissionError("reader_closed"));
|
||||
const pending = (async () => {
|
||||
const binary = await compile(signal);
|
||||
const { stdout } = await execFileOutput(binary, arguments_, {
|
||||
@@ -670,25 +671,21 @@ export const createDarwinProcessRunTokenReader = (
|
||||
const stdout = await runHelper(pids, signal);
|
||||
const decoded = safeParseJson(stdout);
|
||||
if (!decoded.ok)
|
||||
throw new Error(
|
||||
`macOS process token helper returned invalid JSON: ${decoded.error}`,
|
||||
throw new ProcessAdmissionError(
|
||||
"token_json_invalid",
|
||||
{ parser: decoded.error },
|
||||
{ cause: decoded.cause },
|
||||
);
|
||||
const parsed = responseSchema.safeParse(decoded.value);
|
||||
if (!parsed.success)
|
||||
throw new Error(
|
||||
"macOS process token helper returned an invalid response",
|
||||
{
|
||||
cause: parsed.error,
|
||||
},
|
||||
);
|
||||
throw new ProcessAdmissionError("token_response_invalid", undefined, {
|
||||
cause: parsed.error,
|
||||
});
|
||||
const requested = new Set(processes.map(({ pid }) => pid));
|
||||
const observations = new Map<number, ProcessRunTokenObservation>();
|
||||
for (const result of parsed.data.results) {
|
||||
if (!requested.has(result.pid) || observations.has(result.pid))
|
||||
throw new Error(
|
||||
"macOS process token helper returned an unexpected PID set",
|
||||
);
|
||||
throw new ProcessAdmissionError("token_pid_set");
|
||||
observations.set(
|
||||
result.pid,
|
||||
result.state === "readable"
|
||||
@@ -701,9 +698,7 @@ export const createDarwinProcessRunTokenReader = (
|
||||
);
|
||||
}
|
||||
if (observations.size !== requested.size)
|
||||
throw new Error(
|
||||
"macOS process token helper returned an incomplete PID set",
|
||||
);
|
||||
throw new ProcessAdmissionError("token_pid_incomplete");
|
||||
return observations;
|
||||
};
|
||||
|
||||
@@ -716,25 +711,21 @@ export const createDarwinProcessRunTokenReader = (
|
||||
const stdout = await runHelper(["--identities", ...pids], signal);
|
||||
const decoded = safeParseJson(stdout);
|
||||
if (!decoded.ok)
|
||||
throw new Error(
|
||||
`macOS process identity helper returned invalid JSON: ${decoded.error}`,
|
||||
throw new ProcessAdmissionError(
|
||||
"identity_json_invalid",
|
||||
{ parser: decoded.error },
|
||||
{ cause: decoded.cause },
|
||||
);
|
||||
const parsed = identityResponseSchema.safeParse(decoded.value);
|
||||
if (!parsed.success)
|
||||
throw new Error(
|
||||
"macOS process identity helper returned an invalid response",
|
||||
{
|
||||
cause: parsed.error,
|
||||
},
|
||||
);
|
||||
throw new ProcessAdmissionError("identity_response_invalid", undefined, {
|
||||
cause: parsed.error,
|
||||
});
|
||||
const requested = new Set(processes.map(({ pid }) => pid));
|
||||
const observations = new Map<number, ProcessIdentityObservation>();
|
||||
for (const result of parsed.data.results) {
|
||||
if (!requested.has(result.pid) || observations.has(result.pid))
|
||||
throw new Error(
|
||||
"macOS process identity helper returned an unexpected PID set",
|
||||
);
|
||||
throw new ProcessAdmissionError("identity_pid_set");
|
||||
observations.set(
|
||||
result.pid,
|
||||
result.state === "readable"
|
||||
@@ -768,9 +759,7 @@ export const createDarwinProcessRunTokenReader = (
|
||||
);
|
||||
}
|
||||
if (observations.size !== requested.size)
|
||||
throw new Error(
|
||||
"macOS process identity helper returned an incomplete PID set",
|
||||
);
|
||||
throw new ProcessAdmissionError("identity_pid_incomplete");
|
||||
return observations;
|
||||
};
|
||||
|
||||
|
||||
@@ -176,7 +176,9 @@ it.each([
|
||||
expect(physical.outcome.value).toEqual(configured);
|
||||
else
|
||||
expect(physical.outcome.cause).toMatchObject({
|
||||
message: "Cannot establish inherited Linux limit: Max address space",
|
||||
_tag: "ProcessAdmissionError",
|
||||
code: "limit_unreadable",
|
||||
detail: { limit: "address_space_bytes" },
|
||||
});
|
||||
expect(physical.owner.handle.fd).toBeGreaterThanOrEqual(0);
|
||||
if (scoped)
|
||||
@@ -267,9 +269,11 @@ it.each([16 * 1024, 16 * 1024 + 1])(
|
||||
const prefix = report("unlimited unlimited");
|
||||
await kernelReportFixture(prefix.padEnd(bytes, " "));
|
||||
kernel.maximumRead = Buffer.byteLength(prefix);
|
||||
await expect(readLinuxResourceLimits(configured)).rejects.toThrow(
|
||||
"Inherited Linux limits exceed the report budget.",
|
||||
);
|
||||
await expect(readLinuxResourceLimits(configured)).rejects.toMatchObject({
|
||||
_tag: "ProcessAdmissionError",
|
||||
code: "report_exceeds_budget",
|
||||
detail: { budget_bytes: 16 * 1024 },
|
||||
});
|
||||
expect(kernel.reads.length).toBeGreaterThan(1);
|
||||
expect(
|
||||
kernel.reads.every(
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { ProcessAdmissionError } from "./ProcessAdmissionError.js";
|
||||
import { open } from "node:fs/promises";
|
||||
import { readFileHandleRange } from "../filesystem/readFileHandleRange.js";
|
||||
import { withArtifactFileHandle } from "../artifacts/ArtifactFileOperation.js";
|
||||
@@ -57,12 +58,12 @@ export const clampLinuxResourceLimits = (
|
||||
.filter((line) => line.startsWith(`${rows[key]} `));
|
||||
const values = matches[0]?.slice(rows[key].length).trim().split(/\s+/u);
|
||||
if (matches.length !== 1 || values === undefined || values.length !== 3)
|
||||
throw new Error(`Cannot establish inherited Linux limit: ${rows[key]}`);
|
||||
throw new ProcessAdmissionError("limit_unreadable", { limit: key });
|
||||
let ceiling = BigInt(configured[key]);
|
||||
for (const token of values.slice(0, 2)) {
|
||||
if (token === "unlimited") continue;
|
||||
if (!/^[0-9]+$/u.test(token))
|
||||
throw new Error(`Invalid inherited Linux limit: ${rows[key]}`);
|
||||
throw new ProcessAdmissionError("limit_invalid", { limit: key });
|
||||
const value = BigInt(token);
|
||||
if (value < ceiling) ceiling = value;
|
||||
}
|
||||
@@ -86,7 +87,9 @@ export const readLinuxResourceLimits = async (
|
||||
const reportBudget = 16 * 1024;
|
||||
const buffer = await readFileHandleRange(handle, 0, reportBudget);
|
||||
if (buffer.length === reportBudget)
|
||||
throw new Error("Inherited Linux limits exceed the report budget.");
|
||||
throw new ProcessAdmissionError("report_exceeds_budget", {
|
||||
budget_bytes: reportBudget,
|
||||
});
|
||||
return clampLinuxResourceLimits(configured, buffer.toString("utf8"));
|
||||
},
|
||||
);
|
||||
|
||||
@@ -331,9 +331,10 @@ it("runs deferred delegated cleanup before registering a pending start acquisiti
|
||||
expect(stopped.status).not.toBe("incomplete");
|
||||
runtime.retainForCleanup();
|
||||
|
||||
await expect(runtime.start(startOptions())).rejects.toThrow(
|
||||
"Owned worker runtime is closing",
|
||||
);
|
||||
await expect(runtime.start(startOptions())).rejects.toMatchObject({
|
||||
_tag: "ProcessAdmissionError",
|
||||
code: "runtime_closing",
|
||||
});
|
||||
expect(delegateTimedOut).toBe(false);
|
||||
await delegatedClose;
|
||||
expect(runtime.rootPath).toBeUndefined();
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { ProcessAdmissionError } from "./ProcessAdmissionError.js";
|
||||
import { PrivateRuntimeRoot } from "./PrivateRuntimeRoot.js";
|
||||
import type { Readable, Writable } from "node:stream";
|
||||
import {
|
||||
@@ -211,14 +212,14 @@ export class OwnedWorkerRuntime {
|
||||
initialize?: (supervisor: ProviderProcessSupervisor) => Promise<void>,
|
||||
): Promise<void> {
|
||||
if (this.#terminalCloseRequested || this.#closing)
|
||||
return Promise.reject(new Error("Owned worker runtime is closing"));
|
||||
return Promise.reject(new ProcessAdmissionError("runtime_closing"));
|
||||
if (this.#acquisition !== undefined) return this.#acquisition;
|
||||
if (this.#supervisor !== undefined && !this.#registry.has(this))
|
||||
return Promise.resolve();
|
||||
if (this.options.retryCleanup !== undefined && this.#registry.has(this))
|
||||
return this.#retryOwnedCleanup().then(() => {
|
||||
if (this.#terminalCloseRequested || this.#closing)
|
||||
throw new Error("Owned worker runtime is closing");
|
||||
throw new ProcessAdmissionError("runtime_closing");
|
||||
return this.start(spawnOptions, initialize);
|
||||
});
|
||||
const ownRetry = this.#registry.has(this)
|
||||
@@ -241,7 +242,7 @@ export class OwnedWorkerRuntime {
|
||||
onStdout(listener: (chunk: Buffer | string) => void): void {
|
||||
const stream = this.#supervisor?.launch.process.stdout;
|
||||
if (stream === null || stream === undefined)
|
||||
throw new Error("Owned worker runtime has no stdout stream");
|
||||
throw new ProcessAdmissionError("stdout_missing");
|
||||
stream.on("data", listener);
|
||||
this.#stdoutListeners.add({ stream, listener });
|
||||
}
|
||||
@@ -330,12 +331,12 @@ export class OwnedWorkerRuntime {
|
||||
| undefined,
|
||||
): Promise<void> {
|
||||
await this.#registry.retry(this);
|
||||
if (this.#closing) throw new Error("Owned worker runtime is closing");
|
||||
if (this.#closing) throw new ProcessAdmissionError("runtime_closing");
|
||||
this.#root ??= await PrivateRuntimeRoot.create({
|
||||
prefix: this.options.rootPrefix,
|
||||
});
|
||||
this.#cleanupEpoch ??= new OwnedWorkerRuntimeCleanupEpoch(this.#root.path);
|
||||
if (this.#closing) throw new Error("Owned worker runtime is closing");
|
||||
if (this.#closing) throw new ProcessAdmissionError("runtime_closing");
|
||||
this.#lastRunId = spawnOptions.runId;
|
||||
const spawned = await (this.options.launcher ?? spawnOwnedProviderProcess)({
|
||||
...spawnOptions,
|
||||
@@ -356,9 +357,9 @@ export class OwnedWorkerRuntime {
|
||||
);
|
||||
// A close may have begun while spawn was pending. Adopt the process first;
|
||||
// close waits for this acquisition and then stops the exact adopted owner.
|
||||
if (this.#closing) throw new Error("Owned worker runtime is closing");
|
||||
if (this.#closing) throw new ProcessAdmissionError("runtime_closing");
|
||||
await initialize?.(this.#supervisor);
|
||||
if (this.#closing) throw new Error("Owned worker runtime is closing");
|
||||
if (this.#closing) throw new ProcessAdmissionError("runtime_closing");
|
||||
}
|
||||
|
||||
async #stop(): Promise<ProviderProcessStopResult> {
|
||||
@@ -430,7 +431,9 @@ export class OwnedWorkerRuntime {
|
||||
this.#inputFailure =
|
||||
cause instanceof Error
|
||||
? cause
|
||||
: new Error("Owned worker request stream failed", { cause });
|
||||
: new ProcessAdmissionError("request_stream_failed", undefined, {
|
||||
cause,
|
||||
});
|
||||
this.options.onInputFailure?.(this.#inputFailure);
|
||||
};
|
||||
const onClose = (): void => {
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { ProcessAdmissionError } from "./ProcessAdmissionError.js";
|
||||
|
||||
/** Timeout and cancellation projections for one correlated operation. */
|
||||
export type PendingOperationOptions<Value> = {
|
||||
readonly signal?: AbortSignal;
|
||||
@@ -36,7 +38,7 @@ export class PendingOperations<Key, Value> {
|
||||
/** Register one unique operation key and await its reply or cancellation. */
|
||||
wait(key: Key, options: PendingOperationOptions<Value>): Promise<Value> {
|
||||
if (this.#pending.has(key))
|
||||
throw new Error("Pending operation key is already registered");
|
||||
throw new ProcessAdmissionError("duplicate_key");
|
||||
if (options.signal?.aborted === true)
|
||||
return Promise.resolve(options.cancelledValue());
|
||||
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
const PROCESS_ADMISSION_CODES = [
|
||||
"runtime_closing",
|
||||
"stdout_missing",
|
||||
"request_stream_failed",
|
||||
"finalization_unobserved",
|
||||
"prefix_identity_lost",
|
||||
"prefix_unavailable",
|
||||
"identity_snapshot_unavailable",
|
||||
"current_process_unreported",
|
||||
"baseline_unavailable",
|
||||
"memory_limit_unsupported",
|
||||
"launcher_pid_missing",
|
||||
"duplicate_key",
|
||||
"reader_closed",
|
||||
"token_json_invalid",
|
||||
"token_response_invalid",
|
||||
"token_pid_set",
|
||||
"token_pid_incomplete",
|
||||
"identity_json_invalid",
|
||||
"identity_response_invalid",
|
||||
"identity_pid_set",
|
||||
"identity_pid_incomplete",
|
||||
"limit_unreadable",
|
||||
"limit_invalid",
|
||||
"report_exceeds_budget",
|
||||
] as const;
|
||||
|
||||
export type ProcessAdmissionCode = (typeof PROCESS_ADMISSION_CODES)[number];
|
||||
|
||||
/** Process ownership, worker, or limit admission failure. `code` is the stable identity. */
|
||||
export class ProcessAdmissionError extends Error {
|
||||
readonly _tag = "ProcessAdmissionError";
|
||||
|
||||
constructor(
|
||||
readonly code: ProcessAdmissionCode,
|
||||
readonly detail?: Readonly<Record<string, string | number>>,
|
||||
options?: ErrorOptions,
|
||||
) {
|
||||
super(code, options);
|
||||
this.name = "ProcessAdmissionError";
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { ProcessAdmissionError } from "./ProcessAdmissionError.js";
|
||||
import { readFile, stat as statPath } from "node:fs/promises";
|
||||
import { performance } from "node:perf_hooks";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
@@ -1348,9 +1349,9 @@ export const createSystemProcessOwnershipHost = (
|
||||
liveAfterRetry.some((entry) => entry.pid === pid),
|
||||
);
|
||||
if (stillUnresolved.length > 0)
|
||||
throw new Error(
|
||||
`process identity snapshot is unavailable for ${String(stillUnresolved.length)} live processes`,
|
||||
);
|
||||
throw new ProcessAdmissionError("identity_snapshot_unavailable", {
|
||||
live_processes: stillUnresolved.length,
|
||||
});
|
||||
}
|
||||
return processes.map(({ pid }) => {
|
||||
const observation = identities.get(pid);
|
||||
@@ -1369,7 +1370,7 @@ export const createSystemProcessOwnershipHost = (
|
||||
try {
|
||||
const processes = await listProcesses(signal);
|
||||
if (!processes.some(({ pid }) => pid === process.pid))
|
||||
throw new Error("ps did not report REA's current process");
|
||||
throw new ProcessAdmissionError("current_process_unreported");
|
||||
} catch (cause: unknown) {
|
||||
if (isExpectedAbort(cause, signal)) throw cause;
|
||||
throw new LinuxOwnedProcessPrerequisiteError(cause);
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { ProcessAdmissionError } from "./ProcessAdmissionError.js";
|
||||
import { spawn, type ChildProcess } from "node:child_process";
|
||||
import type { EventEmitter } from "node:events";
|
||||
import type { Readable, Writable } from "node:stream";
|
||||
@@ -263,7 +264,7 @@ export const spawnOwnedProviderProcess = async (
|
||||
if (sweepTokenOwnedProcesses) {
|
||||
const capture = systemProcessOwnershipHost.captureBaseline;
|
||||
if (capture === undefined)
|
||||
throw new Error("Process ownership baseline is unavailable on this host");
|
||||
throw new ProcessAdmissionError("baseline_unavailable");
|
||||
captureBaseline = await capture(options.signal);
|
||||
options.signal?.throwIfAborted();
|
||||
}
|
||||
@@ -306,9 +307,7 @@ export const spawnOwnedProviderProcess = async (
|
||||
};
|
||||
}
|
||||
if (options.memoryLimitBytes !== undefined)
|
||||
throw new Error(
|
||||
"Owned process memory limits require the Windows Job Object launcher; no limit can be enforced on this host.",
|
||||
);
|
||||
throw new ProcessAdmissionError("memory_limit_unsupported");
|
||||
const child = spawn(options.command, [...options.arguments], {
|
||||
shell: false,
|
||||
windowsHide: true,
|
||||
@@ -320,7 +319,8 @@ export const spawnOwnedProviderProcess = async (
|
||||
});
|
||||
await waitForSpawn(child);
|
||||
const pid = child.pid;
|
||||
if (pid === undefined) throw new Error("Provider launcher has no process ID");
|
||||
if (pid === undefined)
|
||||
throw new ProcessAdmissionError("launcher_pid_missing");
|
||||
const ownership: OwnedProcessGroup = {
|
||||
runId: options.runId,
|
||||
leaderPid: pid,
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { ProcessAdmissionError } from "../ProcessAdmissionError.js";
|
||||
import type { IPty } from "@lydell/node-pty";
|
||||
import type { ProcessCapture } from "../../domain/process/processCaptureParsing.js";
|
||||
import type { ProcessScenario } from "../../domain/process/processScenario.js";
|
||||
@@ -846,9 +847,7 @@ const runProcessScenario = async (
|
||||
};
|
||||
if (exit.reason === "cancelled")
|
||||
throw processCaptureCancelled(signal?.reason);
|
||||
throw new Error(
|
||||
"The captured process exit was not observed because the finalization SIGKILL could not be delivered.",
|
||||
);
|
||||
throw new ProcessAdmissionError("finalization_unobserved");
|
||||
}
|
||||
if (exit.finalization !== undefined)
|
||||
observations.finalization = {
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { ProcessAdmissionError } from "../ProcessAdmissionError.js";
|
||||
import { AnalysisError } from "../../domain/analysisErrorBase.js";
|
||||
import { AnalysisResourceConstraintError } from "../../domain/analysisErrorCore.js";
|
||||
import { err, ok, type Result } from "../../domain/result.js";
|
||||
@@ -134,7 +135,7 @@ export class ProcessSensitiveOutput {
|
||||
#prefix(index: number): Prefix {
|
||||
const value = this.prefixes[index];
|
||||
if (value === undefined)
|
||||
throw new Error("Sensitive output matcher lost its prefix identity");
|
||||
throw new ProcessAdmissionError("prefix_identity_lost");
|
||||
return value;
|
||||
}
|
||||
|
||||
@@ -342,7 +343,7 @@ export const createProcessSensitiveOutput = (
|
||||
const read = (index: number): Prefix => {
|
||||
const node = prefixes[index];
|
||||
if (node === undefined)
|
||||
throw new Error("Sensitive output prefix is unavailable");
|
||||
throw new ProcessAdmissionError("prefix_unavailable");
|
||||
return node;
|
||||
};
|
||||
try {
|
||||
|
||||
Reference in new issue
Block a user