From 715d93e954f238e009d9a425b6c430ce131eb51b Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Mon, 19 Oct 2026 13:12:25 +0800 Subject: [PATCH] refactor(process): share ownership for isolated analysis workers 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. --- .../analysis/JavaScriptAnalysisWorker.ts | 236 ++++++----- src/process/OwnedWorkerRuntime.test.ts | 245 +++++++++++ src/process/OwnedWorkerRuntime.ts | 391 ++++++++++++++++++ .../javascriptSourceAnalysisWorker.test.ts | 13 +- 4 files changed, 774 insertions(+), 111 deletions(-) create mode 100644 src/process/OwnedWorkerRuntime.test.ts create mode 100644 src/process/OwnedWorkerRuntime.ts diff --git a/src/javascript/analysis/JavaScriptAnalysisWorker.ts b/src/javascript/analysis/JavaScriptAnalysisWorker.ts index 5a96f4a92a..c6bf54021b 100644 --- a/src/javascript/analysis/JavaScriptAnalysisWorker.ts +++ b/src/javascript/analysis/JavaScriptAnalysisWorker.ts @@ -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(); /** 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 { - 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 { - 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 { - 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 { - 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; } diff --git a/src/process/OwnedWorkerRuntime.test.ts b/src/process/OwnedWorkerRuntime.test.ts new file mode 100644 index 0000000000..4f8eeb311f --- /dev/null +++ b/src/process/OwnedWorkerRuntime.test.ts @@ -0,0 +1,245 @@ +import { access } from "node:fs/promises"; +import { randomUUID } from "node:crypto"; +import { expect, it } from "vitest"; + +import { ProviderCleanupError } from "../domain/providerCleanupError.js"; +import { cleanupOwnedProcessGroup } from "./ProcessOwnership.js"; +import { + OwnedWorkerRuntime, + OwnedWorkerRuntimeCleanupError, + type OwnedWorkerRuntimeOptions, +} from "./OwnedWorkerRuntime.js"; +import { spawnOwnedProviderProcess } from "./ProviderProcess.js"; + +const runtimeOptions = ( + overrides: Partial = {}, +): OwnedWorkerRuntimeOptions => ({ + rootPrefix: "rea-owned-worker-test-", + ...overrides, +}); + +const startOptions = () => ({ + command: process.execPath, + arguments: ["-e", "process.stdin.resume(); setInterval(() => {}, 1000);"], + runId: `rea-owned-worker-test-${randomUUID()}`, + stdin: "pipe" as const, +}); + +const waitFor = async (predicate: () => boolean): Promise => { + for (let attempt = 0; attempt < 100; attempt++) { + if (predicate()) return; + await new Promise((resolve) => setTimeout(resolve, 5)); + } + throw new Error("Timed out waiting for owned worker startup"); +}; + +it("waits for an admitted late handshake before closing its process and root", async () => { + const runtime = new OwnedWorkerRuntime(runtimeOptions()); + let announceInitialize: () => void = () => {}; + const initializing = new Promise((resolve) => { + announceInitialize = resolve; + }); + let releaseInitialize: () => void = () => {}; + const held = new Promise((resolve) => { + releaseInitialize = resolve; + }); + let closeSettled = false; + const starting = runtime.start(startOptions(), async () => { + announceInitialize(); + await held; + }); + await initializing; + await waitFor(() => runtime.supervisor !== undefined); + const rootPath = runtime.rootPath; + expect(rootPath).toBeDefined(); + const closing = runtime.close().then(() => { + closeSettled = true; + }); + await new Promise((resolve) => setImmediate(resolve)); + expect(closeSettled).toBe(false); + expect(runtime.supervisor).toBeDefined(); + releaseInitialize(); + await expect(starting).rejects.toThrow("closing"); + await closing; + expect(runtime.supervisor).toBeUndefined(); + expect(runtime.rootPath).toBeUndefined(); + await expect(access(rootPath as string)).rejects.toThrow(); +}); + +it("resolves a bound cleanup error after shared pending-owner retry succeeds", async () => { + let failCleanup = true; + const runtime = new OwnedWorkerRuntime( + runtimeOptions({ + launcher: async (options) => { + const spawned = await spawnOwnedProviderProcess(options); + return { + ...spawned, + cleanup: async () => { + if (failCleanup) { + failCleanup = false; + return { cleaned: false, reason: "controlled cleanup retry" }; + } + return spawned.cleanup === undefined + ? cleanupOwnedProcessGroup(spawned.ownership) + : spawned.cleanup(); + }, + }; + }, + }), + ); + await runtime.start(startOptions()); + const rootPath = runtime.rootPath; + const error = new ProviderCleanupError("owned-worker-test", [], { + reason: "controlled cleanup retry", + }); + await expect(runtime.stop()).resolves.toMatchObject({ status: "incomplete" }); + runtime.bindCleanupFailure(error); + expect(error.cleanupIncomplete).toBe(true); + expect(error.cleanupResources).toContain(rootPath); + + await OwnedWorkerRuntime.retryPending(); + + expect(error.cleanupIncomplete).toBe(false); + expect(error.cleanupResources).toEqual([]); + expect(runtime.rootPath).toBeUndefined(); + await expect(access(rootPath as string)).rejects.toThrow(); +}); + +it("observes request-stream errors and resets them for a fresh child", async () => { + const runtime = new OwnedWorkerRuntime(runtimeOptions()); + try { + await runtime.start(startOptions()); + const stdin = runtime.supervisor?.launch.process.stdin; + expect(stdin).toBeDefined(); + const failure = new Error("controlled request pipe failure"); + stdin?.emit("error", failure); + expect(runtime.inputFailure).toBe(failure); + + const stopped = await runtime.stop(); + expect(stopped.status).not.toBe("incomplete"); + await runtime.start(startOptions()); + expect(runtime.inputFailure).toBeUndefined(); + } finally { + await runtime.close(); + } +}); + +it("retains pending ownership when best-effort stdin.end throws after an incomplete stop", async () => { + let failCleanup = true; + const runtime = new OwnedWorkerRuntime( + runtimeOptions({ + launcher: async (options) => { + const spawned = await spawnOwnedProviderProcess(options); + return { + ...spawned, + cleanup: async () => { + if (failCleanup) + return { cleaned: false, reason: "controlled incomplete stop" }; + return spawned.cleanup === undefined + ? cleanupOwnedProcessGroup(spawned.ownership) + : spawned.cleanup(); + }, + }; + }, + }), + ); + await runtime.start(startOptions()); + const rootPath = runtime.rootPath; + const stdin = runtime.supervisor?.launch.process.stdin; + if (stdin === undefined || stdin === null) + throw new Error("Owned child request stream is unavailable"); + const originalEnd = stdin.end.bind(stdin); + Object.defineProperty(stdin, "end", { + configurable: true, + value: () => { + throw new Error("controlled stdin.end failure"); + }, + }); + try { + await expect(runtime.stop()).resolves.toMatchObject({ + status: "incomplete", + }); + expect(runtime.hasPendingCleanup).toBe(true); + failCleanup = false; + Object.defineProperty(stdin, "end", { + configurable: true, + value: originalEnd, + }); + + await OwnedWorkerRuntime.retryPending(); + + expect(runtime.hasPendingCleanup).toBe(false); + expect(runtime.rootPath).toBeUndefined(); + await expect(access(rootPath as string)).rejects.toThrow(); + } finally { + Object.defineProperty(stdin, "end", { + configurable: true, + value: originalEnd, + }); + failCleanup = false; + await runtime.close(); + } +}); + +it("binds a late cleanup projection to its released root epoch after runtime reuse", async () => { + let failCleanup = true; + const runtime = new OwnedWorkerRuntime( + runtimeOptions({ + launcher: async (options) => { + const spawned = await spawnOwnedProviderProcess(options); + return { + ...spawned, + cleanup: async () => { + if (failCleanup) + return { cleaned: false, reason: "controlled old-epoch cleanup" }; + return spawned.cleanup === undefined + ? cleanupOwnedProcessGroup(spawned.ownership) + : spawned.cleanup(); + }, + }; + }, + }), + ); + let oldCleanupError: OwnedWorkerRuntimeCleanupError | undefined; + try { + await runtime.start(startOptions()); + const oldRoot = runtime.rootPath; + await expect(runtime.stop()).resolves.toMatchObject({ + status: "incomplete", + }); + try { + await OwnedWorkerRuntime.retryPending(); + throw new Error("Expected pending-owner retry to fail"); + } catch (cause) { + if (!(cause instanceof OwnedWorkerRuntimeCleanupError)) throw cause; + oldCleanupError = cause; + } + if (oldCleanupError === undefined) + throw new Error("Expected an owned cleanup error from the old root"); + + failCleanup = false; + await OwnedWorkerRuntime.retryPending(); + expect(runtime.rootPath).toBeUndefined(); + await expect(access(oldRoot as string)).rejects.toThrow(); + + await runtime.start(startOptions()); + const newRoot = runtime.rootPath; + expect(newRoot).toBeDefined(); + expect(newRoot).not.toBe(oldRoot); + + const lateProjection = new ProviderCleanupError( + "owned-worker-test", + oldCleanupError.cleanupResources, + { reason: oldCleanupError.message }, + ); + oldCleanupError.bindCleanupFailure(lateProjection); + + expect(lateProjection.cleanupIncomplete).toBe(false); + expect(lateProjection.cleanupResources).toEqual([]); + expect(runtime.rootPath).toBe(newRoot); + expect(runtime.supervisor).toBeDefined(); + } finally { + failCleanup = false; + await runtime.close(); + } +}); diff --git a/src/process/OwnedWorkerRuntime.ts b/src/process/OwnedWorkerRuntime.ts new file mode 100644 index 0000000000..150ecc01ee --- /dev/null +++ b/src/process/OwnedWorkerRuntime.ts @@ -0,0 +1,391 @@ +import { PrivateRuntimeRoot } from "./PrivateRuntimeRoot.js"; +import type { Readable, Writable } from "node:stream"; +import { + ProviderProcessSupervisor, + spawnOwnedProviderProcess, + type OwnedProviderProcessSpawnOptions, + type ProviderProcessSupervisorOptions, + type ProviderProcessStopResult, +} from "./ProviderProcess.js"; +import { cleanupOwnedProcessGroup } from "./ProcessOwnership.js"; +import type { AnalysisError } from "../domain/analysisErrorBase.js"; + +export interface OwnedWorkerRuntimeOptions { + readonly rootPrefix: string; + readonly launcher?: typeof spawnOwnedProviderProcess; + readonly supervisorOptions?: ProviderProcessSupervisorOptions; +} + +export class OwnedWorkerRuntimeCleanupError extends Error { + readonly #cleanupEpoch: OwnedWorkerRuntimeCleanupEpoch | undefined; + + constructor( + readonly cleanupResources: readonly string[], + readonly processId: number | null, + message: string, + cleanupEpoch: OwnedWorkerRuntimeCleanupEpoch | undefined, + options?: ErrorOptions, + ) { + super(message, options); + this.name = "OwnedWorkerRuntimeCleanupError"; + this.#cleanupEpoch = cleanupEpoch; + } + + bindCleanupFailure(error: OwnedWorkerCleanupFailure | AnalysisError): void { + if (this.#cleanupEpoch === undefined) { + error.bindCleanupResourceOwner(releasedCleanupOwner); + error.resolveCleanup(); + return; + } + this.#cleanupEpoch.bind(error); + } +} + +export interface OwnedWorkerCleanupFailure { + bindCleanupResourceOwner(owner: { + readonly cleanupResources: readonly string[] | undefined; + }): unknown; + resolveCleanup(): unknown; +} + +/** A root's cleanup attribution is stable even after this runtime is reused. */ +class OwnedWorkerRuntimeCleanupEpoch { + #released = false; + #runId: string | undefined; + #failures: WeakRef[] = []; + + constructor(readonly rootPath: string) {} + + get cleanupResources(): readonly string[] { + return this.#released + ? [] + : [...(this.#runId === undefined ? [] : [this.#runId]), this.rootPath]; + } + + setRunId(runId: string | undefined): void { + if (!this.#released) this.#runId = runId; + } + + bind(error: OwnedWorkerCleanupFailure | AnalysisError): void { + error.bindCleanupResourceOwner(this); + if (this.#released) { + error.resolveCleanup(); + return; + } + this.#failures = [ + ...this.#failures.filter((reference) => reference.deref() !== undefined), + new WeakRef(error), + ]; + } + + release(): void { + if (this.#released) return; + this.#released = true; + for (const reference of this.#failures) reference.deref()?.resolveCleanup(); + this.#failures = []; + } +} + +/** Own the private root and process lifecycle shared by sequential workers. */ +export class OwnedWorkerRuntime { + static readonly #pending = new Set(); + + #root: PrivateRuntimeRoot | undefined; + #cleanupEpoch: OwnedWorkerRuntimeCleanupEpoch | undefined; + #supervisor: ProviderProcessSupervisor | undefined; + #processId: number | null = null; + #acquisition: Promise | undefined; + #stopAttempt: Promise | undefined; + #closeAttempt: Promise | undefined; + #closing = false; + #closed = false; + #lastRunId: string | undefined; + #terminalCloseRequested = false; + #inputFailure: Error | undefined; + #stdinGeneration = 0; + readonly #stdoutListeners = new Set<{ + readonly stream: Readable; + readonly listener: (chunk: Buffer | string) => void; + }>(); + + constructor(readonly options: OwnedWorkerRuntimeOptions) {} + + get rootPath(): string | undefined { + return this.#root?.path; + } + + /** Exposes the exact supervisor so consumers retain their protocol policy. */ + get supervisor(): ProviderProcessSupervisor | undefined { + return this.#supervisor; + } + + get cleanupResources(): readonly string[] | undefined { + const resources = [ + ...(this.#supervisor === undefined + ? [] + : [this.#supervisor.launch.ownership?.runId ?? this.#lastRunId]), + ...(this.#root === undefined ? [] : [this.#root.path]), + ].filter((resource): resource is string => resource !== undefined); + return resources.length === 0 ? [] : resources; + } + + /** Whether this exact owner still retains an unconfirmed resource release. */ + get hasPendingCleanup(): boolean { + return OwnedWorkerRuntime.#pending.has(this); + } + + /** First error observed on the current child's request stream, if any. */ + get inputFailure(): Error | undefined { + return this.#inputFailure; + } + + /** Keep a returned typed cleanup error live until this exact owner releases. */ + bindCleanupFailure(error: OwnedWorkerCleanupFailure | AnalysisError): void { + if (this.#cleanupEpoch === undefined) { + error.bindCleanupResourceOwner(releasedCleanupOwner); + error.resolveCleanup(); + return; + } + this.#cleanupEpoch.bind(error); + } + + /** Mark this still-owned runtime for retry before any consumer starts a child. */ + retainForCleanup(): void { + OwnedWorkerRuntime.#pending.add(this); + } + + /** Retry every earlier worker owner before any consumer admits a new child. */ + static async retryPending(except?: OwnedWorkerRuntime): Promise { + for (const owner of [...this.#pending]) + if (owner !== except) await owner.#retryPending(); + } + + /** + * Acquire one child and let the protocol owner complete its own handshake. + * Close waits for this entire callback, including a late handshake. + */ + start( + spawnOptions: Omit, + initialize?: (supervisor: ProviderProcessSupervisor) => Promise, + ): Promise { + if (this.#terminalCloseRequested || this.#closing) + return Promise.reject(new Error("Owned worker runtime is closing")); + if (this.#acquisition !== undefined) return this.#acquisition; + if (this.#supervisor !== undefined) return Promise.resolve(); + const ownRetry = OwnedWorkerRuntime.#pending.has(this) + ? this.#retryPending() + : Promise.resolve(); + const acquisition = ownRetry.then(async () => { + if (this.#closed) this.#closed = false; + await this.#start(spawnOptions, initialize); + }); + this.#acquisition = acquisition; + void acquisition + .finally(() => { + if (this.#acquisition === acquisition) this.#acquisition = undefined; + }) + .catch(() => undefined); + return acquisition; + } + + /** Attach protocol stdout while ensuring it is detached before process stop. */ + onStdout(listener: (chunk: Buffer | string) => void): void { + const stream = this.#supervisor?.launch.process.stdout; + if (stream === null || stream === undefined) + throw new Error("Owned worker runtime has no stdout stream"); + stream.on("data", listener); + this.#stdoutListeners.add({ stream, listener }); + } + + stop(): Promise { + if (this.#stopAttempt !== undefined) return this.#stopAttempt; + const attempt = this.#stop(); + this.#stopAttempt = attempt; + void attempt + .finally(() => { + if (this.#stopAttempt === attempt) this.#stopAttempt = undefined; + }) + .catch(() => undefined); + return attempt; + } + + /** Stop the exact child before releasing its root; every failure is retryable. */ + close(): Promise { + this.#terminalCloseRequested = true; + this.#closing = true; + if (this.#closed) return Promise.resolve(); + if (this.#closeAttempt !== undefined) return this.#closeAttempt; + const attempt = this.#close(); + this.#closeAttempt = attempt; + void attempt + .finally(() => { + if (this.#closeAttempt === attempt) this.#closeAttempt = undefined; + }) + .catch(() => undefined); + return attempt; + } + + async #retryPending(): Promise { + if (this.#terminalCloseRequested) return this.close(); + this.#closing = true; + if (this.#closeAttempt === undefined) { + const attempt = this.#close(); + this.#closeAttempt = attempt; + void attempt + .finally(() => { + if (this.#closeAttempt === attempt) this.#closeAttempt = undefined; + }) + .catch(() => undefined); + } + const attempt = this.#closeAttempt; + if (attempt === undefined) return; + try { + await attempt; + if (!this.#terminalCloseRequested) { + this.#closed = false; + this.#closing = false; + } + } catch (cause: unknown) { + if (!this.#terminalCloseRequested) this.#closing = false; + throw cause; + } + } + + async #start( + spawnOptions: Omit, + initialize: + | ((supervisor: ProviderProcessSupervisor) => Promise) + | undefined, + ): Promise { + await OwnedWorkerRuntime.retryPending(this); + if (this.#closing) throw new Error("Owned worker runtime is closing"); + this.#root ??= await PrivateRuntimeRoot.create({ + prefix: this.options.rootPrefix, + }); + this.#cleanupEpoch ??= new OwnedWorkerRuntimeCleanupEpoch(this.#root.path); + if (this.#closing) throw new Error("Owned worker runtime is closing"); + this.#lastRunId = spawnOptions.runId; + const spawned = await (this.options.launcher ?? spawnOwnedProviderProcess)({ + ...spawnOptions, + cwd: this.#root.path, + }); + this.#inputFailure = undefined; + this.#observeStdin(spawned.process.stdin); + this.#processId = spawned.process.pid ?? null; + this.#cleanupEpoch.setRunId(spawned.ownership?.runId ?? spawnOptions.runId); + this.#supervisor = new ProviderProcessSupervisor( + { + ...spawned, + ownsProcessLifetime: true, + cleanup: + spawned.cleanup ?? + (() => cleanupOwnedProcessGroup(spawned.ownership)), + }, + this.options.supervisorOptions, + ); + // A close may have begun while spawn was pending. Adopt the process first; + // close waits for this acquisition and then stops the exact adopted owner. + if (this.#closing) throw new Error("Owned worker runtime is closing"); + await initialize?.(this.#supervisor); + if (this.#closing) throw new Error("Owned worker runtime is closing"); + } + + async #stop(): Promise { + const acquisition = this.#acquisition; + if (acquisition !== undefined) await acquisition.catch(() => undefined); + const supervisor = this.#supervisor; + if (supervisor === undefined) return { status: "already-exited" }; + this.#detachStdout(); + const stopped = await supervisor.stop(); + if (stopped.status === "incomplete") { + OwnedWorkerRuntime.#pending.add(this); + const stdin = supervisor.launch.process.stdin; + if ( + stdin !== undefined && + stdin !== null && + !stdin.destroyed && + !stdin.writableEnded + ) { + try { + stdin.end(); + } catch { + // Closing stdin is cooperative only; retain the original stop failure. + } + } + return stopped; + } + supervisor.dispose(); + this.#supervisor = undefined; + this.#processId = null; + this.#cleanupEpoch?.setRunId(undefined); + return stopped; + } + + async #close(): Promise { + const acquisition = this.#acquisition; + if (acquisition !== undefined) await acquisition.catch(() => undefined); + const stopped = await this.stop(); + if (stopped.status === "incomplete") + throw this.#cleanupError(stopped.reason); + if (this.#root !== undefined) { + try { + await this.#root.close(); + this.#root = undefined; + this.#cleanupEpoch?.release(); + this.#cleanupEpoch = undefined; + } catch (cause: unknown) { + OwnedWorkerRuntime.#pending.add(this); + throw this.#cleanupError(message(cause), cause); + } + } + this.#closed = true; + OwnedWorkerRuntime.#pending.delete(this); + } + + #detachStdout(): void { + for (const { stream, listener } of this.#stdoutListeners) + stream.off("data", listener); + this.#stdoutListeners.clear(); + } + + #observeStdin(stream: Writable | null | undefined): void { + if (stream === null || stream === undefined) return; + const generation = ++this.#stdinGeneration; + const onError = (cause: unknown): void => { + if ( + generation !== this.#stdinGeneration || + this.#inputFailure !== undefined + ) + return; + this.#inputFailure = + cause instanceof Error + ? cause + : new Error("Owned worker request stream failed", { cause }); + }; + const onClose = (): void => { + stream.off("error", onError); + stream.off("close", onClose); + }; + stream.on("error", onError); + stream.once("close", onClose); + } + + #cleanupError( + reason: string, + cause?: unknown, + ): OwnedWorkerRuntimeCleanupError { + const resources = this.cleanupResources ?? []; + return new OwnedWorkerRuntimeCleanupError( + resources, + this.#supervisor?.launch.process.pid ?? null, + reason, + this.#cleanupEpoch, + cause === undefined ? undefined : { cause }, + ); + } +} + +const releasedCleanupOwner = { cleanupResources: [] } as const; + +const message = (cause: unknown): string => + cause instanceof Error ? cause.message : String(cause); diff --git a/tests/boundary/javascript/javascriptSourceAnalysisWorker.test.ts b/tests/boundary/javascript/javascriptSourceAnalysisWorker.test.ts index 0a6272bfc6..ff7f56aba6 100644 --- a/tests/boundary/javascript/javascriptSourceAnalysisWorker.test.ts +++ b/tests/boundary/javascript/javascriptSourceAnalysisWorker.test.ts @@ -1,6 +1,6 @@ import { createHash } from "node:crypto"; import { execFile } from "node:child_process"; -import { readFile } from "node:fs/promises"; +import { readFile, readdir } from "node:fs/promises"; import { join } from "node:path"; import { promisify } from "node:util"; import { expect, it } from "vitest"; @@ -30,9 +30,11 @@ const processLimit = javaScriptWorkerMemoryLimitBytes(128); it("reuses one real owned analyzer while retaining distinct source identities and exported returns", async () => { const launches: number[] = []; + let runtimeRoot: string | undefined; const worker = new JavaScriptAnalysisWorker( { heapMb: 128, timeoutMs: 30_000 }, async (request) => { + runtimeRoot = request.cwd; const spawned = await spawnOwnedProviderProcess(request); const pid = spawned.process.pid; if (pid === undefined) @@ -74,6 +76,15 @@ it("reuses one real owned analyzer while retaining distinct source identities an ).toBe(true); } expect(launches).toHaveLength(1); + if (runtimeRoot === undefined) + throw new Error( + "Owned analyzer did not expose its runtime root to launcher", + ); + expect( + (await readdir(runtimeRoot)).filter((name) => + /\.(?:source|static\.json|records)$/u.test(name), + ), + ).toEqual([]); } finally { await worker.close(); }