mirror/rea
1
0
Fork 0
mirror of https://github.com/morluto/rea synced 2026-10-26 14:25:49 -04:00
Code Issues Packages Projects Releases Wiki Activity

refactor(process): share ownership for isolated analysis workers

Browse Source 
Separate child and private-root ownership from JavaScript analysis protocols. Retain exact cleanup lifetimes across retries, late handshakes and runtime reuse; observe request-stream errors and preserve per-job file cleanup during worker reuse.
This commit is contained in:
morluto
2026-10-19 13:12:25 +08:00
parent e4c3d72396
commit 715d93e954
4 changed files with 774 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/javascript/analysis/JavaScriptAnalysisWorker.ts
+126 -110
View File
@@ -28,14 +28,13 @@ import { ProviderAdapterError } from "../../domain/providerAdapterError.js";
import { ProviderCleanupError } from "../../domain/providerCleanupError.js";
import { ok, err, type Result } from "../../domain/result.js";
import { javaScriptRuntime } from "../../domain/runtimeVersion.js";
import { PrivateRuntimeRoot } from "../../process/PrivateRuntimeRoot.js";
import { PrivateRuntimeRootUnavailableError } from "../../process/PrivateRuntimeRootUnavailableError.js";
import { javaScriptWorkerMemoryLimitBytes } from "../../process/JavaScriptWorkerMemoryLimit.js";
import { spawnOwnedProviderProcess } from "../../process/ProviderProcess.js";
import {
ProviderProcessSupervisor,
spawnOwnedProviderProcess,
} from "../../process/ProviderProcess.js";
import { cleanupOwnedProcessGroup } from "../../process/ProcessOwnership.js";
OwnedWorkerRuntime,
OwnedWorkerRuntimeCleanupError,
} from "../../process/OwnedWorkerRuntime.js";
import { prepareProcessOwnershipInspection } from "../../process/ProcessOwnershipObservation.js";
import {
ProviderStartupDeadline,
@@ -59,7 +58,6 @@ const OPERATION = "analyze_javascript_application";
const PROVIDER = "javascript-static-analysis";
const MIB = 1024 * 1024;
const STATIC_CHECKPOINT_RETENTION_EXPANSION = 6;
const pendingCleanup = new Set<JavaScriptAnalysisWorker>();
/** The worker runs on this executable; Bun accepts but ignores V8 heap flags. */
const WORKER_HEAP_FLAGS_ENFORCED =
javaScriptRuntime(process.versions).id === "node";
@@ -67,8 +65,7 @@ const WORKER_HEAP_FLAGS_ENFORCED =
/** One sequential owned worker, retained between source files and closed after analysis. */
export class JavaScriptAnalysisWorker {
readonly #runId = `rea-javascript-analysis-${randomUUID()}`;
#root: PrivateRuntimeRoot | undefined;
#supervisor: ProviderProcessSupervisor | undefined;
readonly #runtime: OwnedWorkerRuntime;
#ready = false;
#heapLimitBytes = 0;
#frames: JavaScriptWorkerMessage[] = [];
@@ -84,7 +81,6 @@ export class JavaScriptAnalysisWorker {
readonly #resultBudgetBytes: number;
/** Null when the runtime enforces the V8 heap flags instead. */
readonly #processMemoryLimitBytes: number | null;
readonly #onStdinError = (): void => {};
readonly #onStdout = (chunk: Buffer | string): void => {
try {
this.#pending +=
@@ -117,6 +113,14 @@ export class JavaScriptAnalysisWorker {
readonly launcher: typeof spawnOwnedProviderProcess = spawnOwnedProviderProcess,
sourceCount?: number,
) {
this.#runtime = new OwnedWorkerRuntime({
rootPrefix: "rea-javascript-analysis-",
launcher,
supervisorOptions: {
captureStdout: false,
maxDiagnosticBytes: 256 * 1024,
},
});
if (
!Number.isSafeInteger(limits.heapMb) ||
limits.heapMb < 128 ||
@@ -181,12 +185,12 @@ export class JavaScriptAnalysisWorker {
);
// Host ownership preparation is outside the parser execution deadline.
await prepareProcessOwnershipInspection(signal);
for (const owner of pendingCleanup) await owner.close();
await OwnedWorkerRuntime.retryPending();
deadline = new ProviderStartupDeadline(this.limits.timeoutMs, signal);
await this.#start(deadline);
const root = this.#root;
const supervisor = this.#supervisor;
if (root === undefined || supervisor === undefined)
const rootPath = this.#runtime.rootPath;
const supervisor = this.#runtime.supervisor;
if (rootPath === undefined || supervisor === undefined)
throw new TypeError("JavaScript worker resources were not acquired");
if (this.#frames.length !== 0 || this.#pending !== "")
throw new AnalysisOutputError(
@@ -195,7 +199,7 @@ export class JavaScriptAnalysisWorker {
);
supervisor.resetOutput();
this.#phase = "source-transfer";
await writeFile(join(root.path, `${id}.source`), file.text.value, {
await writeFile(join(rootPath, `${id}.source`), file.text.value, {
flag: "wx",
mode: 0o600,
signal: deadline.signal,
@@ -342,7 +346,7 @@ export class JavaScriptAnalysisWorker {
const retainGraph = moduleEstimate + graphEstimate <= remaining;
this.#phase = "result-admission";
const decoded = await readJavaScriptAnalysisTransfer(
join(root.path, `${id}.records`),
join(rootPath, `${id}.records`),
descriptor,
file,
deadline.signal,
@@ -390,7 +394,7 @@ export class JavaScriptAnalysisWorker {
if (
cleanup === undefined &&
javascript === null &&
this.#root !== undefined
this.#runtime.rootPath !== undefined
) {
try {
javascript = await this.#readCheckpoint(id, file.path);
@@ -407,17 +411,20 @@ export class JavaScriptAnalysisWorker {
this.#lastFailure = result.error.error;
// Child stop must be verified before temporary files are released. On
// cleanup uncertainty, close retains this owner for a later retry.
if (this.#root !== undefined && !pendingCleanup.has(this)) {
const rootPath = this.#runtime.rootPath;
if (rootPath !== undefined && !this.#runtime.hasPendingCleanup) {
try {
for (const suffix of ["source", "static.json", "records"])
await rm(join(this.#root.path, `${id}.${suffix}`), { force: true });
await rm(join(rootPath, `${id}.${suffix}`), { force: true });
} catch (cause: unknown) {
pendingCleanup.add(this);
this.#runtime.retainForCleanup();
const error = this.#cleanupError(message(cause));
this.#runtime.bindCleanupFailure(error);
return err({
javascript,
module,
projection,
error: this.#cleanupError(message(cause)),
error,
});
}
}
@@ -436,75 +443,70 @@ export class JavaScriptAnalysisWorker {
/** Verify process cleanup, then release the private runtime; failed close is retryable. */
async close(): Promise<void> {
const failure = await this.#stop();
if (failure !== undefined) throw failure;
if (this.#root !== undefined) {
try {
await this.#root.close();
this.#root = undefined;
} catch (cause: unknown) {
pendingCleanup.add(this);
throw this.#cleanupError(message(cause));
try {
await this.#runtime.close();
this.#ready = false;
this.#resetState();
} catch (cause: unknown) {
if (cause instanceof OwnedWorkerRuntimeCleanupError) {
const error = this.#cleanupError(
cause.message,
this.#lastFailure,
cause.cleanupResources,
cause.processId,
);
cause.bindCleanupFailure(error);
throw error;
}
throw cause;
}
pendingCleanup.delete(this);
}
async #start(deadline: ProviderStartupDeadline): Promise<void> {
if (this.#supervisor !== undefined && this.#ready) return;
if (this.#runtime.supervisor !== undefined && this.#ready) return;
this.#phase = "startup";
this.#heapLimitBytes = 0;
this.#root ??= await PrivateRuntimeRoot.create({
prefix: "rea-javascript-analysis-",
});
const environment: NodeJS.ProcessEnv = {
...process.env,
UV_THREADPOOL_SIZE: "1",
};
delete environment.NODE_OPTIONS;
const spawned = await this.launcher({
command: process.execPath,
arguments: [
`--max-old-space-size=${String(this.limits.heapMb)}`,
"--max-semi-space-size=8",
"--v8-pool-size=1",
"--expose-gc",
"--stack_size=2048",
fileURLToPath(import.meta.resolve("#javascript-analysis-worker")),
],
runId: `${this.#runId}-${randomUUID()}`,
cwd: this.#root.path,
hostEnvironment: environment,
stdin: "pipe",
...(this.#processMemoryLimitBytes === null
? {}
: { memoryLimitBytes: this.#processMemoryLimitBytes }),
signal: deadline.signal,
});
this.#supervisor = new ProviderProcessSupervisor(
await this.#runtime.start(
{
...spawned,
ownsProcessLifetime: true,
cleanup:
spawned.cleanup ??
(() => cleanupOwnedProcessGroup(spawned.ownership)),
command: process.execPath,
arguments: [
`--max-old-space-size=${String(this.limits.heapMb)}`,
"--max-semi-space-size=8",
"--v8-pool-size=1",
"--expose-gc",
"--stack_size=2048",
fileURLToPath(import.meta.resolve("#javascript-analysis-worker")),
],
runId: `${this.#runId}-${randomUUID()}`,
hostEnvironment: environment,
stdin: "pipe",
...(this.#processMemoryLimitBytes === null
? {}
: { memoryLimitBytes: this.#processMemoryLimitBytes }),
signal: deadline.signal,
},
async () => {
this.#runtime.onStdout(this.#onStdout);
const ready = await this.#next(deadline, "worker startup");
if (ready.kind !== "ready")
throw new AnalysisOutputError(
OPERATION,
"JavaScript worker emitted facts before its startup handshake",
);
if ((ready.heap_limit_bytes === null) === WORKER_HEAP_FLAGS_ENFORCED)
throw new AnalysisOutputError(
OPERATION,
"JavaScript worker heap limit does not match the runtime that launched it",
);
this.#heapLimitBytes = ready.heap_limit_bytes ?? 0;
this.#ready = true;
},
{ captureStdout: false, maxDiagnosticBytes: 256 * 1024 },
);
spawned.process.stdout?.on("data", this.#onStdout);
const ready = await this.#next(deadline, "worker startup");
if (ready.kind !== "ready")
throw new AnalysisOutputError(
OPERATION,
"JavaScript worker emitted facts before its startup handshake",
);
if ((ready.heap_limit_bytes === null) === WORKER_HEAP_FLAGS_ENFORCED)
throw new AnalysisOutputError(
OPERATION,
"JavaScript worker heap limit does not match the runtime that launched it",
);
this.#heapLimitBytes = ready.heap_limit_bytes ?? 0;
this.#ready = true;
}
async #next(
@@ -516,19 +518,27 @@ export class JavaScriptAnalysisWorker {
throw new AnalysisCancelledError(OPERATION);
if (deadline.interruption === "timeout")
throw new AnalysisTimeoutError(OPERATION, this.limits.timeoutMs);
const inputFailure = this.#runtime.inputFailure;
if (inputFailure !== undefined)
throw new AnalysisOutputError(
OPERATION,
`Owned JavaScript worker request stream failed during ${path}`,
{ cause: inputFailure, capturedOutput: this.#captured() },
);
if (this.#protocolFailure !== undefined)
throw new AnalysisOutputError(OPERATION, this.#protocolFailure);
const frame = this.#frames.shift();
if (frame !== undefined) return frame;
const snapshot = this.#supervisor?.snapshot();
const supervisor = this.#runtime.supervisor;
const snapshot = supervisor?.snapshot();
if (
snapshot !== undefined &&
(snapshot.exitCode !== undefined || snapshot.signal !== undefined)
) {
// Exit is not complete output: let any final control frame drain first.
if (!(await this.#supervisor?.waitForOutputClose(10))) continue;
if (!(await supervisor?.waitForOutputClose(10))) continue;
if (this.#frames.length !== 0) continue;
const drained = this.#supervisor?.snapshot() ?? snapshot;
const drained = supervisor?.snapshot() ?? snapshot;
throw this.#memoryLimitReached()
? this.#memoryFailure(path)
: /heap out of memory|heap limit|allocation failed/iu.test(
@@ -560,9 +570,10 @@ export class JavaScriptAnalysisWorker {
sourcePath: string,
expected?: { readonly bytes: number; readonly sha256: string },
): Promise<JavaScriptStaticAnalysis> {
if (this.#root === undefined)
const rootPath = this.#runtime.rootPath;
if (rootPath === undefined)
throw new TypeError("JavaScript checkpoint has no owned runtime");
const path = join(this.#root.path, `${id}.static.json`);
const path = join(rootPath, `${id}.static.json`);
const bytes = (await stat(path)).size;
if (
bytes > DEFAULTS.checkpointBytes ||
@@ -645,7 +656,7 @@ export class JavaScriptAnalysisWorker {
};
}
#memoryLimitReached(): boolean {
return this.#supervisor?.launch.process.memoryLimitReached === true;
return this.#runtime.supervisor?.launch.process.memoryLimitReached === true;
}
#memoryFailure(path: string): AnalysisResourceConstraintError {
return new AnalysisResourceConstraintError(
@@ -683,7 +694,7 @@ export class JavaScriptAnalysisWorker {
);
}
#captured() {
const snapshot = this.#supervisor?.snapshot();
const snapshot = this.#runtime.supervisor?.snapshot();
return {
stdout: "",
stderr: snapshot?.stderr.text ?? "",
@@ -699,12 +710,23 @@ export class JavaScriptAnalysisWorker {
deadline: ProviderStartupDeadline | undefined,
signal: AbortSignal | undefined,
): AnalysisError {
if (cause instanceof OwnedWorkerRuntimeCleanupError) {
const error = this.#cleanupError(
cause.message,
this.#lastFailure,
cause.cleanupResources,
cause.processId,
);
cause.bindCleanupFailure(error);
return error;
}
// Ownership preparation is abortable before the worker deadline exists.
if (
deadline?.interruption === "cancelled" ||
(deadline === undefined &&
signal?.aborted === true &&
!(cause instanceof ProviderCleanupError))
!(cause instanceof ProviderCleanupError) &&
!(cause instanceof OwnedWorkerRuntimeCleanupError))
)
return new AnalysisCancelledError(OPERATION, {
capturedOutput: this.#captured(),
@@ -713,6 +735,13 @@ export class JavaScriptAnalysisWorker {
return new AnalysisTimeoutError(OPERATION, this.limits.timeoutMs, {
capturedOutput: this.#captured(),
});
const inputFailure = this.#runtime.inputFailure;
if (inputFailure !== undefined)
return new AnalysisOutputError(
OPERATION,
`Owned JavaScript worker request stream failed during ${this.#phase}`,
{ cause: inputFailure, capturedOutput: this.#captured() },
);
// A reached process limit explains the worker's exit, broken pipe or
// truncated frame better than the symptom it produced.
if (
@@ -751,13 +780,19 @@ export class JavaScriptAnalysisWorker {
#cleanupError(
reason: string,
original: AnalysisError | undefined = this.#lastFailure,
cleanupResources: readonly string[] = [
this.#runId,
...(this.#runtime.rootPath === undefined ? [] : [this.#runtime.rootPath]),
],
processId: number | null = this.#runtime.supervisor?.launch.process.pid ??
null,
): ProviderCleanupError {
return new ProviderCleanupError(
PROVIDER,
[this.#runId, ...(this.#root === undefined ? [] : [this.#root.path])],
cleanupResources,
{
reason,
worker_pid: this.#supervisor?.launch.process.pid ?? null,
worker_pid: processId,
...(original === undefined
? {}
: {
@@ -784,33 +819,14 @@ export class JavaScriptAnalysisWorker {
async #stop(
original: AnalysisError | undefined = this.#lastFailure,
): Promise<ProviderCleanupError | undefined> {
if (this.#supervisor !== undefined) {
const supervisor = this.#supervisor;
const stopped = await supervisor.stop();
supervisor.launch.process.stdout?.off("data", this.#onStdout);
if (stopped.status === "incomplete") {
try {
const stdin = supervisor.launch.process.stdin;
if (
stdin !== undefined &&
stdin !== null &&
!stdin.destroyed &&
!stdin.writableEnded
) {
stdin.off("error", this.#onStdinError);
stdin.once("error", this.#onStdinError);
stdin.end();
}
} catch {
// Cooperative protocol stdin may already be closed or broken.
}
this.#resetState();
pendingCleanup.add(this);
return this.#cleanupError(stopped.reason, original);
}
supervisor.dispose();
this.#supervisor = undefined;
const stopped = await this.#runtime.stop();
if (stopped.status === "incomplete") {
this.#resetState();
const error = this.#cleanupError(stopped.reason, original);
this.#runtime.bindCleanupFailure(error);
return error;
}
this.#ready = false;
this.#resetState();
return undefined;
}