fix(evm): preserve cancellation and exact cleanup ownership

This commit is contained in:
morluto
2026-10-20 17:17:16 +08:00
parent 419502c8d0
commit 459a356edb
4 changed files with 1037 additions and 116 deletions
+2 -1
View File
@@ -1,5 +1,6 @@
import type { EvmWorkerLimits } from "./EvmWorkerLimits.js";
import { ArtifactReaderFailure } from "../artifacts/ArtifactReader.js";
import { artifactFileCancellationCause } from "../artifacts/ArtifactFileOperation.js";
import {
AnalysisError,
type AnalysisCapturedOutput,
@@ -130,7 +131,7 @@ export const evmInterfaceFailure = (
if (cause instanceof ArtifactReaderFailure) {
if (cause.reason === "cancelled")
return new AnalysisCancelledError("inspect_evm_interface", {
cause,
cause: artifactFileCancellationCause(cause),
...cleanupOptions,
});
if (cause.reason === "integrity")
+304 -73
View File
@@ -15,13 +15,18 @@ import type {
import { readStableArtifact } from "../artifacts/readStableArtifact.js";
import { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js";
import { ArtifactReaderFailure } from "../artifacts/ArtifactReader.js";
import { bindArtifactFileCleanup } from "../artifacts/ArtifactFileOperation.js";
import {
AnalysisCleanupLease,
AnalysisCleanupLeaseGroup,
} from "../artifacts/AnalysisCleanupLease.js";
import {
OwnedCommandFailure,
ownedCommandResources,
} from "../process/OwnedCommand.js";
import type {
import {
AnalysisError,
AnalysisCapturedOutput,
type AnalysisCapturedOutput,
} from "../domain/analysisErrorBase.js";
import {
AnalysisCancelledError,
@@ -84,6 +89,71 @@ const replySchema = z.discriminatedUnion("ok", [
]);
type Launcher = NonNullable<Parameters<typeof runOwnedCommand>[2]>["launcher"];
type EvmOutcome = Result<EvmInterfaceExecution, AnalysisError>;
type EvmRuntimeRoot = Pick<PrivateRuntimeRoot, "path" | "close">;
type PendingEvmCleanup = {
readonly input: InspectEvmInterfaceInput;
readonly result: EvmOutcome;
readonly output: AnalysisCapturedOutput | undefined;
root:
| { readonly owner: EvmRuntimeRoot; readonly lease: AnalysisCleanupLease }
| undefined;
process: { readonly lease: AnalysisCleanupLease } | undefined;
};
const ownerCleanupMarker = (
owners: readonly AnalysisCleanupLease[],
): ProviderCleanupError =>
new ProviderCleanupError(
EVMOLE_PROVIDER_IDENTITY.id,
owners.flatMap((owner) => owner.cleanupResources ?? ["owned-EVM-resource"]),
{ reason: "Exact EVM cleanup owners have not confirmed release." },
{ operation: OPERATION },
);
const bindEvmCleanupOwners = (
failure: AnalysisError,
owners: readonly AnalysisCleanupLease[],
primary?: AnalysisError,
): void => {
const marker = ownerCleanupMarker(owners);
new AnalysisCleanupLeaseGroup(marker, owners);
failure.bindCleanupDependencies([
marker,
...(primary?.cleanupIncomplete === true ? [primary] : []),
]);
};
const retainedCleanupFailure = (cause: unknown): ProviderCleanupError => {
if (cause instanceof ProviderCleanupError) return cause;
if (cause instanceof ArtifactReaderFailure) {
const observation = ArtifactReaderFailure.cleanupObservation(
cause,
"EVM artifact resources",
);
const failure = new ProviderCleanupError(
EVMOLE_PROVIDER_IDENTITY.id,
observation.resources.length === 0
? ["EVM artifact resources"]
: observation.resources,
{ reason: observation.reason },
{ operation: OPERATION, cause },
);
bindArtifactFileCleanup(cause, failure);
return failure;
}
return new ProviderCleanupError(
EVMOLE_PROVIDER_IDENTITY.id,
cause instanceof AnalysisError && cause.cleanupResources.length > 0
? cause.cleanupResources
: ["EVM provider cleanup"],
{ reason: cause instanceof Error ? cause.message : String(cause) },
{
operation: OPERATION,
cause,
...(cause instanceof AnalysisError ? { cleanupPrimary: cause } : {}),
},
);
};
const compactErrorProjection = (error: AnalysisError) => {
const projection = projectAnalysisError(error);
@@ -121,6 +191,7 @@ const evmCleanupFailure = (
},
{
operation: OPERATION,
...(result.ok ? {} : { cleanupPrimary: result.error }),
...(cause === undefined ? {} : { cause }),
...(partialObservation === undefined ? {} : { partialObservation }),
},
@@ -130,15 +201,27 @@ const evmCleanupFailure = (
/** Run unchanged EVMole/WASM in an owned worker; no target, chain or network execution. */
export class EvmoleInterfaceProvider implements EvmInterfacePort {
readonly #resources = new ArtifactResourceScope();
#resourceScopeLease: AnalysisCleanupLease | undefined;
#tail: Promise<void> = Promise.resolve();
#closing = false;
#closePromise: Promise<void> | undefined;
#pendingCleanup:
| {
root: Pick<PrivateRuntimeRoot, "path" | "close"> | undefined;
process: ProviderProcessSupervisor | undefined;
}
| undefined;
#pendingCleanup: PendingEvmCleanup | undefined;
#artifactScopeLease(): AnalysisCleanupLease {
if (this.#resourceScopeLease !== undefined) return this.#resourceScopeLease;
const provider = this;
const owner = {
close: () => provider.#resources.close(),
get cleanupResources() {
return provider.#resources.cleanupResources;
},
};
this.#resourceScopeLease = AnalysisCleanupLease.forCloseOwner(
owner,
"EVM artifact resources",
);
return this.#resourceScopeLease;
}
constructor(
readonly environment: Readonly<NodeJS.ProcessEnv> = process.env,
@@ -159,7 +242,10 @@ export class EvmoleInterfaceProvider implements EvmInterfacePort {
try {
return await this.#resources.run(() => this.#inspect(input, options));
} catch (cause: unknown) {
return err(evmInterfaceFailure(cause, "artifact-read", input.path));
const failure = evmInterfaceFailure(cause, "artifact-read", input.path);
if (cause instanceof ArtifactReaderFailure)
bindArtifactFileCleanup(cause, failure);
return err(failure);
}
});
this.#tail = operation.then(
@@ -176,13 +262,48 @@ export class EvmoleInterfaceProvider implements EvmInterfacePort {
const pending = await this.#retire();
if (pending !== undefined) return err(pending);
let root: Pick<PrivateRuntimeRoot, "path" | "close"> | undefined;
let rootLease: AnalysisCleanupLease | undefined;
let processOwner: ProviderProcessSupervisor | undefined;
let processLease: AnalysisCleanupLease | undefined;
let processCleanup:
| {
readonly reason: string;
readonly resources: readonly string[];
}
| undefined;
const processLeases = new WeakMap<
ProviderProcessSupervisor,
AnalysisCleanupLease
>();
const processLeaseFor = (owner: ProviderProcessSupervisor) => {
let lease = processLeases.get(owner);
if (lease === undefined) {
lease = new AnalysisCleanupLease(
async () => {
const stopped = await owner.stop();
if (
stopped.status === "incomplete" ||
stopped.status === "not-owned"
)
throw new Error(
stopped.status === "incomplete"
? stopped.reason
: "Owned EVM worker cleanup did not confirm ownership.",
{ cause: stopped },
);
},
owner.launch.ownership === undefined
? ["owned-worker"]
: ownedCommandResources(owner.launch.ownership, root?.path),
);
processLeases.set(owner, lease);
}
return lease;
};
const rootLeaseFor = (owner: EvmRuntimeRoot) => {
rootLease ??= AnalysisCleanupLease.forCloseOwner(owner, owner.path);
return rootLease;
};
let retainedOutput: AnalysisCapturedOutput | undefined;
let phase: "configuration" | "artifact-read" | "worker" = "configuration";
let workerLimits: EvmWorkerLimits | undefined;
@@ -251,6 +372,7 @@ export class EvmoleInterfaceProvider implements EvmInterfacePort {
);
phase = "worker";
root = await this.createRuntime();
rootLeaseFor(root);
const snapshotPath = join(root.path, "carrier.snapshot");
const requestPath = join(root.path, "request.json");
const replyPath = join(root.path, "reply.json");
@@ -303,6 +425,7 @@ export class EvmoleInterfaceProvider implements EvmInterfacePort {
...(this.launcher === undefined ? {} : { launcher: this.launcher }),
retainCleanup: (owner, cleanup) => {
processOwner = owner;
processLease = processLeaseFor(owner);
processCleanup = {
reason: cleanup.reason,
resources:
@@ -328,13 +451,17 @@ export class EvmoleInterfaceProvider implements EvmInterfacePort {
} catch (cause: unknown) {
const cleanup =
cause instanceof ArtifactReaderFailure ? cause.cleanup : undefined;
if (options?.signal?.aborted)
throw new AnalysisCancelledError(OPERATION, {
if (options?.signal?.aborted) {
const failure = new AnalysisCancelledError(OPERATION, {
cause: options.signal.reason,
capturedOutput,
...(cleanup === undefined ? {} : { cleanup }),
});
throw new AnalysisOutputError(
if (cause instanceof ArtifactReaderFailure)
bindArtifactFileCleanup(cause, failure);
throw failure;
}
const failure = new AnalysisOutputError(
OPERATION,
`Owned interface worker reply failed for ${input.path}: ${cause instanceof Error ? cause.message : String(cause)}`,
{
@@ -343,6 +470,9 @@ export class EvmoleInterfaceProvider implements EvmInterfacePort {
...(cleanup === undefined ? {} : { cleanup }),
},
);
if (cause instanceof ArtifactReaderFailure)
bindArtifactFileCleanup(cause, failure);
throw failure;
}
if (!reply.ok) {
if (reply.reason === "unsupported")
@@ -446,8 +576,10 @@ export class EvmoleInterfaceProvider implements EvmInterfacePort {
if (
cause instanceof OwnedCommandFailure &&
cause.cleanupOwner !== undefined
)
) {
processOwner = cause.cleanupOwner;
processLease = processLeaseFor(cause.cleanupOwner);
}
let fileFailure: EvmFileSizeFailureEvidence | undefined;
let markerCleanupFailure: ArtifactReaderFailure | undefined;
if (
@@ -487,7 +619,7 @@ export class EvmoleInterfaceProvider implements EvmInterfacePort {
};
}
}
result = err(
const failure =
markerCleanupFailure !== undefined
? evmInterfaceFailure(
markerCleanupFailure,
@@ -517,78 +649,154 @@ export class EvmoleInterfaceProvider implements EvmInterfacePort {
workerLimits,
limiter,
fileFailure,
),
);
);
const artifactFailure =
markerCleanupFailure ??
(cause instanceof ArtifactReaderFailure ? cause : undefined);
if (artifactFailure !== undefined)
bindArtifactFileCleanup(artifactFailure, failure);
result = err(failure);
}
if (result.ok && options?.signal?.aborted)
result = err(
new AnalysisCancelledError(OPERATION, {
cause: options.signal.reason,
partialObservation: evmInterfaceEvidence(input, result.value),
...(retainedOutput === undefined
? {}
: { capturedOutput: retainedOutput }),
}),
);
if (processOwner !== undefined) {
const lease = processLeaseFor(processOwner);
const rootOwnerLease =
root === undefined ? undefined : rootLeaseFor(root);
this.#pendingCleanup = {
root,
process: processOwner,
input,
result,
output: retainedOutput,
root:
root === undefined || rootOwnerLease === undefined
? undefined
: { owner: root, lease: rootOwnerLease },
process: { lease },
};
if (processCleanup === undefined) return result;
return err(
evmCleanupFailure(
if (processCleanup === undefined) {
const owners = [lease, rootOwnerLease].filter(
(owner): owner is AnalysisCleanupLease => owner !== undefined,
);
if (!result.ok && result.error.cleanupIncomplete) {
bindEvmCleanupOwners(result.error, owners);
return result;
}
const failure = evmCleanupFailure(
input,
result,
[
...new Set([
...(root === undefined ? [] : [root.path]),
...processCleanup.resources,
...(lease.cleanupResources ?? []),
]),
],
processCleanup.reason,
"Owned EVM worker cleanup remained unresolved without a failure description.",
retainedOutput,
processOwner,
),
);
bindEvmCleanupOwners(
failure,
owners,
result.ok ? undefined : result.error,
);
return err(failure);
}
const failure = evmCleanupFailure(
input,
result,
[
...new Set([
...(root === undefined ? [] : [root.path]),
...processCleanup.resources,
]),
],
processCleanup.reason,
retainedOutput,
processOwner,
);
bindEvmCleanupOwners(
failure,
[lease, rootOwnerLease].filter(
(lease): lease is AnalysisCleanupLease => lease !== undefined,
),
result.ok ? undefined : result.error,
);
return err(failure);
}
if (root !== undefined) {
try {
await root.close();
await rootLeaseFor(root).release();
} catch (cause: unknown) {
const retainedRoot = root;
this.#pendingCleanup = { root: retainedRoot, process: undefined };
const lease = rootLeaseFor(retainedRoot);
const cleanupResult =
result.ok && options?.signal?.aborted
? err(
new AnalysisCancelledError(OPERATION, {
cause: options.signal.reason,
partialObservation: evmInterfaceEvidence(input, result.value),
...(retainedOutput === undefined
? {}
: { capturedOutput: retainedOutput }),
}),
)
: result;
this.#pendingCleanup = {
input,
result: cleanupResult,
output: retainedOutput,
root: { owner: retainedRoot, lease },
process: undefined,
};
root = undefined;
return err(
evmCleanupFailure(
input,
result,
[retainedRoot.path],
cause instanceof Error ? cause.message : String(cause),
retainedOutput,
cause,
),
const failure = evmCleanupFailure(
input,
cleanupResult,
[retainedRoot.path],
cause instanceof Error ? cause.message : String(cause),
retainedOutput,
cause,
);
bindEvmCleanupOwners(
failure,
[lease],
cleanupResult.ok ? undefined : cleanupResult.error,
);
return err(failure);
}
}
return result.ok && options?.signal?.aborted
? err(
new AnalysisCancelledError(OPERATION, {
cause: options.signal.reason,
partialObservation: evmInterfaceEvidence(input, result.value),
...(retainedOutput === undefined
? {}
: { capturedOutput: retainedOutput }),
}),
)
: result;
return result;
}
async close(): Promise<void> {
this.#closing = true;
this.#closePromise ??= this.#tail
.then(async () => {
const outcomes = await Promise.allSettled([
const [retirement, artifacts] = await Promise.allSettled([
this.#retire(),
this.#resources.close(),
this.#artifactScopeLease().release(),
]);
const failures = outcomes.flatMap((outcome) =>
outcome.status === "rejected"
? [outcome.reason]
: outcome.value === undefined
? []
: [outcome.value],
);
const failures: AnalysisError[] = [];
if (retirement.status === "rejected")
failures.push(retainedCleanupFailure(retirement.reason));
else if (retirement.value !== undefined)
failures.push(retirement.value);
if (artifacts.status === "rejected") {
const scopeLease = this.#artifactScopeLease();
const artifactFailure = retainedCleanupFailure(artifacts.reason);
const marker = ownerCleanupMarker([scopeLease]);
new AnalysisCleanupLeaseGroup(marker, [scopeLease]);
artifactFailure.bindCleanupDependencies([marker]);
failures.push(artifactFailure);
}
if (failures.length > 0) {
const resources = failures.flatMap((failure) =>
failure instanceof ProviderCleanupError
@@ -598,7 +806,7 @@ export class EvmoleInterfaceProvider implements EvmInterfacePort {
? failure.cleanup.resources
: ["artifact-resources"],
);
throw new ProviderCleanupError(
const failure = new ProviderCleanupError(
EVMOLE_PROVIDER_IDENTITY.id,
[...new Set(resources)],
{
@@ -608,6 +816,8 @@ export class EvmoleInterfaceProvider implements EvmInterfacePort {
},
{ operation: OPERATION, cause: new AggregateError(failures) },
);
failure.bindCleanupDependencies(failures);
throw failure;
}
})
.catch((cause: unknown) => {
@@ -630,33 +840,54 @@ export class EvmoleInterfaceProvider implements EvmInterfacePort {
const pending = this.#pendingCleanup;
if (pending === undefined) return undefined;
if (pending.process !== undefined) {
const stopped = await pending.process.stop();
if (stopped.status === "incomplete")
return new ProviderCleanupError(
EVMOLE_PROVIDER_IDENTITY.id,
try {
await pending.process.lease.release();
} catch (cause: unknown) {
const failure = evmCleanupFailure(
pending.input,
pending.result,
[
...(pending.root === undefined ? [] : [pending.root.path]),
...(pending.process.launch.ownership === undefined
? ["owned-worker"]
: ownedCommandResources(pending.process.launch.ownership)),
...new Set([
...(pending.root === undefined ? [] : [pending.root.owner.path]),
...(pending.process.lease.cleanupResources ?? []),
]),
],
{ reason: stopped.reason },
{ operation: OPERATION },
cause instanceof Error ? cause.message : String(cause),
pending.output,
cause,
);
bindEvmCleanupOwners(
failure,
[
pending.process.lease,
...(pending.root === undefined ? [] : [pending.root.lease]),
],
pending.result.ok ? undefined : pending.result.error,
);
return failure;
}
pending.process = undefined;
}
if (pending.root !== undefined) {
const root = pending.root;
const { owner: root, lease } = pending.root;
try {
await root.close();
await lease.release();
pending.root = undefined;
} catch (cause: unknown) {
return new ProviderCleanupError(
EVMOLE_PROVIDER_IDENTITY.id,
[root.path],
{ reason: cause instanceof Error ? cause.message : String(cause) },
{ operation: OPERATION, cause },
const failure = evmCleanupFailure(
pending.input,
pending.result,
lease.cleanupResources ?? [root.path],
cause instanceof Error ? cause.message : String(cause),
pending.output,
cause,
);
bindEvmCleanupOwners(
failure,
[lease],
pending.result.ok ? undefined : pending.result.error,
);
return failure;
}
}
this.#pendingCleanup = undefined;
+648
View File
@@ -0,0 +1,648 @@
import { createHash } from "node:crypto";
import { mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { expect, it, vi } from "vitest";
const limitReport = [
"Max address space unlimited unlimited bytes",
"Max cpu time unlimited unlimited seconds",
"Max file size unlimited unlimited bytes",
].join("\n");
const closeState = vi.hoisted(() => ({
path: "",
failures: 0,
onClosed: undefined as (() => void) | undefined,
onLimits: undefined as (() => void) | undefined,
}));
vi.mock("node:fs/promises", async (importOriginal) => {
const fs = await importOriginal<typeof import("node:fs/promises")>();
return {
...fs,
open: async (...args: Parameters<typeof fs.open>) => {
const handle = await fs.open(...args);
if (args[0] !== closeState.path) return handle;
const close = handle.close.bind(handle);
handle.close = async () => {
if (closeState.failures > 0) {
closeState.failures--;
throw new Error("injected EVM artifact descriptor close failure");
}
await close();
closeState.onClosed?.();
};
return handle;
},
readFile: async (...args: Parameters<typeof fs.readFile>) => {
if (args[0] !== "/proc/self/limits") return fs.readFile(...args);
closeState.onLimits?.();
return limitReport;
},
};
});
import { EvmoleInterfaceProvider } from "../../../src/evm/EvmoleInterfaceProvider.js";
import { spawnOwnedProviderProcess } from "../../../src/process/ProviderProcess.js";
import { cleanupOwnedProcessGroup } from "../../../src/process/ProcessOwnership.js";
import { PrivateRuntimeRoot } from "../../../src/process/PrivateRuntimeRoot.js";
import { projectAnalysisError } from "../../../src/domain/analysisErrorProjection.js";
import { waitForProviderProcessReady } from "../../fixtures/providerProcess.js";
import { EvmInterfaceService } from "../../../src/application/evm/EvmInterfaceService.js";
import { AnalysisCancelledError } from "../../../src/domain/analysisErrorCore.js";
it.each(["pre-aborted", "admission", "close"] as const)(
"preserves opaque caller reasons through service cancellation at %s",
async (stage) => {
const platform = Object.getOwnPropertyDescriptor(process, "platform");
const arch = Object.getOwnPropertyDescriptor(process, "arch");
Object.defineProperty(process, "platform", { ...platform, value: "linux" });
Object.defineProperty(process, "arch", { ...arch, value: "x64" });
const workspace = await mkdtemp(join(tmpdir(), "rea-evm-artifact-cancel-"));
const input = join(workspace, "selected.hex");
await writeFile(input, "0x60006000f3\n");
const createRuntime = vi.fn(async () => {
throw new Error("Worker must not start");
});
const provider = new EvmoleInterfaceProvider(
{ REA_EVM_PRLIMIT_COMMAND: process.execPath },
undefined,
createRuntime,
);
try {
for (const reason of [
null,
"caller stopped",
{ requested: "stop" },
new Error("caller stopped"),
new AnalysisCancelledError("foreign", {
cleanup: { reason: "foreign owner", resources: ["foreign-resource"] },
}),
]) {
const controller = new AbortController();
let closed = false;
closeState.path = input;
closeState.onLimits =
stage === "admission" ? () => controller.abort(reason) : undefined;
closeState.onClosed = () => {
closed = true;
if (stage === "close") controller.abort(reason);
};
if (stage === "pre-aborted") controller.abort(reason);
const result = await new EvmInterfaceService(provider).inspect(
{ path: input, encoding: "hex" },
{ signal: controller.signal },
);
if (result.ok) throw new Error("Expected cancellation");
expect(result.error).toBeInstanceOf(AnalysisCancelledError);
expect(result.error.cause).toBe(reason);
expect(result.error.cleanupIncomplete).toBe(false);
expect(result.error.cleanupResources).toEqual([]);
expect(closed).toBe(stage === "close");
expect(createRuntime).not.toHaveBeenCalled();
}
} finally {
closeState.path = "";
closeState.onClosed = undefined;
closeState.onLimits = undefined;
await provider.close();
await rm(workspace, { recursive: true, force: true });
if (platform !== undefined)
Object.defineProperty(process, "platform", platform);
if (arch !== undefined) Object.defineProperty(process, "arch", arch);
}
},
);
it.each([
"completed",
"malformed-reply",
"cancelled-cleanup",
"physical-before-late-cancel",
] as const)(
"updates a returned cleanup error after later root release: %s",
async (scenario) => {
const platform = Object.getOwnPropertyDescriptor(process, "platform");
const arch = Object.getOwnPropertyDescriptor(process, "arch");
Object.defineProperty(process, "platform", { ...platform, value: "linux" });
Object.defineProperty(process, "arch", { ...arch, value: "x64" });
const workspace = await mkdtemp(join(tmpdir(), "rea-evm-cleanup-retry-"));
const input = join(workspace, "selected.hex");
await writeFile(input, "0x60006000f3\n");
let rootPath = "";
let closeAttempts = 0;
const controller = new AbortController();
const abortReason = { requested: "cancel during root cleanup" };
const provider = new EvmoleInterfaceProvider(
{ REA_EVM_PRLIMIT_COMMAND: process.execPath },
async (spawn) => {
// Emulate only Linux admission and limits; restore Darwin before the
// owned process captures and validates its real host identity.
if (platform !== undefined)
Object.defineProperty(process, "platform", platform);
if (arch !== undefined) Object.defineProperty(process, "arch", arch);
const requestPath = spawn.arguments.at(-1);
if (requestPath === undefined) throw new Error("Request path missing");
const request = JSON.parse(await readFile(requestPath, "utf8")) as {
reply_path: string;
};
if (
scenario === "malformed-reply" ||
scenario === "physical-before-late-cancel"
)
await writeFile(request.reply_path, "malformed worker reply");
else {
const bytes = Buffer.from("60006000f3", "hex");
await writeFile(
request.reply_path,
JSON.stringify({
ok: true,
version: "0.9.3",
bytecode: {
sha256: createHash("sha256").update(bytes).digest("hex"),
bytes: bytes.length,
hex: bytes.toString("hex"),
},
raw: { functions: [] },
}),
);
}
return spawnOwnedProviderProcess({
...spawn,
command: process.execPath,
arguments: [
"-e",
'process.stderr.write("controlled worker complete")',
],
});
},
async () => {
const runtime = await PrivateRuntimeRoot.create({
prefix: "rea-evm-cleanup-retry-",
});
rootPath = runtime.path;
const close = runtime.close.bind(runtime);
return {
path: runtime.path,
close: async () => {
closeAttempts++;
if (scenario === "cancelled-cleanup" && closeAttempts === 1)
controller.abort(abortReason);
if (
scenario === "physical-before-late-cancel" &&
closeAttempts === 1
) {
setImmediate(() => controller.abort(abortReason));
throw new Error("physical root close failure before late abort");
}
if (closeAttempts <= 2)
throw new Error("injected retryable root close failure");
await close();
},
};
},
);
try {
const result = await provider.inspect(
{ path: input, encoding: "hex" },
scenario === "cancelled-cleanup" ||
scenario === "physical-before-late-cancel"
? { signal: controller.signal }
: undefined,
);
expect(result.ok).toBe(false);
if (result.ok) throw new Error("Expected the first root close to fail");
const originalError = result.error;
if (scenario === "physical-before-late-cancel")
await new Promise<void>((resolve) => setImmediate(resolve));
expect(originalError.cleanupIncomplete).toBe(true);
expect(originalError.cleanupResources).toContain(rootPath);
if (scenario === "cancelled-cleanup")
expect(originalError.cleanupPrimary?.cause).toBe(abortReason);
if (scenario === "physical-before-late-cancel")
expect(originalError.cleanupPrimary?.cleanupIncomplete).not.toBe(true);
const closeFailure = await provider.close().then(
() => undefined,
(cause: unknown) => cause,
);
expect(closeFailure).toMatchObject({ cleanupIncomplete: true });
if (!(closeFailure instanceof Error))
throw new Error("Expected close cleanup failure");
const reportedCloseError = closeFailure as typeof originalError;
expect(reportedCloseError.cleanupResources).toContain(rootPath);
expect(closeAttempts).toBe(2);
await provider.close();
expect(closeAttempts).toBe(3);
await expect(stat(rootPath)).rejects.toMatchObject({
code: "ENOENT",
});
expect(originalError.cleanupIncomplete).toBe(false);
expect(originalError.cleanupResources).toEqual([]);
expect(reportedCloseError.cleanupIncomplete).toBe(false);
expect(reportedCloseError.cleanupResources).toEqual([]);
expect(projectAnalysisError(originalError).code).toBe(
scenario === "cancelled-cleanup"
? "cancelled"
: scenario === "malformed-reply" ||
scenario === "physical-before-late-cancel"
? "unreadable_output"
: "execution_failure",
);
} finally {
try {
await provider.close();
} catch {
// Preserve the assertion's original failure; the workspace cleanup follows.
}
await rm(workspace, { recursive: true, force: true });
if (platform !== undefined)
Object.defineProperty(process, "platform", platform);
if (arch !== undefined) Object.defineProperty(process, "arch", arch);
}
},
);
it("keeps descriptor cleanup in close aggregates until the exact file lease releases", async () => {
const platform = Object.getOwnPropertyDescriptor(process, "platform");
const arch = Object.getOwnPropertyDescriptor(process, "arch");
Object.defineProperty(process, "platform", { ...platform, value: "linux" });
Object.defineProperty(process, "arch", { ...arch, value: "x64" });
const workspace = await mkdtemp(join(tmpdir(), "rea-evm-descriptor-retry-"));
const input = join(workspace, "selected.hex");
await writeFile(input, "0x60006000f3\n");
closeState.path = input;
closeState.failures = 2;
const provider = new EvmoleInterfaceProvider({
REA_EVM_PRLIMIT_COMMAND: process.execPath,
});
try {
const result = await provider.inspect({ path: input, encoding: "hex" });
expect(result.ok).toBe(false);
if (result.ok) throw new Error("Expected an artifact descriptor failure");
expect(result.error.cleanupIncomplete).toBe(true);
const aggregate = await provider.close().then(
() => undefined,
(cause: unknown) => cause,
);
expect(aggregate).toMatchObject({
cleanupIncomplete: true,
cleanupResources: [input],
});
if (!(aggregate instanceof Error))
throw new Error("Expected a descriptor cleanup aggregate");
const returnedAggregate = aggregate as typeof result.error;
expect(result.error.cleanupIncomplete).toBe(true);
await provider.close();
expect(result.error.cleanupIncomplete).toBe(false);
expect(result.error.cleanupResources).toEqual([]);
expect(returnedAggregate.cleanupIncomplete).toBe(false);
expect(returnedAggregate.cleanupResources).toEqual([]);
} finally {
closeState.path = "";
closeState.failures = 0;
try {
await provider.close();
} catch {
// Preserve the assertion's original failure while releasing test files.
}
await rm(workspace, { recursive: true, force: true });
if (platform !== undefined)
Object.defineProperty(process, "platform", platform);
if (arch !== undefined) Object.defineProperty(process, "arch", arch);
}
});
it("keeps a mixed root and reply-descriptor aggregate pending until both owners release", async () => {
const platform = Object.getOwnPropertyDescriptor(process, "platform");
const arch = Object.getOwnPropertyDescriptor(process, "arch");
Object.defineProperty(process, "platform", { ...platform, value: "linux" });
Object.defineProperty(process, "arch", { ...arch, value: "x64" });
const workspace = await mkdtemp(join(tmpdir(), "rea-evm-mixed-owners-"));
const input = join(workspace, "selected.hex");
await writeFile(input, "0x60006000f3\n");
closeState.path = "";
closeState.failures = 0;
let replyPath = "";
let rootPath = "";
let rootCloseAttempts = 0;
const provider = new EvmoleInterfaceProvider(
{ REA_EVM_PRLIMIT_COMMAND: process.execPath },
async (spawn) => {
const requestPath = spawn.arguments.at(-1);
if (requestPath === undefined) throw new Error("Request path missing");
const request = JSON.parse(await readFile(requestPath, "utf8")) as {
reply_path: string;
};
replyPath = request.reply_path;
closeState.path = replyPath;
closeState.failures = 2;
await writeFile(replyPath, "malformed worker reply");
if (platform !== undefined)
Object.defineProperty(process, "platform", platform);
if (arch !== undefined) Object.defineProperty(process, "arch", arch);
return spawnOwnedProviderProcess({
...spawn,
command: process.execPath,
arguments: ["-e", 'process.stderr.write("controlled worker complete")'],
});
},
async () => {
const root = await PrivateRuntimeRoot.create({
prefix: "rea-evm-mixed-owners-",
});
rootPath = root.path;
const close = root.close.bind(root);
return {
path: root.path,
close: async () => {
rootCloseAttempts++;
if (rootCloseAttempts <= 3)
throw new Error("injected root cleanup refusal");
await close();
},
};
},
);
try {
const result = await provider.inspect({ path: input, encoding: "hex" });
expect(result.ok).toBe(false);
if (result.ok) throw new Error("Expected reply descriptor cleanup failure");
const original = result.error;
expect(original.cleanupIncomplete).toBe(true);
expect(original.cleanupResources).toContain(rootPath);
expect(original.cleanupResources).toContain(replyPath);
const firstAggregate = await provider.close().then(
() => undefined,
(cause: unknown) => cause,
);
expect(firstAggregate).toMatchObject({ cleanupIncomplete: true });
if (!(firstAggregate instanceof Error))
throw new Error("Expected mixed cleanup aggregate");
const retainedAggregate = firstAggregate as typeof original;
expect(retainedAggregate.cleanupResources).toContain(rootPath);
expect(retainedAggregate.cleanupResources).toContain(replyPath);
await expect(provider.close()).rejects.toMatchObject({
cleanupIncomplete: true,
cleanupResources: [rootPath],
});
expect(retainedAggregate.cleanupIncomplete).toBe(true);
expect(retainedAggregate.cleanupResources).toEqual([rootPath]);
expect(original.cleanupIncomplete).toBe(true);
expect(original.cleanupResources).toEqual([rootPath]);
await provider.close();
expect(retainedAggregate.cleanupIncomplete).toBe(false);
expect(retainedAggregate.cleanupResources).toEqual([]);
expect(original.cleanupIncomplete).toBe(false);
expect(original.cleanupResources).toEqual([]);
} finally {
closeState.path = "";
closeState.failures = 0;
try {
await provider.close();
} catch {
// Preserve the assertion's original failure while releasing test files.
}
await rm(workspace, { recursive: true, force: true });
if (platform !== undefined)
Object.defineProperty(process, "platform", platform);
if (arch !== undefined) Object.defineProperty(process, "arch", arch);
}
});
it("releases a retained process independently while a root cleanup retry remains pending", async () => {
const platform = Object.getOwnPropertyDescriptor(process, "platform");
const arch = Object.getOwnPropertyDescriptor(process, "arch");
Object.defineProperty(process, "platform", { ...platform, value: "linux" });
Object.defineProperty(process, "arch", { ...arch, value: "x64" });
const workspace = await mkdtemp(
join(tmpdir(), "rea-evm-process-root-retry-"),
);
const input = join(workspace, "selected.hex");
await writeFile(input, "0x60006000f3\n");
let rootPath = "";
let rootCloseAttempts = 0;
let cleanupAttempts = 0;
let allowProcessCleanup = false;
const provider = new EvmoleInterfaceProvider(
{ REA_EVM_PRLIMIT_COMMAND: process.execPath },
async (spawn) => {
const requestPath = spawn.arguments.at(-1);
if (requestPath === undefined) throw new Error("Request path missing");
const request = JSON.parse(await readFile(requestPath, "utf8")) as {
reply_path: string;
};
const bytes = Buffer.from("60006000f3", "hex");
await writeFile(
request.reply_path,
JSON.stringify({
ok: true,
version: "0.9.3",
bytecode: {
sha256: createHash("sha256").update(bytes).digest("hex"),
bytes: bytes.length,
hex: bytes.toString("hex"),
},
raw: { functions: [] },
}),
);
if (platform !== undefined)
Object.defineProperty(process, "platform", platform);
if (arch !== undefined) Object.defineProperty(process, "arch", arch);
const launched = await spawnOwnedProviderProcess({
...spawn,
command: process.execPath,
arguments: ["-e", 'process.stderr.write("worker complete")'],
});
const cleanup =
launched.cleanup ??
(() => cleanupOwnedProcessGroup(launched.ownership));
return {
...launched,
cleanup: async () => {
cleanupAttempts++;
if (!allowProcessCleanup)
throw new Error("injected retained process cleanup failure");
return cleanup();
},
};
},
async () => {
const root = await PrivateRuntimeRoot.create({
prefix: "rea-evm-process-root-retry-",
});
rootPath = root.path;
const close = root.close.bind(root);
return {
path: root.path,
close: async () => {
rootCloseAttempts++;
if (rootCloseAttempts === 1)
throw new Error("injected retained root cleanup failure");
await close();
},
};
},
);
try {
const result = await provider.inspect({ path: input, encoding: "hex" });
expect(result.ok).toBe(false);
if (result.ok) throw new Error("Expected retained process cleanup failure");
const original = result.error;
expect(original.cleanupIncomplete).toBe(true);
expect(original.cleanupResources).toContain(rootPath);
expect(original.partialObservation).toMatchObject({
normalized_result: {
bytecode: { hex: "60006000f3" },
evidence_kind: "inferred",
},
});
allowProcessCleanup = true;
const firstClose = await provider.close().then(
() => undefined,
(cause: unknown) => cause,
);
expect(firstClose).toMatchObject({
cleanupIncomplete: true,
cleanupResources: [rootPath],
});
expect(cleanupAttempts).toBe(2);
expect(rootCloseAttempts).toBe(1);
expect(original.cleanupIncomplete).toBe(true);
expect(original.cleanupResources).toEqual([rootPath]);
await provider.close();
expect(original.cleanupIncomplete).toBe(false);
expect(original.cleanupResources).toEqual([]);
expect(cleanupAttempts).toBe(2);
await expect(stat(rootPath)).rejects.toMatchObject({ code: "ENOENT" });
} finally {
allowProcessCleanup = true;
closeState.path = "";
closeState.failures = 0;
try {
await provider.close();
} catch {
// Preserve the assertion's original failure while releasing test files.
}
await rm(workspace, { recursive: true, force: true });
if (platform !== undefined)
Object.defineProperty(process, "platform", platform);
if (arch !== undefined) Object.defineProperty(process, "arch", arch);
}
});
it("preserves an active cancellation through denied owned-group cleanup and later retry", async () => {
const platform = Object.getOwnPropertyDescriptor(process, "platform");
const arch = Object.getOwnPropertyDescriptor(process, "arch");
Object.defineProperty(process, "platform", { ...platform, value: "linux" });
Object.defineProperty(process, "arch", { ...arch, value: "x64" });
const workspace = await mkdtemp(join(tmpdir(), "rea-evm-cancel-process-"));
const input = join(workspace, "selected.hex");
await writeFile(input, "0x60006000f3\n");
const controller = new AbortController();
const abortReason = { requested: "cancel the owned EVM worker" };
let rootPath = "";
let pid: number | undefined;
let cleanupAttempts = 0;
let allowRootClose = false;
const provider = new EvmoleInterfaceProvider(
{ REA_EVM_PRLIMIT_COMMAND: process.execPath },
async (spawn) => {
if (platform !== undefined)
Object.defineProperty(process, "platform", platform);
if (arch !== undefined) Object.defineProperty(process, "arch", arch);
const launched = await spawnOwnedProviderProcess({
...spawn,
command: process.execPath,
arguments: [
"-e",
'process.stdout.write("ready\\n");setInterval(()=>undefined,1000)',
],
});
pid = launched.process.pid;
await waitForProviderProcessReady(launched.process);
setImmediate(() => controller.abort(abortReason));
const cleanup =
launched.cleanup ??
(() => cleanupOwnedProcessGroup(launched.ownership));
return {
...launched,
cleanup: async () => {
cleanupAttempts++;
if (cleanupAttempts === 1)
throw new Error("injected owned-group cleanup refusal");
return cleanup();
},
};
},
async () => {
const root = await PrivateRuntimeRoot.create({
prefix: "rea-evm-cancel-process-",
});
rootPath = root.path;
const close = root.close.bind(root);
return {
path: root.path,
close: async () => {
if (!allowRootClose) {
allowRootClose = true;
throw new Error("injected retained root cleanup failure");
}
await close();
},
};
},
);
try {
const result = await provider.inspect(
{ path: input, encoding: "hex" },
{ signal: controller.signal },
);
expect(result.ok).toBe(false);
if (result.ok) throw new Error("Expected cancelled worker cleanup failure");
const original = result.error;
expect(original.cleanupIncomplete).toBe(true);
expect(original.cleanupResources).toContain(rootPath);
expect(original.cleanupPrimary?.cause).toBe(abortReason);
if (pid === undefined) throw new Error("Owned worker pid was not captured");
const ownedPid = pid;
expect(() => process.kill(ownedPid, 0)).not.toThrow();
const firstClose = await provider.close().then(
() => undefined,
(cause: unknown) => cause,
);
expect(firstClose).toMatchObject({
cleanupIncomplete: true,
cleanupResources: [rootPath],
});
expect(cleanupAttempts).toBe(2);
expect(original.cleanupResources).toEqual([rootPath]);
expect(() => process.kill(ownedPid, 0)).toThrow();
await provider.close();
expect(original.cleanupIncomplete).toBe(false);
expect(original.cleanupResources).toEqual([]);
expect(projectAnalysisError(original).category).toBe("cancelled");
expect(original.cleanupPrimary?.cause).toBe(abortReason);
expect(cleanupAttempts).toBe(2);
await expect(stat(rootPath)).rejects.toMatchObject({ code: "ENOENT" });
} finally {
allowRootClose = true;
try {
await provider.close();
} catch {
// Preserve the assertion's original failure while releasing test files.
}
await rm(workspace, { recursive: true, force: true });
if (platform !== undefined)
Object.defineProperty(process, "platform", platform);
if (arch !== undefined) Object.defineProperty(process, "arch", arch);
}
});
@@ -206,51 +206,72 @@ it.runIf(!unsupportedHost).each(["completed", "malformed-reply"] as const)(
async (scenario) => {
const { path } = await fixture();
let ownedRoot = "";
let denyRootClose = true;
let launches = 0;
let mayClean = false;
const provider = new EvmoleInterfaceProvider({}, async (spawn) => {
launches += 1;
ownedRoot = spawn.cwd ?? "";
const requestPath = spawn.arguments.at(-1);
if (requestPath === undefined) throw new Error("Request missing");
const request = requestSchema.parse(
JSON.parse(await readFile(requestPath, "utf8")),
);
if (scenario === "malformed-reply")
await writeFile(request.reply_path, "malformed worker reply");
else {
const bytes = Buffer.from("60006000f3", "hex");
await writeFile(
request.reply_path,
JSON.stringify({
ok: true,
version: "0.9.3",
bytecode: {
sha256: createHash("sha256").update(bytes).digest("hex"),
bytes: bytes.length,
hex: bytes.toString("hex"),
},
raw: { functions: [] },
}),
const provider = new EvmoleInterfaceProvider(
{},
async (spawn) => {
launches += 1;
ownedRoot = spawn.cwd ?? "";
const requestPath = spawn.arguments.at(-1);
if (requestPath === undefined) throw new Error("Request missing");
const request = requestSchema.parse(
JSON.parse(await readFile(requestPath, "utf8")),
);
}
const launched = await spawnOwnedProviderProcess({
...spawn,
command: process.execPath,
arguments: ["-e", 'process.stderr.write("worker completed")'],
});
const cleanup =
launched.cleanup ??
(() => cleanupOwnedProcessGroup(launched.ownership));
return {
...launched,
cleanup: async () => {
const stopped = await cleanup();
if (!mayClean) throw new Error("injected retained cleanup failure");
return stopped;
},
};
});
if (scenario === "malformed-reply")
await writeFile(request.reply_path, "malformed worker reply");
else {
const bytes = Buffer.from("60006000f3", "hex");
await writeFile(
request.reply_path,
JSON.stringify({
ok: true,
version: "0.9.3",
bytecode: {
sha256: createHash("sha256").update(bytes).digest("hex"),
bytes: bytes.length,
hex: bytes.toString("hex"),
},
raw: { functions: [] },
}),
);
}
const launched = await spawnOwnedProviderProcess({
...spawn,
command: process.execPath,
arguments: ["-e", 'process.stderr.write("worker completed")'],
});
const cleanup =
launched.cleanup ??
(() => cleanupOwnedProcessGroup(launched.ownership));
return {
...launched,
cleanup: async () => {
const stopped = await cleanup();
if (!mayClean) throw new Error("injected retained cleanup failure");
return stopped;
},
};
},
async () => {
const root = await PrivateRuntimeRoot.create({
prefix: "rea-evm-mixed-cleanup-",
});
ownedRoot = root.path;
const close = root.close.bind(root);
return {
path: root.path,
close: async () => {
if (denyRootClose) {
denyRootClose = false;
throw new Error("injected retained root cleanup failure");
}
await close();
},
};
},
);
onTestFinished(async () => {
mayClean = true;
await provider.close();
@@ -282,7 +303,20 @@ it.runIf(!unsupportedHost).each(["completed", "malformed-reply"] as const)(
expect(blocked.ok).toBe(false);
expect(launches).toBe(1);
mayClean = true;
await expect(provider.close()).rejects.toMatchObject({
cleanupIncomplete: true,
cleanupResources: [ownedRoot],
});
expect(result.error.cleanupIncomplete).toBe(true);
expect(result.error.cleanupResources).toContain(ownedRoot);
await provider.close();
expect(result.error.cleanupIncomplete).toBe(false);
expect(result.error.cleanupResources).toEqual([]);
expect(projectAnalysisError(result.error).code).toBe(
scenario === "malformed-reply"
? "unreadable_output"
: "execution_failure",
);
await expect(access(ownedRoot)).rejects.toMatchObject({ code: "ENOENT" });
},
);
@@ -747,6 +781,13 @@ it
if (scenario === "late-cancellation")
expect(result.error.cause).toBe(reason);
else expect(result.error.cleanupIncomplete).toBe(true);
if (scenario === "cancelled-cleanup-failure") {
expect(result.error.cleanupPrimary?.cause).toBe(reason);
await provider.close();
expect(result.error.cleanupIncomplete).toBe(false);
expect(projectAnalysisError(result.error).category).toBe("cancelled");
expect(result.error.cleanupPrimary?.cause).toBe(reason);
}
const forwarded = await new EvmInterfaceService({
inspect: async () => result,
}).inspect({ path, encoding: "hex" });