fix(process): preserve observed command failures and native ownership

This commit is contained in:
morluto committed 2026-10-20 17:17:14 +08:00
1 parent cf3288d8cf
commit e367750b1b
6 files changed
+1079 -88

No files matched your search

+91 -24
View File
@@ -4,6 +4,7 @@ import { createReadStream } from "node:fs";
import { realpath } from "node:fs/promises";
import { err, ok, type Result } from "../domain/result.js";
import { ProviderCleanupError } from "../domain/providerCleanupError.js";
import { DarwinProcessOwnershipInspectionError } from "../process/DarwinProcessRunTokenReader.js";
import {
OwnedCommandFailure,
@@ -45,8 +46,11 @@ export interface NativeCommandCapture {
/** Native command failure with retained partial output and cleanup status. */
export class NativeCommandFailure extends Error {
/** Caller-owned cancellation fact retained separately from the physical cause. */
readonly cancellation: { readonly reason: unknown } | undefined;
readonly cleanupOwner: ProviderProcessSupervisor | undefined;
readonly cleanupFailures: readonly NativeCommandCleanupFailure[];
readonly cleanupMarker: ProviderCleanupError | undefined;
readonly detail: string | null;
readonly preparationFailure:
| DarwinProcessOwnershipInspectionError
@@ -68,8 +72,13 @@ export class NativeCommandFailure extends Error {
readonly cleanupResources: readonly string[] = [],
) {
super(`Native command ${tool} failed: ${reason}`, options);
this.cancellation =
options?.cancellation === undefined
? undefined
: { reason: options.cancellation.reason };
this.cleanupOwner = options?.cleanupOwner;
this.cleanupFailures = options?.cleanupFailures ?? [];
this.cleanupMarker = options?.cleanupMarker;
this.preparationFailure =
options?.cause instanceof DarwinProcessOwnershipInspectionError
? options.cause
@@ -83,10 +92,14 @@ export class NativeCommandFailure extends Error {
/** Named cleanup ownership carried alongside the ordinary error cause. */
export interface NativeCommandFailureOptions extends ErrorOptions {
/** Exact opaque reason recorded by the cancellation-owning boundary. */
readonly cancellation?: { readonly reason: unknown };
/** Typed owner retained when command process cleanup was incomplete. */
readonly cleanupOwner?: ProviderProcessSupervisor;
/** All owners still incomplete after a preflight retry blocked this command. */
readonly cleanupFailures?: readonly NativeCommandCleanupFailure[];
/** Exact cleanup error resolved only by a successful retry of its owner. */
readonly cleanupMarker?: ProviderCleanupError;
/** Host-side reason the command could not be supervised, such as a missing ownership helper compiler. */
readonly detail?: string;
}
@@ -96,6 +109,7 @@ export interface NativeCommandCleanupFailure {
readonly owner: ProviderProcessSupervisor;
readonly reason: string;
readonly resources: readonly string[];
readonly marker?: ProviderCleanupError;
}
/** Partial output and lifecycle details retained when a native command fails. */
@@ -140,7 +154,10 @@ export type NativeToolResolver = (
/** Run allowlisted Xcode tools directly without a shell. */
export class XcrunCommandRunner implements NativeCommandRunner {
private readonly environment: NodeJS.ProcessEnv;
readonly #pendingCleanup = new Set<ProviderProcessSupervisor>();
readonly #cleanupMarkers = new Map<
ProviderProcessSupervisor,
ProviderCleanupError
>();
#cleanupAttempt: Promise<readonly NativeCommandCleanupFailure[]> | undefined;
constructor(
environment: Readonly<NodeJS.ProcessEnv>,
@@ -165,7 +182,7 @@ export class XcrunCommandRunner implements NativeCommandRunner {
/** Retry every retained owner; a failed retry prevents another command launch. */
retryCleanup(): Promise<readonly NativeCommandCleanupFailure[]> {
if (this.#pendingCleanup.size === 0) return Promise.resolve([]);
if (this.#cleanupMarkers.size === 0) return Promise.resolve([]);
if (this.#cleanupAttempt !== undefined) return this.#cleanupAttempt;
this.#cleanupAttempt = this.#retryCleanup().finally(() => {
this.#cleanupAttempt = undefined;
@@ -187,9 +204,7 @@ export class XcrunCommandRunner implements NativeCommandRunner {
this.launcher,
);
if (!resolved.ok) {
if (resolved.error.cleanupOwner !== undefined)
this.#pendingCleanup.add(resolved.error.cleanupOwner);
return resolved;
return err(this.#retainCleanupOwner(resolved.error));
}
const captured = await captureProcess(
resolved.value.path,
@@ -199,8 +214,7 @@ export class XcrunCommandRunner implements NativeCommandRunner {
this.environment,
this.launcher,
);
if (!captured.ok && captured.error.cleanupOwner !== undefined)
this.#pendingCleanup.add(captured.error.cleanupOwner);
if (!captured.ok) return err(this.#retainCleanupOwner(captured.error));
return captured.ok
? ok({
...captured.value,
@@ -217,15 +231,19 @@ export class XcrunCommandRunner implements NativeCommandRunner {
async #retryCleanup(): Promise<readonly NativeCommandCleanupFailure[]> {
const failures: NativeCommandCleanupFailure[] = [];
for (const owner of this.#pendingCleanup) {
for (const [owner, marker] of this.#cleanupMarkers) {
let reason: string | undefined;
try {
const stopped = await owner.stop();
if (stopped.status !== "incomplete") {
this.#pendingCleanup.delete(owner);
if (stopped.status === "not-owned") {
reason = "native command ownership remains unresolved";
} else if (stopped.status !== "incomplete") {
marker.resolveCleanup();
this.#cleanupMarkers.delete(owner);
continue;
}
reason = stopped.reason || "native command cleanup remains unverified";
} else
reason =
stopped.reason || "native command cleanup remains unverified";
} catch (cause: unknown) {
reason =
cause instanceof Error
@@ -236,10 +254,49 @@ export class XcrunCommandRunner implements NativeCommandRunner {
owner,
reason: reason ?? "native command cleanup remains unverified",
resources: commandProcessResources(owner),
marker,
});
}
return failures;
}
#retainCleanupOwner(failure: NativeCommandFailure): NativeCommandFailure {
const owner = failure.cleanupOwner;
if (owner === undefined) return failure;
let marker = failure.cleanupMarker ?? this.#cleanupMarkers.get(owner);
if (marker === undefined) {
marker = new ProviderCleanupError(
"native-macos",
commandProcessResources(owner),
{
reason:
failure.cleanupFailure ??
"native command cleanup remains incomplete",
},
{ operation: "native_command", cause: failure },
);
}
this.#cleanupMarkers.set(owner, marker);
if (failure.cleanupMarker === marker) return failure;
return new NativeCommandFailure(
failure.tool,
failure.reason,
failure.exitCode,
{
cause: failure,
...(failure.cancellation === undefined
? {}
: { cancellation: failure.cancellation }),
cleanupOwner: owner,
cleanupFailures: failure.cleanupFailures,
cleanupMarker: marker,
...(failure.detail === null ? {} : { detail: failure.detail }),
},
failure.capture,
failure.cleanupFailure,
failure.cleanupResources,
);
}
}
const commandProcessResources = (
@@ -332,6 +389,12 @@ export const resolveXcrunTool = async (
...(found.error.cleanupFailures.length === 0
? {}
: { cleanupFailures: found.error.cleanupFailures }),
...(found.error.cleanupMarker === undefined
? {}
: { cleanupMarker: found.error.cleanupMarker }),
...(found.error.cancellation === undefined
? {}
: { cancellation: found.error.cancellation }),
...(found.error.detail === null
? {}
: { detail: found.error.detail }),
@@ -368,22 +431,23 @@ export const resolveXcrunTool = async (
return ok({ path, sha256: await hashFile(path, signal) });
} catch (cause: unknown) {
const capture = failureCapture(found.value, false);
if (signal?.aborted === true)
return err(
new NativeCommandFailure(
tool,
"cancelled",
capture.exitCode,
{ cause },
capture,
),
);
const cancellation =
signal?.aborted === true ? { reason: signal.reason } : undefined;
const abortedHashRead =
cause instanceof Error &&
"code" in cause &&
cause.code === "ABORT_ERR" &&
signal?.aborted === true &&
cause.cause === signal.reason;
return err(
new NativeCommandFailure(
tool,
"io",
abortedHashRead ? "cancelled" : "io",
capture.exitCode,
{ cause },
{
cause,
...(cancellation === undefined ? {} : { cancellation }),
},
capture,
),
);
@@ -476,6 +540,9 @@ const captureProcess = async (
capture?.exitCode ?? null,
{
cause,
...(cause.cancellation === undefined
? {}
: { cancellation: cause.cancellation }),
...(cause.cleanupOwner === undefined
? {}
: { cleanupOwner: cause.cleanupOwner }),
+82 -34
View File
@@ -958,8 +958,8 @@ class NativeMacOSClient implements AnalysisClient {
const nativeCommandCleanupFailure = (
failures: readonly NativeCommandCleanupFailure[],
): ProviderCleanupError =>
new ProviderCleanupError(
): ProviderCleanupError => {
const failure = new ProviderCleanupError(
IDENTITY.id,
[...new Set(failures.flatMap(({ resources }) => resources))],
{
@@ -970,6 +970,18 @@ const nativeCommandCleanupFailure = (
})),
},
);
if (
failures.length === 0 ||
failures.some(({ marker }) => marker === undefined)
)
return failure;
const markers = failures.flatMap(({ marker }) =>
marker === undefined ? [] : [marker],
);
return markers.length === 0
? failure
: failure.bindCleanupDependencies(markers);
};
/** Each file workflow retains its own exact resource set through client close. */
const closeNativeFileScope = async (
@@ -1021,6 +1033,29 @@ const translateCommandFailure = (
operation: NativeToolName,
failure: NativeCommandFailure,
): AnalysisError => {
const cleanupOwners = [
...failure.cleanupFailures,
...(failure.cleanupOwner === undefined
? []
: [
{
owner: failure.cleanupOwner,
marker: failure.cleanupMarker,
reason:
failure.cleanupFailure ??
"native command cleanup remains incomplete",
resources: failure.cleanupResources,
},
]),
];
const markers = cleanupOwners.flatMap(({ marker }) =>
marker === undefined ? [] : [marker],
);
const bindCleanup = (error: AnalysisError): AnalysisError =>
cleanupOwners.length === 0 ||
cleanupOwners.some(({ marker }) => marker === undefined)
? error
: error.bindCleanupDependencies([...new Set(markers)]);
const preparation = failure.preparationFailure;
if (preparation !== undefined) {
const marker = preparation.cleanupFailure;
@@ -1069,9 +1104,9 @@ const translateCommandFailure = (
...options,
cause: preparation.cancellation.reason,
});
return marker === undefined
? error
: error.bindCleanupDependencies([marker]);
return bindCleanup(
marker === undefined ? error : error.bindCleanupDependencies([marker]),
);
}
const capturedOutput =
failure.capture === null
@@ -1098,42 +1133,55 @@ const translateCommandFailure = (
};
if (failure.reason === "unavailable") {
const reason = `${failure.tool} is unavailable through xcrun.`;
return new AnalysisCapabilityUnavailableError(
IDENTITY.id,
operation,
reason,
{ ...errorOptions, userMessage: reason },
return bindCleanup(
new AnalysisCapabilityUnavailableError(IDENTITY.id, operation, reason, {
...errorOptions,
userMessage: reason,
}),
);
}
if (failure.reason === "cancelled")
return new AnalysisCancelledError(operation, errorOptions);
return bindCleanup(
new AnalysisCancelledError(operation, {
...errorOptions,
...(failure.cancellation === undefined
? {}
: { cause: failure.cancellation.reason }),
}),
);
if (failure.reason === "timeout")
return new AnalysisTimeoutError(
operation,
NATIVE_COMMAND_TIMEOUT_MS,
errorOptions,
return bindCleanup(
new AnalysisTimeoutError(
operation,
NATIVE_COMMAND_TIMEOUT_MS,
errorOptions,
),
);
if (failure.reason === "output-limit")
return new AnalysisResourceConstraintError(
operation,
"memory",
`Native command output exceeded ${String(NATIVE_COMMAND_OUTPUT_BUDGET_BYTES)} bytes; complete output is required for analysis.`,
{ max_output_bytes: NATIVE_COMMAND_OUTPUT_BUDGET_BYTES },
errorOptions,
return bindCleanup(
new AnalysisResourceConstraintError(
operation,
"memory",
`Native command output exceeded ${String(NATIVE_COMMAND_OUTPUT_BUDGET_BYTES)} bytes; complete output is required for analysis.`,
{ max_output_bytes: NATIVE_COMMAND_OUTPUT_BUDGET_BYTES },
errorOptions,
),
);
return new ProviderAdapterError(IDENTITY.id, operation, {
cause: failure,
...errorOptions,
diagnostics: {
tool: failure.tool,
reason: failure.reason,
exit_code: failure.capture?.exitCode ?? failure.exitCode,
signal: failure.capture?.signal ?? null,
stdout_bytes: failure.capture?.stdoutBytes ?? 0,
stderr_bytes: failure.capture?.stderrBytes ?? 0,
...(failure.detail === null ? {} : { detail: failure.detail }),
},
});
return bindCleanup(
new ProviderAdapterError(IDENTITY.id, operation, {
cause: failure,
...errorOptions,
diagnostics: {
tool: failure.tool,
reason: failure.reason,
exit_code: failure.capture?.exitCode ?? failure.exitCode,
signal: failure.capture?.signal ?? null,
stdout_bytes: failure.capture?.stdoutBytes ?? 0,
stderr_bytes: failure.capture?.stderrBytes ?? 0,
...(failure.detail === null ? {} : { detail: failure.detail }),
},
}),
);
};
const translateCodeSignExitFailure = (
+77 -27
View File
@@ -219,8 +219,16 @@ const collectOwnedCommand = async (
throwCommandPreparationFailure(cause, signal, deadline.interruption);
}
let exceeded = false;
let outputLimitObservedBeforeInterruption = false;
let processFailure: string | undefined;
let processError: Error | undefined;
let processFailureBeforeInterruption:
| {
readonly message: string;
readonly processError: Error | undefined;
readonly collectionFailure: boolean;
}
| undefined;
const supervisor = new ProviderProcessSupervisor(
{
...launched,
@@ -233,11 +241,37 @@ const collectOwnedCommand = async (
captureStdout: limits.stdout === undefined,
maxDiagnosticBytes: limits.diagnosticBytes,
onDiagnostic: (event) => {
if (event.type === "output" && event.truncated === true)
if (event.type === "output" && event.truncated === true) {
exceeded = true;
if (deadline.interruption === undefined)
outputLimitObservedBeforeInterruption = true;
}
if (event.type === "error") {
processFailure ??= event.message;
processError ??= event.cause;
if (deadline.interruption === undefined)
processFailureBeforeInterruption ??= {
message: processFailure,
processError,
collectionFailure: true,
};
}
if (event.type === "exit" && deadline.interruption === undefined) {
const acceptedNonzeroExit =
limits.acceptNonZeroExit === true &&
event.code !== null &&
event.code !== 0 &&
event.signal === null;
if (
event.signal !== null ||
(event.code !== null && event.code !== 0 && !acceptedNonzeroExit)
)
processFailureBeforeInterruption ??= {
message:
"Command failed; inspect its exit status and diagnostics.",
processError: undefined,
collectionFailure: false,
};
}
},
},
@@ -251,23 +285,51 @@ const collectOwnedCommand = async (
);
void consumed.then((failed) => {
processFailure ??= failed;
if (failed !== undefined && deadline.interruption === undefined)
processFailureBeforeInterruption ??= {
message: processFailure ?? failed,
processError,
collectionFailure: true,
};
});
let failure: OwnedCommandFailure | undefined;
const throwInterruptionOrObservedFailure = (
interruption: "cancelled" | "timeout" | undefined,
snapshot: ProviderProcessSnapshot | null = null,
): void => {
if (interruption === undefined) return;
if (outputLimitObservedBeforeInterruption)
throw new OwnedCommandFailure(
"output-limit",
"Command diagnostic output exceeded its complete-output budget.",
snapshot,
);
if (processFailureBeforeInterruption !== undefined)
throw new OwnedCommandFailure(
"process",
processFailureBeforeInterruption.message,
snapshot,
null,
{
processError: processFailureBeforeInterruption.processError,
collectionFailure: processFailureBeforeInterruption.collectionFailure,
},
);
throw new OwnedCommandFailure(
interruption,
interruption === "cancelled"
? "Command cancelled."
: "Command deadline elapsed.",
snapshot,
null,
interruption === "cancelled"
? { cancellation: { reason: signal.reason } }
: undefined,
);
};
try {
while (!(await supervisor.waitForOutputClose(10))) {
const interruption = deadline.interruption;
if (interruption !== undefined)
throw new OwnedCommandFailure(
interruption,
interruption === "cancelled"
? "Command cancelled."
: "Command deadline elapsed.",
null,
null,
interruption === "cancelled"
? { cancellation: { reason: signal.reason } }
: undefined,
);
throwInterruptionOrObservedFailure(deadline.interruption);
if (exceeded)
throw new OwnedCommandFailure(
"output-limit",
@@ -284,19 +346,7 @@ const collectOwnedCommand = async (
const sinkFailure = await waitForAbortable(consumed, signal);
if (sinkFailure !== ABORTED) processFailure ??= sinkFailure;
const snapshot = supervisor.snapshot();
const interruption = deadline.interruption;
if (interruption !== undefined)
throw new OwnedCommandFailure(
interruption,
interruption === "cancelled"
? "Command cancelled."
: "Command deadline elapsed.",
snapshot,
null,
interruption === "cancelled"
? { cancellation: { reason: signal.reason } }
: undefined,
);
throwInterruptionOrObservedFailure(deadline.interruption, snapshot);
if (exceeded)
throw new OwnedCommandFailure(
"output-limit",
+124 -1
View File
@@ -9,7 +9,10 @@ import {
OwnedCommandFailure,
runOwnedCommand,
} from "../../../src/process/OwnedCommand.js";
import { spawnOwnedProviderProcess } from "../../../src/process/ProviderProcess.js";
import {
ProviderProcessSupervisor,
spawnOwnedProviderProcess,
} from "../../../src/process/ProviderProcess.js";
import {
cleanupOwnedProcessGroup,
type ProcessCleanupResult,
@@ -19,6 +22,7 @@ import {
systemProcessOwnershipHost,
} from "../../../src/process/ProcessOwnershipObservation.js";
import { waitForProviderProcessReady } from "../../fixtures/providerProcess.js";
import { createDeferred } from "../../fixtures/binarySession.js";
const command = (script: string) => ({
command: process.execPath,
@@ -196,6 +200,125 @@ it("retains late cancellation separately from a physical failure with refused cl
}
});
it("keeps an exit failure observed before cancellation primary across delayed classification", async () => {
const controller = new AbortController();
const callerReason = { request: "cancel after process exit" };
let denied = true;
let retainedOwner: ProviderProcessSupervisor | undefined;
const outputClosed = createDeferred<void>();
const continueClassification = createDeferred<void>();
const sink = new Writable({
write(_chunk, _encoding, callback) {
callback();
},
final(_callback) {},
});
const originalWait = ProviderProcessSupervisor.prototype.waitForOutputClose;
const waitForOutputClose = vi.spyOn(
ProviderProcessSupervisor.prototype,
"waitForOutputClose",
);
let delayed = false;
waitForOutputClose.mockImplementation(async function (
this: ProviderProcessSupervisor,
timeoutMs?: number,
signal?: AbortSignal,
) {
const closed = await originalWait.call(this, timeoutMs, signal);
if (closed && !delayed) {
delayed = true;
outputClosed.resolve(undefined);
await continueClassification.promise;
}
return closed;
});
try {
const result = runOwnedCommand(
command(
'process.stdout.write("buffered"); process.stderr.write("authored physical failure"); process.exitCode = 7',
),
{ timeoutMs: 2_000, diagnosticBytes: 1_024, stdout: sink },
{
signal: controller.signal,
retainCleanup: (owner) => {
retainedOwner = owner;
},
launcher: async (input) => {
const acquired = await spawnOwnedProviderProcess(input);
return {
...acquired,
cleanup: async () => {
if (denied)
return { cleaned: false, reason: "controlled cleanup refusal" };
return cleanupOwnedProcessGroup(acquired.ownership);
},
};
},
},
);
await outputClosed.promise;
controller.abort(callerReason);
continueClassification.resolve(undefined);
const failure = await result.catch((cause: unknown) => {
if (!(cause instanceof OwnedCommandFailure)) throw cause;
return cause;
});
expect(failure).toMatchObject({
reason: "process",
snapshot: {
exitCode: 7,
stderr: { text: "authored physical failure" },
},
});
expect(failure).toBeInstanceOf(OwnedCommandFailure);
if (!(failure instanceof OwnedCommandFailure)) return;
expect(failure.cancellation?.reason).toBe(callerReason);
expect(failure.collectionFailure).toBe(false);
expect(failure.processError).toBeUndefined();
expect(sink.destroyed).toBe(true);
expect(retainedOwner).toBeDefined();
denied = false;
expect(await retainedOwner?.stop()).toEqual({
status: "verified-cleanup",
});
} finally {
continueClassification.resolve(undefined);
waitForOutputClose.mockRestore();
denied = false;
await retainedOwner?.stop();
}
});
it("keeps cancellation primary when the owned child exits during cancellation cleanup", async () => {
const controller = new AbortController();
const callerReason = { request: "cancel before process exit" };
const outcome = runOwnedCommand(
command('process.stdout.write("ready"); setInterval(() => {}, 1_000)'),
{ timeoutMs: 2_000, diagnosticBytes: 1_024 },
{
signal: controller.signal,
launcher: async (input) => {
const acquired = await spawnOwnedProviderProcess(input);
acquired.process.stdout?.once("data", () =>
controller.abort(callerReason),
);
return acquired;
},
},
);
const failure = await outcome.catch((cause: unknown) => {
if (!(cause instanceof OwnedCommandFailure)) throw cause;
return cause;
});
expect(failure).toBeInstanceOf(OwnedCommandFailure);
if (!(failure instanceof OwnedCommandFailure)) return;
expect(failure.reason).toBe("cancelled");
expect(failure.cancellation?.reason).toBe(callerReason);
});
it("streams stdout beyond the diagnostic budget into a sink and returns after it finishes", async () => {
let received = 0;
let finished = false;
@@ -0,0 +1,120 @@
import { EventEmitter } from "node:events";
import { PassThrough } from "node:stream";
import { afterEach, expect, it, vi } from "vitest";
import {
resolveXcrunTool,
type NativeCommandLauncher,
} from "../../../../src/native/CommandRunner.js";
const realpathState = vi.hoisted(() => ({
resolve: undefined as ((path: string) => Promise<string>) | undefined,
}));
vi.mock("node:fs/promises", async (importOriginal) => {
const actual = await importOriginal<typeof import("node:fs/promises")>();
return {
...actual,
realpath: (path: string): Promise<string> => {
if (realpathState.resolve === undefined)
throw new Error("realpath fixture was not configured");
return realpathState.resolve(path);
},
};
});
const launcher: NativeCommandLauncher = async (options) => {
const child = Object.assign(new EventEmitter(), {
stdout: new PassThrough(),
stderr: new PassThrough(),
exitCode: null as number | null,
signalCode: null as NodeJS.Signals | null,
kill: () => true,
});
setImmediate(() => {
child.stdout.end("/controlled/native-tool\n");
child.stderr.end();
child.exitCode = 0;
child.emit("exit", 0, null);
child.emit("close", 0, null);
});
return {
process: child,
ownership: {
runId: options.runId,
leaderPid: 48_100,
processGroupId: 48_100,
expectedParentPid: process.pid,
},
cleanup: async () => ({ cleaned: true, signaled: false }),
};
};
afterEach(() => {
realpathState.resolve = undefined;
});
it("keeps a realpath I/O failure primary when cancellation is observed before rejection", async () => {
let realpathStarted!: () => void;
const started = new Promise<void>((resolve) => {
realpathStarted = resolve;
});
let rejectRealpath!: (cause: unknown) => void;
realpathState.resolve = () =>
new Promise<string>((_resolve, reject) => {
rejectRealpath = reject;
realpathStarted();
});
const controller = new AbortController();
const cancellationReason = { source: "caller", phase: "tool resolution" };
const pending = resolveXcrunTool("file", {}, controller.signal, launcher);
await started;
controller.abort(cancellationReason);
const ioFailure = Object.assign(new Error("fixture realpath failure"), {
code: "EACCES",
});
rejectRealpath(ioFailure);
const result = await pending;
if (result.ok) throw new Error("Expected realpath I/O failure");
expect(result.error).toMatchObject({
reason: "io",
cancellation: { reason: cancellationReason },
});
expect(result.error.cause).toBe(ioFailure);
});
it.skipIf(process.platform === "win32")(
"uses the AbortError cause as the exact tool-hash cancellation fact",
async () => {
let finishRealpath!: () => void;
let realpathStarted!: () => void;
const started = new Promise<void>((resolve) => {
realpathStarted = resolve;
});
realpathState.resolve = () =>
new Promise<string>((resolve) => {
finishRealpath = () => resolve("/dev/zero");
realpathStarted();
});
const controller = new AbortController();
const cancellationReason = { source: "caller", phase: "tool hash" };
const pending = resolveXcrunTool("file", {}, controller.signal, launcher);
await started;
finishRealpath();
setImmediate(() => controller.abort(cancellationReason));
const result = await pending;
if (result.ok) throw new Error("Expected hash cancellation");
expect(result.error).toMatchObject({
reason: "cancelled",
cancellation: { reason: cancellationReason },
});
expect(result.error.cause).toMatchObject({
name: "AbortError",
code: "ABORT_ERR",
cause: cancellationReason,
});
},
);
@@ -16,6 +16,7 @@ import {
NATIVE_FAILURE_DIAGNOSTIC_BYTES,
NativeCommandFailure,
XcrunCommandRunner,
type NativeCommandLauncher,
type NativeCommandRunner,
} from "../../../../src/native/CommandRunner.js";
import { err, ok } from "../../../../src/domain/result.js";
@@ -24,7 +25,13 @@ import { parseCodeSignature } from "../../../../src/native/parsers/codesign.js";
import { prepareProcessOwnershipInspection } from "../../../../src/process/ProcessOwnershipObservation.js";
import { projectAnalysisError } from "../../../../src/domain/analysisErrorProjection.js";
import { ProviderCleanupError } from "../../../../src/domain/providerCleanupError.js";
import { ProviderProcessSupervisor } from "../../../../src/process/ProviderProcess.js";
import { cleanupOwnedProcessGroup } from "../../../../src/process/ProcessOwnership.js";
import {
ProviderProcessSupervisor,
spawnOwnedProviderProcess,
type ProviderProcessHandle,
type SpawnedOwnedProviderProcess,
} from "../../../../src/process/ProviderProcess.js";
import {
NativeFixtureRunner as FixtureRunner,
@@ -388,7 +395,479 @@ describe("native command failure capture and limits", () => {
},
});
});
});
describe("native owned cancellation boundary", () => {
it.skipIf(process.platform === "win32")(
"projects an owned Node command cancellation with its exact reason and captured output",
async () => {
directory = await createTestTempDirectory("rea-native-cancel-boundary-");
const ready = join(directory, "ready");
const nodeSource = [
'process.stdout.write("native output before cancellation", () => {',
'process.stderr.write("native diagnostic before cancellation", () => {',
`require("node:fs").writeFileSync(${JSON.stringify(ready)}, "ready");`,
"}); });",
"setInterval(() => {}, 1000);",
].join("\n");
const executable = join(directory, "native-tool");
await writeFile(executable, `#!${process.execPath}\n${nodeSource}\n`, {
mode: 0o700,
});
await chmod(executable, 0o700);
const executableSha256 = createHash("sha256")
.update(await readFile(executable))
.digest("hex");
const launcher: NativeCommandLauncher = (options) =>
spawnOwnedProviderProcess({ ...options, expectedCommand: null });
const runner = new XcrunCommandRunner(
{},
async () => ok({ path: executable, sha256: executableSha256 }),
launcher,
);
const client = new NativeMacOSProvider({}, runner, "darwin").createClient(
machoTarget("/private/fixture"),
);
const controller = new AbortController();
const cancellationReason = {
source: "test caller",
operation: "native command cancellation",
};
try {
const pending = client.execute(
"list_architectures",
{},
{ signal: controller.signal },
);
const deadline = Date.now() + 5_000;
let readyObserved = false;
while (Date.now() < deadline) {
try {
readyObserved = (await readFile(ready, "utf8")) === "ready";
} catch (cause: unknown) {
if (
!(
cause instanceof Error &&
"code" in cause &&
cause.code === "ENOENT"
)
)
throw cause;
}
if (readyObserved) break;
await new Promise<void>((resolve) => setTimeout(resolve, 10));
}
expect(
readyObserved,
"the owned Node process reached its output barrier",
).toBe(true);
controller.abort(cancellationReason);
const result = await pending;
if (result.ok) throw new Error("Expected owned command cancellation");
expect(result.error._tag).toBe("AnalysisCancelledError");
expect(result.error.cause).toBe(cancellationReason);
expect(result.error.capturedOutput).toMatchObject({
stdout: "native output before cancellation",
stderr: "native diagnostic before cancellation",
truncated: false,
});
expect(result.error.cleanupIncomplete).toBe(false);
await expect(client.close()).resolves.toMatchObject({ ok: true });
} finally {
await client.close();
}
},
);
});
describe("native cancellation cleanup provenance", () => {
it.skipIf(process.platform === "win32")(
"resolves only the owners that succeed when cleanup retries cover multiple native commands",
async () => {
directory = await createTestTempDirectory("rea-native-mixed-cleanup-");
const executable = join(directory, "native-tool");
const nodeSource =
'require("node:fs").writeFileSync(require("node:path").join(' +
JSON.stringify(directory) +
', process.env.REA_PROCESS_RUN_ID), "ready"); setInterval(() => {}, 1000);\n';
await writeFile(executable, "#!" + process.execPath + "\n" + nodeSource, {
mode: 0o700,
});
await chmod(executable, 0o700);
const executableSha256 = createHash("sha256")
.update(await readFile(executable))
.digest("hex");
const owners: Array<{
readonly runId: string;
attempts: number;
cleanupVerified: boolean;
process: SpawnedOwnedProviderProcess["process"] | undefined;
}> = [];
const launcher: NativeCommandLauncher = async (options) => {
const owner: (typeof owners)[number] = {
runId: options.runId,
attempts: 0,
cleanupVerified: false,
process: undefined,
};
owners.push(owner);
const launched = await spawnOwnedProviderProcess({
...options,
expectedCommand: null,
});
owner.process = launched.process;
return {
...launched,
cleanup: async () => {
owner.attempts += 1;
if (
owner.attempts === 1 ||
(owner === owners[1] && owner.attempts === 2)
)
return {
cleaned: false,
reason: `controlled owner ${owner.runId} refusal ${owner.attempts}`,
};
const cleanup = await cleanupOwnedProcessGroup(launched.ownership);
owner.cleanupVerified = cleanup.cleaned;
return cleanup;
},
};
};
const runner = new XcrunCommandRunner(
{},
async () => ok({ path: executable, sha256: executableSha256 }),
launcher,
);
const client = new NativeMacOSProvider({}, runner, "darwin").createClient(
machoTarget("/private/fixture"),
);
const controllers = [new AbortController(), new AbortController()];
const cancellationReason = { source: "mixed native owner retry" };
try {
const pending = controllers.map((controller) =>
client.execute(
"list_architectures",
{},
{ signal: controller.signal },
),
);
const deadline = Date.now() + 5_000;
let ready = false;
while (Date.now() < deadline && !ready) {
if (owners.length === 2) {
const markers = await Promise.all(
owners.map(async ({ runId }) => {
try {
return (
(await readFile(join(directory!, runId), "utf8")) ===
"ready"
);
} catch (cause: unknown) {
if (
!(
cause instanceof Error &&
"code" in cause &&
cause.code === "ENOENT"
)
)
throw cause;
return false;
}
}),
);
ready = markers.every(Boolean);
}
if (!ready)
await new Promise<void>((resolve) => setTimeout(resolve, 10));
}
expect(
ready,
"both owned processes reached their filesystem barriers",
).toBe(true);
for (const controller of controllers)
controller.abort(cancellationReason);
const results = await Promise.all(pending);
const failures = results.map((result) => {
if (result.ok) throw new Error("Expected command cancellation");
expect(result.error.cleanupIncomplete).toBe(true);
return result.error;
});
const failuresByOwner = owners.map((owner) => {
const failure = failures.find((error) =>
error.cleanupResources.includes(owner.runId),
);
if (failure === undefined)
throw new Error("Missing error for retained native owner");
return failure;
});
await runner.retryCleanup?.();
expect(failuresByOwner[0]?.cleanupIncomplete).toBe(false);
expect(failuresByOwner[0]?.cleanupResources).toEqual([]);
expect(owners[0]?.cleanupVerified).toBe(true);
expect(
owners[0]?.process !== undefined &&
(owners[0].process.exitCode !== null ||
owners[0].process.signalCode !== null),
).toBe(true);
expect(failuresByOwner[1]?.cleanupIncomplete).toBe(true);
expect(failuresByOwner[1]?.cleanupResources.length).toBeGreaterThan(0);
await runner.retryCleanup?.();
expect(failuresByOwner[1]?.cleanupIncomplete).toBe(false);
expect(failuresByOwner[1]?.cleanupResources).toEqual([]);
expect(owners[1]?.cleanupVerified).toBe(true);
expect(
owners[1]?.process !== undefined &&
(owners[1].process.exitCode !== null ||
owners[1].process.signalCode !== null),
).toBe(true);
} finally {
for (const { process: child } of owners)
if (child?.exitCode === null) child.kill("SIGKILL");
await client.close();
}
},
);
});
describe("native returned cleanup ownership", () => {
it.skipIf(process.platform === "win32")(
"resolves a returned command error when its exact retained owner later stops",
async () => {
directory = await createTestTempDirectory("rea-native-cleanup-retry-");
const ready = join(directory, "ready");
const executable = join(directory, "native-tool");
await writeFile(
executable,
`#!${process.execPath}\nrequire("node:fs").writeFileSync(${JSON.stringify(ready)}, "ready"); setInterval(() => {}, 1000);\n`,
{ mode: 0o700 },
);
await chmod(executable, 0o700);
const executableSha256 = createHash("sha256")
.update(await readFile(executable))
.digest("hex");
let cleanupCalls = 0;
let verifiedCleanup = false;
let ownedProcess: ProviderProcessHandle | undefined;
const launcher: NativeCommandLauncher = async (options) => {
const launched = await spawnOwnedProviderProcess({
...options,
expectedCommand: null,
});
ownedProcess = launched.process;
return {
...launched,
cleanup: async () => {
cleanupCalls += 1;
if (cleanupCalls < 3)
return {
cleaned: false,
reason: "controlled first cleanup refusal",
};
const cleanup = await cleanupOwnedProcessGroup(launched.ownership);
verifiedCleanup = cleanup.cleaned;
return cleanup;
},
};
};
const runner = new XcrunCommandRunner(
{},
async () => ok({ path: executable, sha256: executableSha256 }),
launcher,
);
const client = new NativeMacOSProvider({}, runner, "darwin").createClient(
machoTarget("/private/fixture"),
);
const controller = new AbortController();
const cancellationReason = { source: "native cleanup retry regression" };
try {
const pending = client.execute(
"list_architectures",
{},
{ signal: controller.signal },
);
const deadline = Date.now() + 5_000;
let readyObserved = false;
while (Date.now() < deadline) {
try {
readyObserved = (await readFile(ready, "utf8")) === "ready";
} catch (cause: unknown) {
if (
!(
cause instanceof Error &&
"code" in cause &&
cause.code === "ENOENT"
)
)
throw cause;
}
if (readyObserved) break;
await new Promise<void>((resolve) => setTimeout(resolve, 10));
}
expect(
readyObserved,
"owned Node process reached filesystem barrier",
).toBe(true);
controller.abort(cancellationReason);
const original = await pending;
if (original.ok) throw new Error("Expected command cancellation");
expect(original.error.cause).toBe(cancellationReason);
expect(original.error.cleanupIncomplete).toBe(true);
expect(
original.error.cleanupResources.some((resource) =>
resource.startsWith("process-group:"),
),
).toBe(true);
expect(cleanupCalls).toBe(1);
// Client close retries the same retained owner and returns a live
// aggregate marker when that retry is still incomplete.
const closed = await client.close();
expect(cleanupCalls).toBe(2);
expect(closed.ok).toBe(false);
if (closed.ok) throw new Error("Expected cleanup to remain incomplete");
expect(closed.error.cleanupIncomplete).toBe(true);
expect(closed.error.cleanupResources).toEqual(
original.error.cleanupResources,
);
await runner.retryCleanup?.();
expect(cleanupCalls).toBe(3);
expect(verifiedCleanup).toBe(true);
expect(
ownedProcess !== undefined &&
(ownedProcess.exitCode !== null ||
ownedProcess.signalCode !== null),
).toBe(true);
expect(original.error.cleanupIncomplete).toBe(false);
expect(original.error.cleanupResources).toEqual([]);
expect(closed.error.cleanupIncomplete).toBe(false);
expect(closed.error.cleanupResources).toEqual([]);
await expect(client.close()).resolves.toMatchObject({ ok: true });
} finally {
if (ownedProcess !== undefined && ownedProcess.exitCode === null)
ownedProcess.kill("SIGKILL");
await client.close();
}
},
);
});
describe("native late cancellation with cleanup", () => {
it("keeps a late cancellation fact with the physical command failure and retries its retained owner", async () => {
const cancellationReason = { caller: "stop after process failure" };
const controller = new AbortController();
let commandStarted!: () => void;
const started = new Promise<void>((resolve) => {
commandStarted = resolve;
});
let cleanupStarted!: () => void;
const cleanupAttempt = new Promise<void>((resolve) => {
cleanupStarted = resolve;
});
let releaseFirstCleanup!: () => void;
const firstCleanup = new Promise<void>((resolve) => {
releaseFirstCleanup = resolve;
});
let cleanupCalls = 0;
let launches = 0;
const launcher: NativeCommandLauncher = async (options) => {
launches += 1;
const child = Object.assign(new EventEmitter(), {
stdout: new PassThrough(),
stderr: new PassThrough(),
exitCode: null as number | null,
signalCode: null as NodeJS.Signals | null,
kill: () => true,
});
setImmediate(() => {
child.stdout.write("partial native output");
child.stderr.write("native diagnostic");
child.stdout.end();
child.stderr.end();
child.exitCode = 7;
child.emit("exit", 7, null);
child.emit("close", 7, null);
commandStarted();
});
return {
process: child,
ownership: {
runId: options.runId,
leaderPid: 47_100 + launches,
processGroupId: 47_100 + launches,
expectedParentPid: process.pid,
},
cleanup: async () => {
cleanupCalls += 1;
if (cleanupCalls === 1) {
cleanupStarted();
await firstCleanup;
return {
cleaned: false,
reason: "controlled cleanup remains incomplete",
};
}
return { cleaned: true, signaled: false };
},
};
};
const runner = new XcrunCommandRunner(
{},
async () =>
ok({ path: "/controlled/native-tool", sha256: "a".repeat(64) }),
launcher,
);
const client = new NativeMacOSProvider({}, runner, "darwin").createClient(
machoTarget("/private/fixture"),
);
try {
const pending = client.execute(
"list_architectures",
{},
{ signal: controller.signal },
);
await started;
await cleanupAttempt;
controller.abort(cancellationReason);
releaseFirstCleanup();
const result = await pending;
if (result.ok)
throw new Error("Expected the process failure to remain primary");
expect(result.error._tag).toBe("ProviderAdapterError");
expect(result.error.cause).toMatchObject({
reason: "nonzero-exit",
capture: {
stdout: "partial native output",
stderr: "native diagnostic",
exitCode: 7,
},
cleanupFailure: "controlled cleanup remains incomplete",
});
expect(result.error.cause).toMatchObject({
cancellation: { reason: cancellationReason },
});
expect(result.error.cause).toHaveProperty("cleanupOwner");
expect(result.error.cleanup).toMatchObject({
reason: "controlled cleanup remains incomplete",
});
expect(cleanupCalls).toBe(1);
expect(launches).toBe(1);
await expect(client.close()).resolves.toMatchObject({ ok: true });
expect(cleanupCalls).toBe(2);
expect(launches).toBe(1);
} finally {
releaseFirstCleanup();
await client.close();
}
});
});
describe("native command failure capture and limits", () => {
it("bounds both failure streams at a UTF-8 boundary while keeping their totals", async () => {
const runner = new XcrunCommandRunner({}, () =>
Promise.resolve(ok({ path: process.execPath, sha256: "a".repeat(64) })),
@@ -453,6 +932,49 @@ describe("native command failure capture and limits", () => {
});
describe("native command cleanup lifecycle", () => {
it("keeps a not-owned retry unresolved instead of discarding its marker", async () => {
const child = Object.assign(new EventEmitter(), {
stdout: new PassThrough(),
stderr: new PassThrough(),
exitCode: null as number | null,
signalCode: null as NodeJS.Signals | null,
kill: () => true,
});
const owner = new ProviderProcessSupervisor({
process: child,
ownsProcessLifetime: false,
});
let resolveCalls = 0;
const runner = new XcrunCommandRunner({}, async (tool) => {
resolveCalls += 1;
return err(
new NativeCommandFailure(
tool,
"io",
null,
{ cleanupOwner: owner },
null,
"initial cleanup remains unresolved",
["run:unowned"],
),
);
});
const initial = await runner.run("file", [], {});
if (initial.ok) throw new Error("Expected resolver failure");
expect(initial.error.cleanupMarker?.cleanupIncomplete).toBe(true);
const blocked = await runner.run("file", [], {});
if (blocked.ok)
throw new Error("Expected unresolved cleanup to block launch");
expect(blocked.error.cleanupFailures).toHaveLength(1);
expect(blocked.error.cleanupFailures[0]?.reason).toBe(
"native command ownership remains unresolved",
);
expect(await runner.retryCleanup()).toHaveLength(1);
expect(initial.error.cleanupMarker?.cleanupIncomplete).toBe(true);
expect(resolveCalls).toBe(1);
});
it("waits for an active command before retrying its failed owner at close", async () => {
let finishCommand!: (
result: Awaited<ReturnType<NativeCommandRunner["run"]>>,
@@ -544,7 +1066,67 @@ describe("native command cleanup lifecycle", () => {
});
describe("native command failure projection", () => {
it("does not clear mixed cleanup facts when one injected owner has no marker", async () => {
const marker = new ProviderCleanupError("native-macos", ["run:known"], {
reason: "known owner remains incomplete",
});
const failures = [
{
owner: cleanupOwner("run:known"),
reason: "known owner remains incomplete",
resources: ["run:known"],
marker,
},
{
owner: cleanupOwner("run:unknown"),
reason: "unknown owner remains incomplete",
resources: ["run:unknown"],
},
];
const runner: NativeCommandRunner = {
run: async (tool) =>
err(
new NativeCommandFailure(
tool,
"io",
null,
{ cleanupFailures: failures },
null,
"mixed cleanup remains incomplete",
["run:known", "run:unknown"],
),
),
retryCleanup: async () => failures,
};
const client = new NativeMacOSProvider({}, runner, "darwin").createClient(
machoTarget("/private/fixture"),
);
try {
const result = await client.execute("list_architectures", {});
if (result.ok) throw new Error("Expected native command failure");
expect(result.error.cleanupIncomplete).toBe(true);
marker.resolveCleanup();
expect(result.error.cleanupIncomplete).toBe(true);
expect(result.error.cleanupResources).toEqual([
"run:known",
"run:unknown",
]);
const closed = await client.close();
expect(closed.ok).toBe(false);
if (closed.ok) throw new Error("Expected unknown owner cleanup facts");
expect(closed.error.cleanupIncomplete).toBe(true);
expect(closed.error.cleanupResources).toEqual([
"run:known",
"run:unknown",
]);
} finally {
await client.close();
}
});
it("projects native cancellation output and incomplete cleanup to callers", async () => {
const cancellationReason = { source: "caller", token: "exact reason" };
const capture = {
stdout: "partial output",
stderr: "partial diagnostic",
@@ -561,7 +1143,7 @@ describe("native command failure projection", () => {
tool,
"cancelled",
null,
undefined,
{ cancellation: { reason: cancellationReason } },
capture,
"process group cleanup was not confirmed",
["process-group:1234"],
@@ -575,6 +1157,7 @@ describe("native command failure projection", () => {
expect(result.ok).toBe(false);
if (result.ok) return;
expect(result.error.cause).toBe(cancellationReason);
expect(result.error.capturedOutput).toMatchObject({
stdout: "partial output",
stderr: "partial diagnostic",