diff --git a/scripts/rea.mjs b/scripts/rea.mjs index ab16851541..321c801aad 100755 --- a/scripts/rea.mjs +++ b/scripts/rea.mjs @@ -80,7 +80,7 @@ if (!(await compiledRuntimeExists(runtimeFiles))) { writeStdout(result.output); process.exitCode = result.exitCode; } else { - const { cliConfigurationError, createCli } = + const { cliConfigurationError, closeCliJavaScriptResources, createCli } = await import("../dist/cli-runtime/cli.js"); const { createCliResultOutput } = await import("../dist/cli-runtime/cliResultOutput.js"); @@ -94,47 +94,97 @@ if (!(await compiledRuntimeExists(runtimeFiles))) { const cliArgs = normalizeCliArguments(args); const outputArguments = validateCliOutputArguments(cliArgs); const configurationError = cliConfigurationError(process.env); - // Doctor is the readiness report for this failure. Other commands still stop - // here so they do not run with an unusable log level. - if (configurationError !== undefined && !isDoctorCommand(cliArgs)) { - process.stderr.write(`${configurationError}\n`); - process.exitCode = 1; - } else if (!outputArguments.ok) { - writeStdout(renderCliOutputArgumentError(outputArguments)); - process.exitCode = 1; - } else { - let wroteOutput = false; - const resultOutput = createCliResultOutput(cliArgs, process.stdout); - try { - await createCli(process.env, resultOutput).serve(cliArgs, { - stdout: (output) => { - if (resultOutput?.handled) { - if (resultOutput.failed && !stdoutClosed) - process.stderr.write(output); - return; - } - if (output.length > 0) wroteOutput = true; - writeStdout(output); - }, - // A command failure caused by the closed pipe keeps the pipe's status. - exit: (code) => process.exit(stdoutClosed ? SIGPIPE_EXIT_STATUS : code), - }); - } catch (cause) { - if (wroteOutput) throw cause; - const recovered = await resultOutput - ?.recoverFormattingFailure(cause) - .catch((writeFailure) => { - // Recovery writes outside Incur's command boundary. A reader that - // declined this document keeps the same quiet SIGPIPE outcome. - if (!stdoutClosed) throw writeFailure; - return true; + let dispatchFailure; + // Incur can exit from the callback below before this function returns. + // That callback owns the status, including a closed-pipe exit of 141. + let exiting = false; + try { + // Doctor is the readiness report for this failure. Other commands still stop + // here so they do not run with an unusable log level. + if (configurationError !== undefined && !isDoctorCommand(cliArgs)) { + process.stderr.write(`${configurationError}\n`); + process.exitCode = 1; + } else if (!outputArguments.ok) { + writeStdout(renderCliOutputArgumentError(outputArguments)); + process.exitCode = 1; + } else { + let wroteOutput = false; + const resultOutput = createCliResultOutput(cliArgs, process.stdout); + try { + await createCli(process.env, resultOutput).serve(cliArgs, { + stdout: (output) => { + if (resultOutput?.handled) { + if (resultOutput.failed && !stdoutClosed) + process.stderr.write(output); + return; + } + if (output.length > 0) wroteOutput = true; + writeStdout(output); + }, + // A command failure caused by the closed pipe keeps the pipe's status. + // Worker shutdown is shared with the dispatch finally below. A failed + // shutdown is reported here because this callback is what exits. + exit: (code) => { + exiting = true; + const status = stdoutClosed ? SIGPIPE_EXIT_STATUS : code; + void Promise.resolve() + .then(() => closeCliJavaScriptResources()) + .then( + () => { + process.exit(status); + }, + (cleanup) => { + process.stderr.write( + `Failed to close the JavaScript analysis worker: ${cleanupFailureText(cleanup)}\n`, + ); + process.exit( + stdoutClosed + ? SIGPIPE_EXIT_STATUS + : status === 0 + ? 1 + : status, + ); + }, + ); + }, }); - if (!recovered) throw cause; - process.exitCode = stdoutClosed ? SIGPIPE_EXIT_STATUS : 1; + } catch (cause) { + if (wroteOutput) throw cause; + const recovered = await resultOutput + ?.recoverFormattingFailure(cause) + .catch((writeFailure) => { + // Recovery writes outside Incur's command boundary. A reader that + // declined this document keeps the same quiet SIGPIPE outcome. + if (!stdoutClosed) throw writeFailure; + return true; + }); + if (!recovered) throw cause; + process.exitCode = stdoutClosed ? SIGPIPE_EXIT_STATUS : 1; + } + if (!wroteOutput && !resultOutput?.handled) + writeStdout(renderEmptyFilteredCliOutput(cliArgs) ?? ""); + } + } catch (cause) { + dispatchFailure = cause; + } finally { + // Serve and output rendering can throw after a worker is parked. Stop it + // on that path too, and keep the dispatch failure when both fail. + try { + await closeCliJavaScriptResources(); + } catch (cleanup) { + // The exit callback reports this same shared failure and then exits. + if (!exiting) { + dispatchFailure = + dispatchFailure === undefined + ? cleanup + : new AggregateError( + [dispatchFailure, cleanup], + "CLI dispatch failed and the JavaScript analysis worker remained owned", + ); + } } - if (!wroteOutput && !resultOutput?.handled) - writeStdout(renderEmptyFilteredCliOutput(cliArgs) ?? ""); } + if (!exiting && dispatchFailure !== undefined) throw dispatchFailure; } /** True when the first positional command is `doctor`, after global output flags. */ @@ -164,6 +214,11 @@ function isDoctorCommand(arguments_) { * with the status a SIGPIPE-terminated writer reports. MCP stdio keeps its own * client lifecycle. */ +function cleanupFailureText(cleanup) { + if (cleanup instanceof Error) return cleanup.stack ?? cleanup.message; + return String(cleanup); +} + function guardClosedStdout() { process.stdout.on("error", (cause) => { if (cause?.code === "EPIPE") { diff --git a/src/application/javascript/JavaScriptArtifactReconstruction.ts b/src/application/javascript/JavaScriptArtifactReconstruction.ts index 1a2befbd9c..2252e5a6b1 100644 --- a/src/application/javascript/JavaScriptArtifactReconstruction.ts +++ b/src/application/javascript/JavaScriptArtifactReconstruction.ts @@ -158,7 +158,7 @@ export const reconstructJavaScriptArtifact = async ( ? {} : { fileNodes: input.max_semantic_nodes_per_file }), }); - const worker = createJavaScriptSourceAnalysis( + const worker = await createJavaScriptSourceAnalysis( { heapMb: input.max_heap_mb ?? JAVASCRIPT_ANALYSIS_RESOURCE_DEFAULTS.heapMb, @@ -169,6 +169,7 @@ export const reconstructJavaScriptArtifact = async ( files.files.filter( (file) => file.kind === "javascript" && file.text.included, ).length, + { scope }, ); let outcome; let cleanupFailure: AnalysisError | undefined; diff --git a/src/artifacts/ArtifactResourceScope.test.ts b/src/artifacts/ArtifactResourceScope.test.ts index f9bffdb185..417e8ee6e7 100644 --- a/src/artifacts/ArtifactResourceScope.test.ts +++ b/src/artifacts/ArtifactResourceScope.test.ts @@ -55,6 +55,89 @@ it("waits for admitted operations and shares concurrent close", async () => { expect(closed).toBe(true); }); +it("stops a close hook even when retained cleanup is still failing", async () => { + const scope = new ArtifactResourceScope(); + let releases = 0; + const owner = { + kind: "lease" as const, + resource: "stuck lease", + lease: { + async release() { + releases += 1; + if (releases < 3) throw new Error("still owned"); + }, + }, + }; + expect(await scope.release(owner)).toMatchObject({ kind: "failed" }); + let hookRuns = 0; + scope.addCloseHook(async () => { + hookRuns += 1; + if (hookRuns === 1) throw new Error("hook still stopping"); + }); + + const closeFailure = await scope.close().then( + () => { + throw new Error("close succeeded while cleanup and the hook failed"); + }, + (cause: unknown) => cause, + ); + expect(closeFailure).toMatchObject({ + message: "Artifact cleanup remains incomplete: stuck lease, close hook", + failedOwners: [owner], + cleanup: { + reason: "still owned; hook still stopping", + resources: ["stuck lease", "close hook"], + }, + }); + expect(closeFailure).toBeInstanceOf(ArtifactResourceRetryFailure); + const causes = (closeFailure as ArtifactResourceRetryFailure).cause; + expect(causes).toBeInstanceOf(AggregateError); + expect( + (causes as AggregateError).errors.map((error) => + error instanceof Error ? error.message : String(error), + ), + ).toEqual([ + "Artifact cleanup remains incomplete: stuck lease", + "hook still stopping", + ]); + expect(releases).toBe(2); + expect(hookRuns).toBe(1); + await scope.close(); + expect(releases).toBe(3); + expect(hookRuns).toBe(2); +}); + +it("runs later close hooks when an earlier hook fails", async () => { + const scope = new ArtifactResourceScope(); + const order: string[] = []; + scope.addCloseHook(async () => { + order.push("first"); + throw new Error("first hook failed"); + }); + scope.addCloseHook(async () => { + order.push("second"); + throw new Error("second hook failed"); + }); + + const failure = await scope.close().then( + () => { + throw new Error("close succeeded while hooks failed"); + }, + (cause: unknown) => cause, + ); + expect(order).toEqual(["first", "second"]); + expect(failure).toBeInstanceOf(AggregateError); + expect( + (failure as AggregateError).errors.map((error) => + error instanceof Error ? error.message : String(error), + ), + ).toEqual(["first hook failed", "second hook failed"]); + + order.length = 0; + await expect(scope.close()).rejects.toBeInstanceOf(AggregateError); + expect(order).toEqual(["first", "second"]); +}); + it("retains the same failed reader owner in a retry failure and retries before admitting work", async () => { const scope = new ArtifactResourceScope(); let closeAttempts = 0; diff --git a/src/artifacts/ArtifactResourceScope.ts b/src/artifacts/ArtifactResourceScope.ts index 9a7cf89e78..1b2848e76a 100644 --- a/src/artifacts/ArtifactResourceScope.ts +++ b/src/artifacts/ArtifactResourceScope.ts @@ -95,6 +95,8 @@ export class ArtifactResourceScope { >(); readonly #active = new Set>(); #draining: Promise | undefined; + #closeInFlight: Promise | undefined; + readonly #closeHooks: Array<() => Promise> = []; #closed = false; get hasPending(): boolean { @@ -203,10 +205,102 @@ export class ArtifactResourceScope { return this.#draining; } + /** + * Run after retained owners are released. Hooks stay registered so a later + * close retries them, and a failing hook does not skip the ones after it. + * Concurrent closes share one in-flight attempt. + */ + addCloseHook(hook: () => Promise): void { + this.#closeHooks.push(hook); + } + /** Stop admission, wait for active operations, and retry every retained owner. */ close(): Promise { this.#closed = true; - return this.retryCleanup(); + if (this.#closeInFlight !== undefined) return this.#closeInFlight; + const attempt = this.#finishClose(); + let published: Promise; + published = attempt.finally(() => { + if (this.#closeInFlight === published) this.#closeInFlight = undefined; + }); + this.#closeInFlight = published; + return published; + } + + /** Every registered hook runs. One failure is returned unchanged; several are aggregated. */ + async #runCloseHooks(): Promise { + const failures: unknown[] = []; + for (const hook of this.#closeHooks) { + try { + await hook(); + } catch (cause: unknown) { + failures.push(cause); + } + } + if (failures.length === 0) return undefined; + if (failures.length === 1) return failures[0]; + return new AggregateError( + failures, + failures + .map((cause) => + cause instanceof Error ? cause.message : String(cause), + ) + .join("; "), + ); + } + + async #finishClose(): Promise { + // A parked analyzer is not a pending owner. Stop it even when another + // retained cleanup is still failing. A later close retries both. + const cleanup = await this.retryCleanup().then( + () => undefined, + (cause: unknown) => cause, + ); + const hookFailure = await this.#runCloseHooks(); + if (cleanup !== undefined && hookFailure !== undefined) + throw this.#composeCloseFailures(cleanup, hookFailure); + if (hookFailure !== undefined) throw hookFailure; + if (cleanup !== undefined) throw cleanup; + } + + /** + * Report both outstanding owners. The pending set and the close hook stay + * registered, so the next close retries each failed path. + */ + #composeCloseFailures( + cleanup: unknown, + hookFailure: unknown, + ): ArtifactReaderFailure { + const retained = ArtifactReaderFailure.cleanupObservation( + cleanup, + "retained artifact cleanup", + ); + const hook = ArtifactReaderFailure.cleanupObservation( + hookFailure, + "close hook", + ); + const resources = [...new Set([...retained.resources, ...hook.resources])]; + const options = { + cause: new AggregateError( + [cleanup, hookFailure], + "Retained cleanup and a close hook both failed", + ), + cleanup: { + reason: `${retained.reason}; ${hook.reason}`, + resources, + }, + }; + const message = `Artifact cleanup remains incomplete: ${resources.join(", ")}`; + if (cleanup instanceof ArtifactResourceRetryFailure) + return new ArtifactResourceRetryFailure( + cleanup.failedOwners.map((owner) => ({ + owner, + lease: this.#lease(owner), + })), + message, + options, + ); + return new ArtifactReaderFailure("unavailable", message, options); } async #drainAndRetry(): Promise { diff --git a/src/artifacts/javascript/JavaScriptSourceAnalysis.ts b/src/artifacts/javascript/JavaScriptSourceAnalysis.ts index ac37fc75f2..e30049e412 100644 --- a/src/artifacts/javascript/JavaScriptSourceAnalysis.ts +++ b/src/artifacts/javascript/JavaScriptSourceAnalysis.ts @@ -3,10 +3,153 @@ import type { JavaScriptSourceAnalysisPort, } from "../../domain/javascript/javascriptSourceAnalysis.js"; import { JavaScriptAnalysisWorker } from "../../javascript/analysis/JavaScriptAnalysisWorker.js"; +import type { spawnOwnedProviderProcess } from "../../process/ProviderProcess.js"; +import { ArtifactResourceScope } from "../ArtifactResourceScope.js"; + +interface RetainedJavaScriptSourceAnalysis { + readonly worker: JavaScriptAnalysisWorker; + readonly limits: JavaScriptSourceAnalysisLimits; + busy: boolean; +} + +const retained = new WeakMap< + ArtifactResourceScope, + RetainedJavaScriptSourceAnalysis +>(); +const closeHookRegistered = new WeakSet(); + +export interface JavaScriptSourceAnalysisAcquisition { + /** Keep a healthy worker until this scope closes. */ + readonly scope?: ArtifactResourceScope; + /** Test seam. Reuse keeps the process already launched for the scope. */ + readonly launcher?: typeof spawnOwnedProviderProcess; +} /** Construct the inert artifact analyzer; resources are acquired on the first source. */ export const createJavaScriptSourceAnalysis = ( limits: JavaScriptSourceAnalysisLimits, sourceCount: number, -): JavaScriptSourceAnalysisPort => - new JavaScriptAnalysisWorker(limits, undefined, sourceCount); + acquisition: JavaScriptSourceAnalysisAcquisition = {}, +): Promise => { + if (acquisition.scope === undefined) + return Promise.resolve( + new JavaScriptAnalysisWorker(limits, acquisition.launcher, sourceCount), + ); + return checkoutJavaScriptSourceAnalysis( + acquisition.scope, + limits, + sourceCount, + acquisition.launcher, + ); +}; + +const checkoutJavaScriptSourceAnalysis = async ( + scope: ArtifactResourceScope, + limits: JavaScriptSourceAnalysisLimits, + sourceCount: number, + launcher: typeof spawnOwnedProviderProcess | undefined, +): Promise => { + registerJavaScriptSourceAnalysisClose(scope); + const current = retained.get(scope); + if (current !== undefined && !current.busy) { + if ( + sameLimits(current.limits, limits) && + current.worker.prepareForAnotherAnalysis(sourceCount) + ) { + current.busy = true; + return new ScopeJavaScriptSourceAnalysis(scope, current.worker); + } + retained.delete(scope); + try { + await current.worker.close(); + } catch (cause) { + // The old process is still alive. Keep it owned and fail this request + // before another worker is constructed, including on MCP where the + // analysis result is returned without a later cleanup retry. + if (retained.get(scope) === undefined) { + retained.set(scope, { + worker: current.worker, + limits: current.limits, + busy: false, + }); + } else { + scope.retain({ + kind: "lease", + resource: "replaced JavaScript source analyzer", + lease: { release: () => current.worker.close() }, + }); + } + throw cause; + } + } + const worker = new JavaScriptAnalysisWorker(limits, launcher, sourceCount); + // A concurrent analysis may have parked another healthy worker while the + // replaced process was closing. Claim the slot only when it is still empty. + // This extra process is closed through the port returned below. + if (retained.get(scope) === undefined) + retained.set(scope, { worker, limits, busy: true }); + return new ScopeJavaScriptSourceAnalysis(scope, worker); +}; + +const registerJavaScriptSourceAnalysisClose = ( + scope: ArtifactResourceScope, +): void => { + if (closeHookRegistered.has(scope)) return; + closeHookRegistered.add(scope); + scope.addCloseHook(async () => { + const current = retained.get(scope); + if (current === undefined) return; + await current.worker.close(); + if (retained.get(scope) === current) retained.delete(scope); + }); +}; + +const sameLimits = ( + left: JavaScriptSourceAnalysisLimits, + right: JavaScriptSourceAnalysisLimits, +): boolean => + left.heapMb === right.heapMb && left.timeoutMs === right.timeoutMs; + +/** + * Park a healthy worker on its scope. The next checkout with the same limits + * reuses that process; scope close is what actually stops it. + */ +class ScopeJavaScriptSourceAnalysis implements JavaScriptSourceAnalysisPort { + #failed = false; + #closed = false; + + constructor( + private readonly scope: ArtifactResourceScope, + private readonly worker: JavaScriptAnalysisWorker, + ) {} + + analyze( + ...args: Parameters + ): ReturnType { + return this.worker.analyze(...args).then((result) => { + if (!result.ok) this.#failed = true; + return result; + }); + } + + async close(): Promise { + if (this.#closed) return; + this.#closed = true; + try { + const current = retained.get(this.scope); + if ( + current?.worker === this.worker && + !this.#failed && + this.worker.canReuse() + ) { + current.busy = false; + return; + } + if (current?.worker === this.worker) retained.delete(this.scope); + await this.worker.close(); + } catch (cause) { + this.#closed = false; + throw cause; + } + } +} diff --git a/src/cli.ts b/src/cli.ts index d7e9e72ad9..586942bcd5 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -58,6 +58,8 @@ import { import { registerReverseEngineeringCommands } from "./cli/reverseEngineeringCommands.js"; import { registerCutterCommands } from "./cli/cutterCommands.js"; +export { closeCliJavaScriptResources } from "./cli/javascriptApplicationAnalysis.js"; + /** * Describe CLI settings that are invalid before any command runs, as MCP does. * Keep this aligned with PROCESS_START_CONFIGURATION_SETTINGS. diff --git a/src/cli/javascriptApplicationAnalysis.ts b/src/cli/javascriptApplicationAnalysis.ts index e2f171bec4..f4fcc95575 100644 --- a/src/cli/javascriptApplicationAnalysis.ts +++ b/src/cli/javascriptApplicationAnalysis.ts @@ -7,7 +7,43 @@ import { projectAnalysisError } from "../domain/analysisErrorProjection.js"; import type { JsonValue } from "../domain/jsonValue.js"; import { ProviderCleanupError } from "../domain/providerCleanupError.js"; -/** Execute the shared one-shot CLI boundary for static JavaScript analysis. */ +/** + * One scope for every JavaScript analysis in this process. A healthy worker + * stays parked between commands; {@link closeCliJavaScriptResources} stops it. + * Pending cleanup is still retried before each command returns. + */ +let retainedScope: ArtifactResourceScope | undefined; +/** One shutdown shared by the exit callback and the dispatch `finally`. */ +let closing: Promise | undefined; + +const cliAnalysisScope = (): ArtifactResourceScope => { + retainedScope ??= new ArtifactResourceScope(); + return retainedScope; +}; + +/** + * Stop the process-owned JavaScript worker. Callers that overlap share the + * attempt in flight. The scope stays owned when shutdown fails, so the next + * call retries instead of treating the worker as already released. + */ +export const closeCliJavaScriptResources = (): Promise => { + if (closing !== undefined) return closing; + const scope = retainedScope; + if (scope === undefined) return Promise.resolve(); + let published: Promise; + published = scope + .close() + .then(() => { + if (retainedScope === scope) retainedScope = undefined; + }) + .finally(() => { + if (closing === published) closing = undefined; + }); + closing = published; + return published; +}; + +/** Execute the shared CLI boundary for static JavaScript analysis. */ export const runCliJavaScriptApplicationAnalysis = async ( input: unknown, signal?: AbortSignal, @@ -20,13 +56,14 @@ export const runCliJavaScriptApplicationAnalysis = async ( }, { minimumIntervalMs: 0 }, ); - const resources = new ArtifactResourceScope(); + const resources = cliAnalysisScope(); const result = await analyzeJavaScriptApplication(input, resources, { progress, ...(signal === undefined ? {} : { signal }), }); try { - await resources.close(); + // Retry failed owners without stopping a healthy parked worker. + await resources.retryCleanup(); } catch (cause: unknown) { const cleanup = ArtifactReaderFailure.cleanupObservation( cause, diff --git a/src/javascript/analysis/JavaScriptAnalysisWorker.ts b/src/javascript/analysis/JavaScriptAnalysisWorker.ts index dea646d263..7f978e2352 100644 --- a/src/javascript/analysis/JavaScriptAnalysisWorker.ts +++ b/src/javascript/analysis/JavaScriptAnalysisWorker.ts @@ -105,7 +105,7 @@ export class JavaScriptAnalysisWorker { #parentCapacityExhausted = false; #sourcesRemaining: number | undefined; #deadline: ProviderStartupDeadline | undefined; - readonly #resultBudgetBytes: number; + #resultBudgetBytes = 0; /** Null when the runtime enforces the V8 heap flags instead. */ readonly #processMemoryLimitBytes: number | null; readonly #onStdout = (chunk: Buffer | string): void => { @@ -179,8 +179,50 @@ export class JavaScriptAnalysisWorker { this.#processMemoryLimitBytes = javaScriptWorkerMemoryLimitBytes( limits.heapMb, ); - // Reserve host-side control, schema/Evidence work and a maximum record's - // transient copies. Admission also accounts for retained earlier files. + this.#refreshResultBudget(); + } + + /** + * Whether this idle process can analyze another application. A failed, + * cancelled, or still-open job must not be reused. + */ + canReuse(): boolean { + return ( + this.#ready && + this.#activeJobId === undefined && + this.#primaryFailure === undefined && + this.#lastFailure === undefined && + this.#frames.length === 0 && + this.#pending === "" && + this.#runtime.inputFailure === undefined && + !this.#runtime.hasPendingCleanup && + this.#runtime.supervisor !== undefined + ); + } + + /** + * Reset per-analysis admission for one idle process. The worker heap limit + * stays the one selected at launch; callers must not reuse across limits. + */ + prepareForAnotherAnalysis(sourceCount: number): boolean { + if (!this.canReuse()) return false; + if (!Number.isSafeInteger(sourceCount) || sourceCount < 0) + throw new RangeError( + "JavaScript worker source count must be a nonnegative safe integer", + ); + this.#sourcesRemaining = sourceCount; + this.#estimatedRetainedBytes = 0; + this.#parentCapacityExhausted = false; + this.#applicationProjectionFailure = undefined; + this.#refreshResultBudget(); + return true; + } + + /** + * Reserve host-side control, schema/Evidence work, and a maximum record's + * transient copies. Admission also accounts for retained earlier files. + */ + #refreshResultBudget(): void { this.#resultBudgetBytes = Math.max( 0, Math.floor( diff --git a/tests/boundary/cli/cliJavaScriptResourceClose.test.ts b/tests/boundary/cli/cliJavaScriptResourceClose.test.ts new file mode 100644 index 0000000000..2ba0873d39 --- /dev/null +++ b/tests/boundary/cli/cliJavaScriptResourceClose.test.ts @@ -0,0 +1,66 @@ +import { afterEach, expect, it } from "vitest"; + +import { ArtifactResourceScope } from "../../../src/artifacts/ArtifactResourceScope.js"; +import { + closeCliJavaScriptResources, + runCliJavaScriptApplicationAnalysis, +} from "../../../src/cli/javascriptApplicationAnalysis.js"; + +const originalClose = ArtifactResourceScope.prototype.close; + +afterEach(async () => { + ArtifactResourceScope.prototype.close = originalClose; + await closeCliJavaScriptResources(); +}); + +/** Create the process-owned scope without starting a worker. */ +const openCliScope = (): Promise => + runCliJavaScriptApplicationAnalysis({}); + +it("retries JavaScript worker shutdown when the first close fails", async () => { + let calls = 0; + ArtifactResourceScope.prototype.close = function ( + this: ArtifactResourceScope, + ) { + calls += 1; + if (calls === 1) return Promise.reject(new Error("worker still owned")); + return originalClose.call(this); + }; + + await openCliScope(); + await expect(closeCliJavaScriptResources()).rejects.toThrow( + "worker still owned", + ); + expect(calls).toBe(1); + await closeCliJavaScriptResources(); + expect(calls).toBe(2); + await closeCliJavaScriptResources(); + expect(calls).toBe(2); +}); + +it("shares one in-flight JavaScript worker shutdown between callers", async () => { + let calls = 0; + let release = (): void => {}; + const gate = new Promise((resolve) => { + release = resolve; + }); + ArtifactResourceScope.prototype.close = function ( + this: ArtifactResourceScope, + ) { + calls += 1; + return gate.then(() => originalClose.call(this)); + }; + + try { + await openCliScope(); + const first = closeCliJavaScriptResources(); + const second = closeCliJavaScriptResources(); + expect(second).toBe(first); + expect(calls).toBe(1); + release(); + await first; + expect(calls).toBe(1); + } finally { + release(); + } +}); diff --git a/tests/boundary/cli/dispatcher.test.ts b/tests/boundary/cli/dispatcher.test.ts index 316a2cff21..40b46736de 100644 --- a/tests/boundary/cli/dispatcher.test.ts +++ b/tests/boundary/cli/dispatcher.test.ts @@ -4,32 +4,39 @@ import { join } from "node:path"; import { describe, expect } from "vitest"; import { workspaceCliTest } from "../../support/cli/workspaceCliFixture.js"; +import type { TestWorkspace } from "../../support/workspace/workspaceFixture.js"; + +const CLI_OUTPUT_RUNTIME = + "export const normalizeCliArguments = (args) => args; export const validateCliOutputArguments = () => ({ ok: true }); export const renderCliOutputArgumentError = () => ''; export const renderEmptyFilteredCliOutput = () => undefined;\n"; + +const installCliRuntime = async ( + workspace: TestWorkspace, + cliSource: string, + resultOutput = "export const createCliResultOutput = () => undefined;\n", +): Promise => { + await workspace.mkdir("scripts"); + await workspace.mkdir("dist/cli-runtime"); + await Promise.all([ + copyFile("scripts/rea.mjs", workspace.path("scripts/rea.mjs")), + workspace.write( + "package.json", + JSON.stringify({ type: "module", version: "1.0.0" }), + ), + workspace.write("dist/cli-runtime/cli.js", cliSource), + workspace.write("dist/cli-runtime/cliOutput.js", CLI_OUTPUT_RUNTIME), + workspace.write("dist/cli-runtime/cliResultOutput.js", resultOutput), + ]); +}; describe("executable dispatcher", () => { workspaceCliTest( "does not append recovery output after delegated stdout has begun", async ({ processes, workspace }) => { - await workspace.mkdir("scripts"); - await workspace.mkdir("dist/cli-runtime"); - await Promise.all([ - copyFile("scripts/rea.mjs", workspace.path("scripts/rea.mjs")), - workspace.write( - "package.json", - JSON.stringify({ type: "module", version: "1.0.0" }), - ), - workspace.write( - "dist/cli-runtime/cli.js", - 'export const cliConfigurationError = () => undefined; export const createCli = () => ({ serve: (_args, options) => { options.stdout("first document\\n"); throw new RangeError("Invalid string length"); } });\n', - ), - workspace.write( - "dist/cli-runtime/cliOutput.js", - "export const normalizeCliArguments = (args) => args; export const validateCliOutputArguments = () => ({ ok: true }); export const renderCliOutputArgumentError = () => ''; export const renderEmptyFilteredCliOutput = () => undefined;\n", - ), - workspace.write( - "dist/cli-runtime/cliResultOutput.js", - 'export const createCliResultOutput = () => ({ handled: false, failed: false, recoverFormattingFailure: () => { process.stdout.write("unexpected second document"); return true; } });\n', - ), - ]); + await installCliRuntime( + workspace, + 'export const cliConfigurationError = () => undefined; export const closeCliJavaScriptResources = () => {}; export const createCli = () => ({ serve: (_args, options) => { options.stdout("first document\\n"); throw new RangeError("Invalid string length"); } });\n', + 'export const createCliResultOutput = () => ({ handled: false, failed: false, recoverFormattingFailure: () => { process.stdout.write("unexpected second document"); return true; } });\n', + ); const result = await processes.run( process.execPath, [workspace.path("scripts/rea.mjs"), "fixture"], @@ -60,7 +67,7 @@ describe("executable dispatcher", () => { ), workspace.write( "dist/cli-runtime/cli.js", - 'export const cliConfigurationError = () => undefined; export const createCli = () => ({ serve: (args) => process.stdout.write(JSON.stringify({ adapter: "cli", args })) });\n', + 'export const cliConfigurationError = () => undefined; export const closeCliJavaScriptResources = () => {}; export const createCli = () => ({ serve: (args) => process.stdout.write(JSON.stringify({ adapter: "cli", args })) });\n', ), workspace.write( "dist/cli-runtime/cliOutput.js", @@ -155,3 +162,39 @@ describe("executable dispatcher", () => { }, ); }); + +describe("executable dispatcher JavaScript shutdown", () => { + workspaceCliTest( + "reports a failed JavaScript worker shutdown instead of exiting successfully", + async ({ processes, workspace }) => { + await installCliRuntime( + workspace, + 'export const cliConfigurationError = () => undefined; export const closeCliJavaScriptResources = () => Promise.reject(new Error("worker still owned")); export const createCli = () => ({ serve: (_args, options) => { options.exit(0); } });\n', + ); + const result = await processes.run( + process.execPath, + [workspace.path("scripts/rea.mjs"), "fixture"], + { cwd: workspace.root }, + ); + expect(result.exitCode).toBe(1); + expect(result.stderr).toContain("worker still owned"); + }, + ); + + workspaceCliTest( + "keeps the closed-pipe status when JavaScript worker shutdown fails", + async ({ processes, workspace }) => { + await installCliRuntime( + workspace, + 'export const cliConfigurationError = () => undefined; export const closeCliJavaScriptResources = () => Promise.reject(new Error("worker still owned")); export const createCli = () => ({ serve: (_args, options) => { process.stdout.emit("error", Object.assign(new Error("broken pipe"), { code: "EPIPE" })); options.exit(0); } });\n', + ); + const result = await processes.run( + process.execPath, + [workspace.path("scripts/rea.mjs"), "fixture"], + { cwd: workspace.root }, + ); + expect(result.exitCode).toBe(141); + expect(result.stderr).toContain("worker still owned"); + }, + ); +}); diff --git a/tests/boundary/filesystem/javascript-applications/javascriptApplicationService.test.ts b/tests/boundary/filesystem/javascript-applications/javascriptApplicationService.test.ts index 4dbf49ae0e..af7b0777ca 100644 --- a/tests/boundary/filesystem/javascript-applications/javascriptApplicationService.test.ts +++ b/tests/boundary/filesystem/javascript-applications/javascriptApplicationService.test.ts @@ -38,7 +38,7 @@ describe("JavaScript source selection cancellation boundary", () => { controller.signal, sourcePaths, ); - const worker = createJavaScriptSourceAnalysis( + const worker = await createJavaScriptSourceAnalysis( { heapMb: 512, timeoutMs: 30_000 }, 1, ); diff --git a/tests/boundary/javascript/javascriptSourceAnalysisWorker.test.ts b/tests/boundary/javascript/javascriptSourceAnalysisWorker.test.ts index ff7f56aba6..117230e2e1 100644 --- a/tests/boundary/javascript/javascriptSourceAnalysisWorker.test.ts +++ b/tests/boundary/javascript/javascriptSourceAnalysisWorker.test.ts @@ -5,6 +5,8 @@ import { join } from "node:path"; import { promisify } from "node:util"; import { expect, it } from "vitest"; import { JavaScriptAnalysisWorker } from "../../../src/javascript/analysis/JavaScriptAnalysisWorker.js"; +import { createJavaScriptSourceAnalysis } from "../../../src/artifacts/javascript/JavaScriptSourceAnalysis.js"; +import { ArtifactResourceScope } from "../../../src/artifacts/ArtifactResourceScope.js"; import { cleanupOwnedProcessGroup } from "../../../src/process/ProcessOwnership.js"; import type { OwnedProcessGroup } from "../../../src/process/ProcessOwnership.js"; import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js"; @@ -90,6 +92,278 @@ it("reuses one real owned analyzer while retaining distinct source identities an } }); +it("reuses one owned worker across analyses on the same resource scope", async () => { + const scope = new ArtifactResourceScope(); + const launches: number[] = []; + const exits = new Map>(); + const launcher: typeof spawnOwnedProviderProcess = async (request) => { + const previous = launches.at(-1); + if (previous !== undefined) await exits.get(previous); + const spawned = await spawnOwnedProviderProcess(request); + const pid = spawned.process.pid; + if (pid === undefined) + throw new Error("Owned analyzer has no observed PID"); + exits.set( + pid, + new Promise((resolve) => { + spawned.process.once("exit", () => { + resolve(); + }); + }), + ); + launches.push(pid); + return spawned; + }; + const limits = { heapMb: 128, timeoutMs: 30_000 }; + const first = source("first.mjs", "export const observed = 1;"); + const second = source("second.mjs", "export const observed = 2;"); + const files = [first, second]; + try { + for (const file of files) { + const worker = await createJavaScriptSourceAnalysis(limits, 1, { + scope, + launcher, + }); + try { + const result = await worker.analyze(file, { + nodes: 100, + unknowns: 100, + }); + if (!result.ok) throw result.error.error; + expect(result.value.javascript.parse_status).toBe("complete"); + } finally { + await worker.close(); + } + } + expect(launches).toHaveLength(1); + const replaced = await createJavaScriptSourceAnalysis( + { heapMb: 256, timeoutMs: 30_000 }, + 1, + { scope, launcher }, + ); + try { + const result = await replaced.analyze(first, { + nodes: 100, + unknowns: 100, + }); + if (!result.ok) throw result.error.error; + expect(result.value.javascript.parse_status).toBe("complete"); + } finally { + await replaced.close(); + } + expect(launches).toHaveLength(2); + await scope.close(); + const reopened = await createJavaScriptSourceAnalysis(limits, 1, { + scope, + launcher, + }); + try { + const result = await reopened.analyze(first, { + nodes: 100, + unknowns: 100, + }); + if (!result.ok) throw result.error.error; + } finally { + await reopened.close(); + } + expect(launches).toHaveLength(3); + } finally { + await scope.close(); + } +}); + +it("keeps a worker parked while a limit change closes the previous process", async () => { + const scope = new ArtifactResourceScope(); + const launches: number[] = []; + const exited = new Set(); + const exits = new Map>(); + const launcher: typeof spawnOwnedProviderProcess = async (request) => { + const spawned = await spawnOwnedProviderProcess(request); + const pid = spawned.process.pid; + if (pid === undefined) + throw new Error("Owned analyzer has no observed PID"); + exits.set( + pid, + new Promise((resolve) => { + spawned.process.once("exit", () => { + exited.add(pid); + resolve(); + }); + }), + ); + launches.push(pid); + return spawned; + }; + const limits = { heapMb: 128, timeoutMs: 30_000 }; + const file = source("parked.mjs", "export const observed = 1;"); + const originalClose = JavaScriptAnalysisWorker.prototype.close; + let releaseClose = (): void => {}; + const gate = new Promise((resolve) => { + releaseClose = resolve; + }); + let closeCalls = 0; + let markCloseStarted = (): void => {}; + const closeStarted = new Promise((resolve) => { + markCloseStarted = resolve; + }); + let replacement: + | ReturnType + | undefined; + try { + const initial = await createJavaScriptSourceAnalysis(limits, 1, { + scope, + launcher, + }); + const first = await initial.analyze(file, { nodes: 100, unknowns: 100 }); + if (!first.ok) throw first.error.error; + await initial.close(); + const replacedPid = launches[0]; + if (replacedPid === undefined) + throw new Error("Replaced analyzer has no observed PID"); + + JavaScriptAnalysisWorker.prototype.close = async function ( + this: JavaScriptAnalysisWorker, + ) { + closeCalls += 1; + if (closeCalls === 1) { + markCloseStarted(); + await gate; + } + return originalClose.call(this); + }; + replacement = createJavaScriptSourceAnalysis( + { heapMb: 256, timeoutMs: 30_000 }, + 1, + { scope, launcher }, + ); + await closeStarted; + + const concurrent = await createJavaScriptSourceAnalysis(limits, 1, { + scope, + launcher, + }); + const concurrentResult = await concurrent.analyze(file, { + nodes: 100, + unknowns: 100, + }); + if (!concurrentResult.ok) throw concurrentResult.error.error; + await concurrent.close(); + const parkedPid = launches[1]; + if (parkedPid === undefined) + throw new Error("Parked analyzer has no observed PID"); + + releaseClose(); + const extra = await replacement; + const extraResult = await extra.analyze(file, { + nodes: 100, + unknowns: 100, + }); + if (!extraResult.ok) throw extraResult.error.error; + const extraPid = launches[2]; + if (extraPid === undefined) + throw new Error("Replacement analyzer has no observed PID"); + await extra.close(); + await exits.get(replacedPid); + await exits.get(extraPid); + expect(exited.has(replacedPid)).toBe(true); + expect(exited.has(extraPid)).toBe(true); + expect(exited.has(parkedPid)).toBe(false); + + await scope.close(); + await exits.get(parkedPid); + expect(exited.has(parkedPid)).toBe(true); + } finally { + JavaScriptAnalysisWorker.prototype.close = originalClose; + releaseClose(); + const port = await replacement?.catch(() => undefined); + await port?.close().catch(() => undefined); + await scope.close(); + } +}); + +it("does not start a replacement when closing the previous worker fails", async () => { + const scope = new ArtifactResourceScope(); + const launches: number[] = []; + const exited = new Set(); + const exits = new Map>(); + const launcher: typeof spawnOwnedProviderProcess = async (request) => { + const previous = launches.at(-1); + if (previous !== undefined) await exits.get(previous); + const spawned = await spawnOwnedProviderProcess(request); + const pid = spawned.process.pid; + if (pid === undefined) + throw new Error("Owned analyzer has no observed PID"); + exits.set( + pid, + new Promise((resolve) => { + spawned.process.once("exit", () => { + exited.add(pid); + resolve(); + }); + }), + ); + launches.push(pid); + return spawned; + }; + const limits = { heapMb: 128, timeoutMs: 30_000 }; + const file = source("replaced.mjs", "export const observed = 1;"); + const originalClose = JavaScriptAnalysisWorker.prototype.close; + let closeCalls = 0; + try { + const initial = await createJavaScriptSourceAnalysis(limits, 1, { + scope, + launcher, + }); + const first = await initial.analyze(file, { nodes: 100, unknowns: 100 }); + if (!first.ok) throw first.error.error; + await initial.close(); + const replacedPid = launches[0]; + if (replacedPid === undefined) + throw new Error("Replaced analyzer has no observed PID"); + + JavaScriptAnalysisWorker.prototype.close = async function ( + this: JavaScriptAnalysisWorker, + ) { + closeCalls += 1; + if (closeCalls === 1) throw new Error("worker still owned"); + return originalClose.call(this); + }; + await expect( + createJavaScriptSourceAnalysis({ heapMb: 256, timeoutMs: 30_000 }, 1, { + scope, + launcher, + }), + ).rejects.toThrow("worker still owned"); + expect(launches).toEqual([replacedPid]); + expect(exited.has(replacedPid)).toBe(false); + + JavaScriptAnalysisWorker.prototype.close = originalClose; + const replacement = await createJavaScriptSourceAnalysis( + { heapMb: 256, timeoutMs: 30_000 }, + 1, + { scope, launcher }, + ); + try { + const result = await replacement.analyze(file, { + nodes: 100, + unknowns: 100, + }); + if (!result.ok) throw result.error.error; + } finally { + await replacement.close(); + } + expect(launches).toHaveLength(2); + expect(exited.has(replacedPid)).toBe(true); + const replacementPid = launches[1]; + if (replacementPid === undefined) + throw new Error("Replacement analyzer has no observed PID"); + expect(exited.has(replacementPid)).toBe(false); + } finally { + JavaScriptAnalysisWorker.prototype.close = originalClose; + await scope.close(); + } +}); + it("reports an actual startup deadline and closes its owned resources before subsequent analysis", async () => { const file = source("app.mjs", "export const observed = 7;"); const expired = new JavaScriptAnalysisWorker({ heapMb: 128, timeoutMs: 1 }); diff --git a/tests/boundary/javascriptSemanticTraceEvidence.test.ts b/tests/boundary/javascriptSemanticTraceEvidence.test.ts index 25316c648c..94157c1307 100644 --- a/tests/boundary/javascriptSemanticTraceEvidence.test.ts +++ b/tests/boundary/javascriptSemanticTraceEvidence.test.ts @@ -1,13 +1,13 @@ import { mkdir, writeFile } from "node:fs/promises"; import { join } from "node:path"; -import { expect, it } from "vitest"; +import { afterAll, expect, it } from "vitest"; import { Ajv2020 } from "ajv/dist/2020.js"; import { createApplicationMcpHarness } from "../fixtures/applicationMcpHarness.js"; import { parseMcpToolError } from "../fixtures/mcpToolError.js"; import { createTestTempDirectory } from "../fixtures/temporaryDirectory.js"; -import { createCli } from "../../src/cli.js"; +import { closeCliJavaScriptResources, createCli } from "../../src/cli.js"; import { createEvidence, parseEvidence } from "../../src/domain/evidence.js"; import type { JsonValue } from "../../src/domain/jsonValue.js"; import { analysisCliErrorEnvelopeSchema } from "../../src/contracts/errorSchemas.js"; @@ -18,6 +18,10 @@ import { resolveJavaScriptSemanticEvidence } from "../../src/domain/javascript/j const SEMANTIC_GRAPH_ID = `jsrg_${"0".repeat(64)}`; +// In-process serve keeps the CLI worker for this file. Stop it before the +// worker process exits so the private runtime root is not left behind. +afterAll(closeCliJavaScriptResources); + it("reports invalid nested semantic graph Evidence as caller input through CLI and SDK MCP", async () => { const root = await createTestTempDirectory("rea-semantic-trace-input-"); await mkdir(join(root, "src")); diff --git a/tests/support/cli/cliCommandSessionWorker.mjs b/tests/support/cli/cliCommandSessionWorker.mjs index 63ce69542c..683682542a 100644 --- a/tests/support/cli/cliCommandSessionWorker.mjs +++ b/tests/support/cli/cliCommandSessionWorker.mjs @@ -50,7 +50,7 @@ if (!(await compiledRuntimeExists(runtimeFiles))) { process.exit(1); } -const { cliConfigurationError, createCli } = +const { cliConfigurationError, closeCliJavaScriptResources, createCli } = await import("../../../dist/cli-runtime/cli.js"); const { createCliResultOutput } = await import("../../../dist/cli-runtime/cliResultOutput.js"); @@ -89,7 +89,10 @@ process.stdin.on("data", (chunk) => { } }); process.stdin.on("end", () => { - chain.finally(() => process.exit(0)); + chain.finally(async () => { + await closeCliJavaScriptResources(); + process.exit(0); + }); }); async function handle(value) { diff --git a/tests/support/cli/compiledCliWorker.mjs b/tests/support/cli/compiledCliWorker.mjs index e8012c7510..50b08093d9 100644 --- a/tests/support/cli/compiledCliWorker.mjs +++ b/tests/support/cli/compiledCliWorker.mjs @@ -1,7 +1,7 @@ import { Writable } from "node:stream"; import { createInterface } from "node:readline"; -const { cliConfigurationError, createCli } = +const { cliConfigurationError, closeCliJavaScriptResources, createCli } = await import("../../../dist/cli-runtime/cli.js"); const { createCliResultOutput } = await import("../../../dist/cli-runtime/cliResultOutput.js"); @@ -137,3 +137,4 @@ for await (const line of lines) { if (line.trim().length === 0) continue; await acceptRequest(line); } +await closeCliJavaScriptResources();