diff --git a/src/browser/WebBundleAnalysisWorker.test.ts b/src/browser/WebBundleAnalysisWorker.test.ts index a75be3a8c5..643d916ed9 100644 --- a/src/browser/WebBundleAnalysisWorker.test.ts +++ b/src/browser/WebBundleAnalysisWorker.test.ts @@ -45,6 +45,7 @@ const makeScript = (text: string): IncludedScript => ({ const sourceWorkerProgram = ( hangMarker?: string, oomMarker?: string, + stderrMarker?: string, ): string => ` const { serveWebBundleAnalysisWorker } = await import(${JSON.stringify(new URL("./WebBundleAnalysisWorkerServer.ts", import.meta.url).href)}); const { analyzeCapturedWebScript } = await import(${JSON.stringify(new URL("../domain/webBundleAnalyzer.ts", import.meta.url).href)}); @@ -56,6 +57,8 @@ const sourceWorkerProgram = ( const retained = []; for (;;) retained.push(new Array(50_000).fill({ marker: "bounded-child-oom" })); } + if (${JSON.stringify(stderrMarker)} !== undefined) + process.stderr.write(${JSON.stringify(stderrMarker)} + "\\n"); return analyzeCapturedWebScript(script); }; await serveWebBundleAnalysisWorker(analyze); @@ -66,29 +69,89 @@ type TestMode = readonly kind: "source"; readonly hangMarker?: string; readonly oomMarker?: string; + readonly stderrMarker?: string; } | { readonly kind: "delayed-ready" } + | { readonly kind: "exit-before-ready" } | { readonly kind: "protocol"; - readonly failure: "wrong-id" | "oversized-control" | "oversized-result"; + readonly failure: + | "wrong-id" + | "oversized-control" + | "oversized-result" + | "late-failure" + | "exit-7" + | "truncated-exit" + | "abort-first-exit" + | "tail-exit" + | "malformed" + | "wrong-job-failure" + | "resource-failure" + | "final-frame-held-pipe" + | "zero-exit-held-pipe"; }; const protocolWorkerProgram = ( - failure: "wrong-id" | "oversized-control" | "oversized-result", + failure: + | "wrong-id" + | "oversized-control" + | "oversized-result" + | "late-failure" + | "exit-7" + | "truncated-exit" + | "abort-first-exit" + | "tail-exit" + | "malformed" + | "wrong-job-failure" + | "resource-failure" + | "final-frame-held-pipe" + | "zero-exit-held-pipe", ): string => ` import { createInterface } from "node:readline"; + import { spawn } from "node:child_process"; import { getHeapStatistics } from "node:v8"; const send = (value) => process.stdout.write(JSON.stringify(value) + "\\n"); send({ kind: "ready", heap_limit_bytes: getHeapStatistics().heap_size_limit }); + ${failure === "abort-first-exit" ? `process.on("SIGTERM", () => {}); setTimeout(() => process.stderr.write("late natural exit 7\\n", () => process.exit(7)), 50);` : ""} const input = createInterface({ input: process.stdin, crlfDelay: Infinity }); input.once("line", (line) => { const request = JSON.parse(line); ${ - failure === "wrong-id" - ? `send({ kind: "phase", id: ${JSON.stringify(randomUUID())}, phase: "analyze" });` - : failure === "oversized-control" - ? `process.stdout.write("x".repeat(70 * 1024) + "\\n");` - : `send({ kind: "complete", id: request.id, result: { bytes: request.result_budget_bytes + 1, sha256: "${"0".repeat(64)}" } });` + failure === "zero-exit-held-pipe" + ? `const program = "setTimeout(() => process.exit(0), 300);"; + spawn(process.execPath, ["-e", program], { stdio: ["ignore", "inherit", "inherit"] }); + process.exit(0);` + : failure === "exit-7" + ? `process.stderr.write("controlled exit 7\\n", () => process.exit(7));` + : failure === "tail-exit" + ? `const lateProgram = "setTimeout(() => { process.stderr.write('late tail marker'); setTimeout(() => process.exit(0), 500); }, 50);"; + spawn(process.execPath, ["-e", lateProgram], { stdio: ["ignore", "ignore", "inherit"] }); + process.stderr.write("early exit marker\\n", () => process.exit(7));` + : failure === "truncated-exit" + ? `process.stderr.write("x".repeat(300 * 1024), () => process.exit(7));` + : failure === "abort-first-exit" + ? `setInterval(() => {}, 1_000);` + : failure === "malformed" + ? `process.stdout.write("{not-json}\\n"); setInterval(() => {}, 1_000);` + : failure === "wrong-job-failure" + ? `send({ kind: "failure", id: ${JSON.stringify(randomUUID())}, reason: "output", message: "failure for another request" }); + setInterval(() => {}, 1_000);` + : failure === "resource-failure" + ? `send({ kind: "failure", id: request.id, reason: "resource", message: "bounded result capacity" }); + setInterval(() => {}, 1_000);` + : failure === "final-frame-held-pipe" + ? `const frame = JSON.stringify({ kind: "failure", id: request.id, reason: "output", message: "final frame from descendant" }); + const program = "setTimeout(() => { process.stdout.write(" + JSON.stringify(frame + "\\n") + "); setInterval(() => {}, 1000); }, 100);"; + spawn(process.execPath, ["-e", program], { stdio: ["ignore", "inherit", "inherit"] }); + process.exit(7);` + : failure === "late-failure" + ? `process.stdout.write(JSON.stringify({ kind: "failure", id: request.id, reason: "output", message: "failure frame before abort" }) + "\\n"); + setInterval(() => {}, 1_000);` + : failure === "wrong-id" + ? `send({ kind: "phase", id: ${JSON.stringify(randomUUID())}, phase: "analyze" });` + : failure === "oversized-control" + ? `process.stdout.write("x".repeat(70 * 1024) + "\\n");` + : `send({ kind: "complete", id: request.id, result: { bytes: request.result_budget_bytes + 1, sha256: "${"0".repeat(64)}" } });` } }); `; @@ -102,6 +165,8 @@ const delayedReadyProgram = ` createInterface({ input: process.stdin, crlfDelay: Infinity }); `; +const exitBeforeReadyProgram = `process.exit(0);`; + interface SpawnedTestProcess { readonly process: ProviderProcessHandle; readonly root: string | undefined; @@ -119,10 +184,16 @@ const launcherFor = ( const loader = import.meta.resolve("tsx"); const program = mode.kind === "source" - ? sourceWorkerProgram(mode.hangMarker, mode.oomMarker) + ? sourceWorkerProgram( + mode.hangMarker, + mode.oomMarker, + mode.stderrMarker, + ) : mode.kind === "delayed-ready" ? delayedReadyProgram - : protocolWorkerProgram(mode.failure); + : mode.kind === "exit-before-ready" + ? exitBeforeReadyProgram + : protocolWorkerProgram(mode.failure); const spawned = await spawnOwnedProviderProcess({ ...request, arguments: [ @@ -209,6 +280,450 @@ it("reuses one real owned worker for completed facts from consecutive scripts", } }); +it("preserves an observed failure frame and the opaque later abort reason", async () => { + const reason = { opaque: "caller cancellation reason" }; + const controller = new AbortController(); + const worker = new WebBundleAnalysisWorker( + WEB_BUNDLE_WORKER_LIMITS.heapMb, + launcherFor( + { kind: "protocol", failure: "late-failure" }, + { + onSpawn: ({ process: child }) => { + child.stdout?.on("data", (chunk) => { + if (String(chunk).includes("failure frame before abort")) + queueMicrotask(() => controller.abort(reason)); + }); + }, + }, + ), + ); + try { + const result = await worker.analyze( + makeScript("fetch('/late');"), + 5_000, + controller.signal, + ); + const error = failureOf(result); + expect(error).toBeInstanceOf(AnalysisOutputError); + expect(error.message).toContain("failure frame before abort"); + expect(error.secondaryFailures).toHaveLength(1); + expect(error.secondaryFailures[0]).toBeInstanceOf(AnalysisCancelledError); + expect(error.secondaryFailures[0]?.cause).toBe(reason); + } finally { + await worker.close(); + } +}); + +it("retains exit-7 output when abort follows the observed exit", async () => { + for (const abortAfterExit of [false, true]) { + const reason = { opaque: "exit observed before abort" }; + const controller = new AbortController(); + const worker = new WebBundleAnalysisWorker( + 128, + launcherFor( + { kind: "protocol", failure: "exit-7" }, + { + onSpawn: ({ process: child }) => { + if (abortAfterExit) + child.once("exit", () => + queueMicrotask(() => controller.abort(reason)), + ); + }, + }, + ), + ); + try { + const result = await worker.analyze( + makeScript("fetch('/exit-7');"), + 5_000, + controller.signal, + ); + const error = failureOf(result); + expect(error).toBeInstanceOf(AnalysisOutputError); + expect(error.capturedOutput?.exit_code).toBe(7); + expect(error.capturedOutput?.stderr).toContain("controlled exit 7"); + if (abortAfterExit) { + expect(error.secondaryFailures).toHaveLength(1); + expect(error.secondaryFailures[0]).toBeInstanceOf( + AnalysisCancelledError, + ); + expect(error.secondaryFailures[0]?.cause).toBe(reason); + } else { + expect(error.secondaryFailures).toHaveLength(0); + } + } finally { + await worker.close(); + } + } +}); + +it("keeps cancellation when abort is observed before a later child exit", async () => { + const controller = new AbortController(); + let childExit: Promise | undefined; + const worker = new WebBundleAnalysisWorker( + 128, + launcherFor( + { kind: "protocol", failure: "abort-first-exit" }, + { + onSpawn: ({ process: child }) => { + childExit = new Promise((resolve) => + child.once("exit", (code) => resolve(code)), + ); + child.stdout?.once("data", () => + queueMicrotask(() => controller.abort(null)), + ); + }, + }, + ), + ); + try { + const result = await worker.analyze( + makeScript("fetch('/abort-first');"), + 5_000, + controller.signal, + ); + const error = failureOf(result); + expect(error).toBeInstanceOf(AnalysisCancelledError); + expect(error.cause).toBeNull(); + expect(error.capturedOutput).toBeUndefined(); + expect(await childExit).toBe(7); + } finally { + await worker.close(); + } +}); + +it("keeps a code-0 exit pending while an inherited output pipe can still produce a frame", async () => { + for (const abortAfterExit of [false, true]) { + const reason = { opaque: "code-0 exit while inherited pipe is open" }; + const controller = new AbortController(); + const worker = new WebBundleAnalysisWorker( + 128, + launcherFor( + { kind: "protocol", failure: "zero-exit-held-pipe" }, + { + onSpawn: ({ process: child }) => { + if (abortAfterExit) + child.once("exit", () => + queueMicrotask(() => controller.abort(reason)), + ); + }, + }, + ), + ); + try { + const result = await worker.analyze( + makeScript("fetch('/zero-exit-held-pipe');"), + 5_000, + controller.signal, + ); + const error = failureOf(result); + if (abortAfterExit) { + expect(error).toBeInstanceOf(AnalysisCancelledError); + expect(error.cause).toBe(reason); + } else { + expect(error).toBeInstanceOf(AnalysisOutputError); + expect(error.capturedOutput?.exit_code).toBe(0); + } + } finally { + await worker.close(); + } + } +}); + +it("retains EOF without a terminal response when abort follows the close event", async () => { + const reason = { opaque: "abort queued after owned output close" }; + const controller = new AbortController(); + const worker = new WebBundleAnalysisWorker( + 128, + launcherFor( + { kind: "protocol", failure: "zero-exit-held-pipe" }, + { + onSpawn: ({ process: child }) => { + child.once("close", () => + queueMicrotask(() => controller.abort(reason)), + ); + }, + }, + ), + ); + try { + const result = await worker.analyze( + makeScript("fetch('/eof-before-abort');"), + 5_000, + controller.signal, + ); + const error = failureOf(result); + expect(error).toBeInstanceOf(AnalysisOutputError); + expect(error.message).toContain("output closed before completing"); + expect(error.capturedOutput?.exit_code).toBe(0); + expect(error.secondaryFailures).toHaveLength(1); + expect(error.secondaryFailures[0]).toBeInstanceOf(AnalysisCancelledError); + expect(error.secondaryFailures[0]?.cause).toBe(reason); + } finally { + await worker.close(); + } +}); + +it("retains startup EOF without a ready frame when abort follows the close event", async () => { + for (const abortAfterClose of [false, true]) { + const reason = { opaque: "abort queued after startup output close" }; + const controller = new AbortController(); + const worker = new WebBundleAnalysisWorker( + 128, + launcherFor( + { kind: "exit-before-ready" }, + { + onSpawn: ({ process: child }) => { + if (abortAfterClose) + child.once("close", () => + queueMicrotask(() => controller.abort(reason)), + ); + }, + }, + ), + ); + try { + const result = await worker.analyze( + makeScript("fetch('/startup-eof');"), + 5_000, + controller.signal, + ); + const error = failureOf(result); + expect(error).toBeInstanceOf(AnalysisOutputError); + expect(error.message).toContain("output closed before completing"); + expect(error.capturedOutput?.exit_code).toBe(0); + if (abortAfterClose) { + expect(error.secondaryFailures).toHaveLength(1); + expect(error.secondaryFailures[0]).toBeInstanceOf( + AnalysisCancelledError, + ); + expect(error.secondaryFailures[0]?.cause).toBe(reason); + } else { + expect(error.secondaryFailures).toHaveLength(0); + } + } finally { + await worker.close(); + } + } +}); + +it("retains exit-time truncation when abort follows an oversized stderr exit", async () => { + const reason = { opaque: "oversized stderr exit observed before abort" }; + const controller = new AbortController(); + const worker = new WebBundleAnalysisWorker( + 128, + launcherFor( + { kind: "protocol", failure: "truncated-exit" }, + { + onSpawn: ({ process: child }) => { + child.once("exit", () => + queueMicrotask(() => controller.abort(reason)), + ); + }, + }, + ), + ); + try { + const result = await worker.analyze( + makeScript("fetch('/truncated-exit');"), + 5_000, + controller.signal, + ); + const error = failureOf(result); + expect(error).toBeInstanceOf(AnalysisOutputError); + expect(error.capturedOutput?.exit_code).toBe(7); + expect(error.capturedOutput?.truncated).toBe(true); + expect(Buffer.byteLength(error.capturedOutput?.stderr ?? "")).toBeLessThan( + 300 * 1024, + ); + expect(error.secondaryFailures[0]?.cause).toBe(reason); + } finally { + await worker.close(); + } +}); + +it("keeps malformed protocol output primary when abort follows its observation", async () => { + const reason = { opaque: "malformed frame observed before abort" }; + for (const abortAfterFrame of [false, true]) { + const controller = new AbortController(); + const worker = new WebBundleAnalysisWorker( + 128, + launcherFor( + { kind: "protocol", failure: "malformed" }, + { + onSpawn: ({ process: child }) => { + if (abortAfterFrame) + child.stdout?.on("data", (chunk) => { + if (String(chunk).includes("{not-json}")) + queueMicrotask(() => controller.abort(reason)); + }); + }, + }, + ), + ); + try { + const result = await worker.analyze( + makeScript("fetch('/malformed');"), + 5_000, + controller.signal, + ); + const error = failureOf(result); + expect(error).toBeInstanceOf(AnalysisOutputError); + expect(error.message).toContain("Invalid web analyzer control frame"); + if (abortAfterFrame) { + expect(error.secondaryFailures).toHaveLength(1); + expect(error.secondaryFailures[0]?.cause).toBe(reason); + } else { + expect(error.secondaryFailures).toHaveLength(0); + } + } finally { + await worker.close(); + } + } +}); + +it("does not let a failure frame for another job replace cancellation", async () => { + const reason = { opaque: "unrelated frame observed before abort" }; + const controller = new AbortController(); + const worker = new WebBundleAnalysisWorker( + 128, + launcherFor( + { kind: "protocol", failure: "wrong-job-failure" }, + { + onSpawn: ({ process: child }) => { + child.stdout?.on("data", (chunk) => { + if (String(chunk).includes("failure for another request")) + queueMicrotask(() => controller.abort(reason)); + }); + }, + }, + ), + ); + try { + const result = await worker.analyze( + makeScript("fetch('/wrong-job');"), + 5_000, + controller.signal, + ); + const error = failureOf(result); + expect(error).toBeInstanceOf(AnalysisCancelledError); + expect(error.cause).toBe(reason); + expect(error.secondaryFailures).toHaveLength(0); + } finally { + await worker.close(); + } +}); + +it("keeps the resource budget on an observed worker resource frame", async () => { + const reason = { opaque: "resource frame before abort" }; + const controller = new AbortController(); + const worker = new WebBundleAnalysisWorker( + 128, + launcherFor( + { kind: "protocol", failure: "resource-failure" }, + { + onSpawn: ({ process: child }) => { + child.stdout?.on("data", (chunk) => { + if (String(chunk).includes("bounded result capacity")) + queueMicrotask(() => controller.abort(reason)); + }); + }, + }, + ), + ); + try { + const result = await worker.analyze( + makeScript("fetch('/resource-failure');"), + 5_000, + controller.signal, + ); + const error = failureOf(result); + if (!(error instanceof AnalysisResourceConstraintError)) + throw new Error("Expected a worker resource constraint failure"); + expect(error.reportedLimits?.result_budget_bytes).toBeGreaterThan(0); + expect(error.secondaryFailures[0]?.cause).toBe(reason); + } finally { + await worker.close(); + } +}); + +it("consumes a final frame before an inherited descendant pipe closes", async () => { + const worker = new WebBundleAnalysisWorker( + 128, + launcherFor({ kind: "protocol", failure: "final-frame-held-pipe" }), + ); + try { + const result = await worker.analyze( + makeScript("fetch('/descendant-frame');"), + 5_000, + ); + const error = failureOf(result); + expect(error).toBeInstanceOf(AnalysisOutputError); + expect(error.message).toContain("final frame from descendant"); + } finally { + await worker.close(); + } +}); + +it("does not classify heap-marker stderr from a healthy worker as exhaustion", async () => { + const worker = new WebBundleAnalysisWorker( + 128, + launcherFor({ + kind: "source", + stderrMarker: + "FATAL ERROR: Allocation failed - JavaScript heap out of memory", + }), + ); + try { + const result = await worker.analyze( + makeScript("fetch('/healthy');"), + 5_000, + ); + expect(result.ok).toBe(true); + if (result.ok) expect(result.value.status).toBe("analyzed"); + } finally { + await worker.close(); + } +}); + +it("projects late stderr tails with completeness from the actual selection point", async () => { + for (const abortAfterExit of [false, true]) { + const reason = { opaque: "abort after exit but after late stderr tail" }; + const controller = new AbortController(); + const worker = new WebBundleAnalysisWorker( + 128, + launcherFor( + { kind: "protocol", failure: "tail-exit" }, + { + onSpawn: ({ process: child }) => { + if (abortAfterExit) + child.once("exit", () => + setTimeout(() => controller.abort(reason), 150), + ); + }, + }, + ), + ); + try { + const result = await worker.analyze( + makeScript("fetch('/late-tail');"), + 5_000, + controller.signal, + ); + const error = failureOf(result); + expect(error).toBeInstanceOf(AnalysisOutputError); + expect(error.capturedOutput?.stderr).toContain("early exit marker"); + expect(error.capturedOutput?.stderr).toContain("late tail marker"); + expect(error.capturedOutput?.truncated).toBe(abortAfterExit); + if (abortAfterExit) + expect(error.secondaryFailures[0]?.cause).toBe(reason); + else expect(error.secondaryFailures).toHaveLength(0); + } finally { + await worker.close(); + } + } +}); + it("kills a hung owned child at the deadline and starts a fresh worker for the next script", async () => { const children: ProviderProcessHandle[] = []; const worker = new WebBundleAnalysisWorker( diff --git a/src/browser/WebBundleAnalysisWorker.ts b/src/browser/WebBundleAnalysisWorker.ts index 74a99c7182..ade9364c59 100644 --- a/src/browser/WebBundleAnalysisWorker.ts +++ b/src/browser/WebBundleAnalysisWorker.ts @@ -59,6 +59,30 @@ type WebBundleResultDescriptor = Extract< { readonly kind: "complete" } >["result"]; +type WebBundleObservedFailure = + | { readonly kind: "input"; readonly cause: Error } + | { readonly kind: "protocol"; readonly cause: unknown } + | { readonly kind: "incomplete-output" } + | { + readonly kind: "frame"; + readonly frame: Extract< + WebBundleWorkerMessage, + { readonly kind: "failure" } + >; + readonly resultBudgetBytes: number; + } + | { + readonly kind: "exit"; + readonly code: number | null; + readonly signal: NodeJS.Signals | null; + readonly heapExhaustion: boolean; + }; + +type WebBundleActiveJob = { + readonly id: string; + readonly resultBudgetBytes: number; +}; + const stableSnapshotValue = ( value: unknown, ): StableArtifactSnapshot | undefined => { @@ -90,9 +114,15 @@ export class WebBundleAnalysisWorker { #pending = ""; #frames: WebBundleWorkerMessage[] = []; #protocolFailure: unknown; + #observedFailure: WebBundleObservedFailure | undefined; + #activeDeadline: ProviderStartupDeadline | undefined; + #activeJob: WebBundleActiveJob | undefined; + #activeTerminalResponseObserved = false; + readonly #frameWaiters = new Set<(outcome: "frame" | "failure") => void>(); #retainedResultBytes = 0; #ready = false; readonly #stdout = (chunk: Buffer | string): void => { + if (this.#activeDeadline?.interruption !== undefined) return; if (this.#protocolFailure !== undefined) return; try { // Refuse the chunk before appending it to any retained control text. @@ -113,11 +143,15 @@ export class WebBundleAnalysisWorker { JSON.parse(this.#pending.slice(0, end)), ), ); + this.#observeFrame(this.#frames.at(-1)!); this.#pending = this.#pending.slice(end + 1); } + this.#wakeFrameWaiters("frame"); } catch (cause) { this.#pending = ""; this.#protocolFailure = cause; + this.#latchFailure({ kind: "protocol", cause }); + this.#wakeFrameWaiters("failure"); } }; @@ -133,9 +167,53 @@ export class WebBundleAnalysisWorker { rootPrefix: "rea-web-bundle-analysis-", launcher, retryCleanup: () => this.close(), + onInputFailure: (cause) => { + if (this.#activeDeadline?.interruption !== undefined) return; + this.#latchFailure({ kind: "input", cause }); + this.#wakeFrameWaiters("failure"); + }, supervisorOptions: { captureStdout: false, maxDiagnosticBytes: 256 * 1024, + onDiagnostic: (event) => { + if (this.#activeDeadline?.interruption !== undefined) return; + if ( + event.type === "exit" && + (event.code !== 0 || event.signal !== null) + ) { + const stderr = event.snapshot.stderr.text; + this.#latchFailure({ + kind: "exit", + code: event.code, + signal: event.signal, + heapExhaustion: + event.snapshot.memoryLimitReached === true || + /heap out of memory|heap limit|allocation failed/iu.test( + stderr, + ), + }); + } else if (event.type === "output-close") { + if ( + this.#activeDeadline !== undefined && + (this.#activeJob === undefined || + !this.#activeTerminalResponseObserved) + ) + this.#latchFailure({ kind: "incomplete-output" }); + this.#wakeFrameWaiters("failure"); + } else if (event.type === "output") { + const failure = this.#observedFailure; + if ( + failure?.kind === "exit" && + (this.#runtime.supervisor?.snapshot().memoryLimitReached === + true || + (event.stream === "stderr" && + /heap out of memory|heap limit|allocation failed/iu.test( + this.#runtime.supervisor?.snapshot().stderr.text ?? "", + ))) + ) + this.#observedFailure = { ...failure, heapExhaustion: true }; + } + }, }, }); const owner = this; @@ -161,6 +239,9 @@ export class WebBundleAnalysisWorker { ): Promise> { let deadline: ProviderStartupDeadline | undefined; const id = randomUUID(); + this.#observedFailure = undefined; + this.#activeJob = undefined; + this.#activeTerminalResponseObserved = false; let result: Result; let completed: WebBundleScriptAnalysis | null = null; let continuation: "result_capacity_exhausted" | undefined; @@ -199,6 +280,7 @@ export class WebBundleAnalysisWorker { } await prepareProcessOwnershipInspection(signal); deadline = new ProviderStartupDeadline(timeoutMs, signal); + this.#activeDeadline = deadline; await this.#start(deadline); if (this.#frames.length > 0 || this.#pending !== "") throw new AnalysisOutputError( @@ -223,6 +305,8 @@ export class WebBundleAnalysisWorker { const stdin = supervisor.launch.process.stdin; if (stdin === null || stdin === undefined) throw new TypeError("Web analyzer request stream is unavailable"); + this.#activeJob = { id, resultBudgetBytes: resultBudget }; + this.#activeTerminalResponseObserved = false; await new Promise((resolve, reject) => stdin.write(frame, (error) => error == null ? resolve() : reject(error), @@ -311,25 +395,33 @@ export class WebBundleAnalysisWorker { : deadline?.signal.reason, }) : undefined; + const observedFailure = + this.#observedFailure === undefined + ? undefined + : this.#projectObservedFailure(); + if (observedFailure !== undefined && interrupted !== undefined) + observedFailure.retainSecondaryFailure(interrupted); const failure = cause instanceof ArtifactFileCleanupFailure && cause.fileCleanupFailure.outcome.kind === "completed" ? this.#cleanupFailure(cause, interrupted) - : (interrupted ?? - (cause instanceof PrivateRuntimeRootUnavailableError - ? new AnalysisCapabilityUnavailableError( - PROVIDER, - OPERATION, - cause.message, - { cause, retryable: cause.retryable }, - ) - : cause instanceof AnalysisError - ? cause - : new AnalysisOutputError( + : interrupted !== undefined && observedFailure !== undefined + ? observedFailure + : (interrupted ?? + (cause instanceof PrivateRuntimeRootUnavailableError + ? new AnalysisCapabilityUnavailableError( + PROVIDER, OPERATION, - "Owned web-script analysis failed", - { cause }, - ))); + cause.message, + { cause, retryable: cause.retryable }, + ) + : cause instanceof AnalysisError + ? cause + : new AnalysisOutputError( + OPERATION, + "Owned web-script analysis failed", + { cause }, + ))); result = err(failure); try { const stopped = await this.#runtime.stop(); @@ -339,7 +431,10 @@ export class WebBundleAnalysisWorker { result = err(this.#cleanupFailure(cleanupCause, failure)); } } finally { + this.#activeDeadline = undefined; deadline?.dispose(); + this.#activeJob = undefined; + this.#activeTerminalResponseObserved = false; } // Never remove job files while a failed descriptor or child remains owned. if (!result.ok && result.error.cleanupIncomplete) @@ -455,43 +550,129 @@ export class WebBundleAnalysisWorker { cause: deadline.signal.reason, }); deadline.signal.throwIfAborted(); - if (this.#runtime.inputFailure !== undefined) - throw new AnalysisOutputError( - OPERATION, - "Web analyzer request stream failed", - { cause: this.#runtime.inputFailure }, - ); - if (this.#protocolFailure !== undefined) - throw new AnalysisOutputError( - OPERATION, - "Invalid web analyzer control frame", - { cause: this.#protocolFailure }, - ); const frame = this.#frames.shift(); if (frame !== undefined) return frame; + if (this.#runtime.inputFailure !== undefined) + this.#latchFailure({ + kind: "input", + cause: this.#runtime.inputFailure, + }); + if (this.#protocolFailure !== undefined) + this.#latchFailure({ kind: "protocol", cause: this.#protocolFailure }); + if ( + this.#observedFailure !== undefined && + this.#observedFailure.kind !== "exit" + ) + throw this.#projectObservedFailure(); const supervisor = this.#runtime.supervisor; const snapshot = supervisor?.snapshot(); if ( snapshot !== undefined && (snapshot.exitCode !== undefined || snapshot.signal !== undefined) ) { - if (!(await supervisor?.waitForOutputClose(10))) continue; - if (this.#frames.length > 0) continue; - const finalSnapshot = supervisor?.snapshot() ?? snapshot; - const capturedOutput = { - stdout: "", - stderr: finalSnapshot.stderr.text, - truncated: finalSnapshot.diagnosticTruncated ?? false, - exit_code: finalSnapshot.exitCode ?? null, - signal: finalSnapshot.signal ?? null, - }; - if ( - supervisor?.launch.process.memoryLimitReached === true || - /heap out of memory|heap limit|allocation failed/iu.test( - finalSnapshot.stderr.text, - ) - ) - throw new AnalysisResourceConstraintError( + if (supervisor === undefined) continue; + const outcome = await this.#waitForFrameOrOutputClose( + supervisor, + deadline, + ); + if (outcome === "interrupted") { + const completedFrame = this.#frames.shift(); + if (completedFrame !== undefined) return completedFrame; + throw deadline.interruption === "timeout" + ? new AnalysisTimeoutError(OPERATION, deadline.timeoutMs) + : new AnalysisCancelledError(OPERATION, { + cause: deadline.signal.reason, + }); + } + const completedFrame = this.#frames.shift(); + if (completedFrame !== undefined) return completedFrame; + if (outcome === "failure") throw this.#projectObservedFailure(); + if (outcome === "frame") continue; + if (this.#observedFailure !== undefined) + throw this.#projectObservedFailure(); + throw new AnalysisOutputError( + OPERATION, + "Owned web analyzer exited before completing this script", + { capturedOutput: this.#capturedOutput() }, + ); + } + await deadline.wait(10); + } + } + + #latchFailure(failure: WebBundleObservedFailure): void { + if ( + this.#observedFailure === undefined && + this.#activeDeadline?.interruption === undefined + ) + this.#observedFailure = failure; + } + + #observeFrame(frame: WebBundleWorkerMessage): void { + if ( + this.#activeDeadline?.interruption !== undefined || + this.#activeJob === undefined || + frame.kind === "ready" || + frame.id !== this.#activeJob.id + ) + return; + if (frame.kind === "complete" || frame.kind === "failure") + this.#activeTerminalResponseObserved = true; + if (frame.kind !== "failure") return; + this.#latchFailure({ + kind: "frame", + frame, + resultBudgetBytes: this.#activeJob.resultBudgetBytes, + }); + } + + #projectObservedFailure(): AnalysisError { + const failure = this.#observedFailure; + if (failure === undefined) + return new AnalysisOutputError( + OPERATION, + "Owned web analyzer exited before completing this script", + { capturedOutput: this.#capturedOutput() }, + ); + const capturedOutput = this.#capturedOutput(); + switch (failure.kind) { + case "input": + return new AnalysisOutputError( + OPERATION, + "Web analyzer request stream failed", + { cause: failure.cause, capturedOutput }, + ); + case "protocol": + return new AnalysisOutputError( + OPERATION, + "Invalid web analyzer control frame", + { cause: failure.cause, capturedOutput }, + ); + case "incomplete-output": + return new AnalysisOutputError( + OPERATION, + "Owned web analyzer output closed before completing this script", + { capturedOutput }, + ); + case "frame": + return failure.frame.reason === "resource" + ? new AnalysisResourceConstraintError( + OPERATION, + "memory", + failure.frame.message, + { + result_budget_bytes: failure.resultBudgetBytes, + }, + { capturedOutput }, + ) + : new AnalysisOutputError( + OPERATION, + `Web script analysis failed (${failure.frame.reason}): ${failure.frame.message}`, + { capturedOutput }, + ); + case "exit": + if (failure.heapExhaustion) + return new AnalysisResourceConstraintError( OPERATION, "memory", "Owned web analyzer exhausted its configured memory limit", @@ -502,13 +683,63 @@ export class WebBundleAnalysisWorker { }, { capturedOutput }, ); - throw new AnalysisOutputError( + return new AnalysisOutputError( OPERATION, "Owned web analyzer exited before completing this script", { capturedOutput }, ); - } - await deadline.wait(10); + } + } + + #capturedOutput() { + const supervisor = this.#runtime.supervisor; + const snapshot = supervisor?.snapshot(); + return { + stdout: "", + stderr: snapshot?.stderr.text ?? "", + truncated: + snapshot?.diagnosticTruncated === true || + snapshot?.outputIncomplete === true || + (supervisor !== undefined && !supervisor.observedOutputClose()), + exit_code: snapshot?.exitCode ?? null, + signal: snapshot?.signal ?? null, + }; + } + + #wakeFrameWaiters(outcome: "frame" | "failure"): void { + for (const wake of this.#frameWaiters) wake(outcome); + this.#frameWaiters.clear(); + } + + async #waitForFrameOrOutputClose( + supervisor: NonNullable, + deadline: ProviderStartupDeadline, + ): Promise<"frame" | "failure" | "closed" | "interrupted"> { + if (this.#frames.length > 0) return "frame"; + let wakeFrame!: (outcome: "frame" | "failure") => void; + const frameAvailable = new Promise<"frame" | "failure">((resolve) => { + wakeFrame = resolve; + this.#frameWaiters.add(wakeFrame); + }); + const outputCloseController = new AbortController(); + const interruptOutputClose = () => + outputCloseController.abort(deadline.signal.reason); + if (deadline.signal.aborted) interruptOutputClose(); + else + deadline.signal.addEventListener("abort", interruptOutputClose, { + once: true, + }); + try { + const outputClose = supervisor + .waitForOutputClose(undefined, outputCloseController.signal) + .then((closed): "closed" | "interrupted" => + closed ? "closed" : "interrupted", + ); + return await Promise.race([frameAvailable, outputClose]); + } finally { + outputCloseController.abort(); + deadline.signal.removeEventListener("abort", interruptOutputClose); + this.#frameWaiters.delete(wakeFrame); } } @@ -566,6 +797,11 @@ export class WebBundleAnalysisWorker { this.#pending = ""; this.#decoder = new StringDecoder("utf8"); this.#protocolFailure = undefined; + this.#observedFailure = undefined; + this.#activeDeadline = undefined; + this.#activeJob = undefined; + this.#activeTerminalResponseObserved = false; + this.#wakeFrameWaiters("failure"); } #cleanupFailure(cause: unknown, primary?: AnalysisError): AnalysisError { diff --git a/src/ghidra/GhidraClient.ts b/src/ghidra/GhidraClient.ts index 76a7fd81c7..a18f086ab0 100644 --- a/src/ghidra/GhidraClient.ts +++ b/src/ghidra/GhidraClient.ts @@ -1198,6 +1198,7 @@ export class GhidraClient { return; } this.#processSnapshot = event.snapshot; + if (event.type === "output-close") return; this.#options.onDiagnostic?.({ type: "launcher-exit", code: event.code, diff --git a/src/javascript/analysis/JavaScriptAnalysisWorker.ts b/src/javascript/analysis/JavaScriptAnalysisWorker.ts index 895d2f42ac..67ae3aaa08 100644 --- a/src/javascript/analysis/JavaScriptAnalysisWorker.ts +++ b/src/javascript/analysis/JavaScriptAnalysisWorker.ts @@ -96,6 +96,8 @@ export class JavaScriptAnalysisWorker { #primaryFailure: WorkerPrimaryFailure | undefined; #activeJobId: string | undefined; #activeJobPhase = "startup"; + #staticResponseId: string | undefined; + #terminalResponseId: string | undefined; #phase = "startup"; #lastFailure: AnalysisError | undefined; #estimatedRetainedBytes = 0; @@ -262,6 +264,8 @@ export class JavaScriptAnalysisWorker { throw new TypeError("JavaScript worker has no writable request stream"); this.#activeJobId = id; this.#activeJobPhase = this.#phase; + this.#staticResponseId = undefined; + this.#terminalResponseId = undefined; await new Promise((resolve, reject) => stdin.write(frame, (error) => error === null || error === undefined ? resolve() : reject(error), @@ -736,6 +740,7 @@ export class JavaScriptAnalysisWorker { stderr: snapshot?.stderr.text ?? "", truncated: snapshot?.diagnosticTruncated === true || + snapshot?.outputIncomplete === true || (supervisor !== undefined && !supervisor.observedOutputClose()), stderr_bytes: snapshot?.stderr.observedBytes ?? 0, exit_code: snapshot?.exitCode ?? null, @@ -745,6 +750,17 @@ export class JavaScriptAnalysisWorker { #observeProcessDiagnostic(event: ProviderProcessDiagnostic): void { if (this.#deadline?.interruption !== undefined) return; this.#observeMemoryLimit(); + if (event.type === "output-close") { + if ( + this.#deadline !== undefined && + (this.#activeJobId === undefined || + this.#terminalResponseId !== this.#activeJobId) + ) + this.#latchOutputFailure( + "JavaScript worker closed output before the operation received its terminal response", + ); + return; + } if ( event.type === "output" && event.stream === "stderr" && @@ -812,6 +828,24 @@ export class JavaScriptAnalysisWorker { this.#activeJobPhase = frame.phase; return; } + if (frame.kind === "static") { + this.#staticResponseId = frame.id; + return; + } + if (frame.kind === "complete") { + if (this.#staticResponseId !== frame.id) { + if (this.#primaryFailure === undefined) + this.#primaryFailure = { + kind: "output", + reason: "JavaScript worker completed before a static checkpoint", + }; + this.#wakeFrameWaiters("failure"); + return; + } + this.#terminalResponseId = frame.id; + return; + } + if (frame.kind === "failure") this.#terminalResponseId = frame.id; if (frame.kind === "failure" && this.#primaryFailure === undefined) this.#primaryFailure = { kind: "protocol", @@ -1101,6 +1135,9 @@ export class JavaScriptAnalysisWorker { this.#pending = ""; this.#decoder = new StringDecoder("utf8"); this.#primaryFailure = undefined; + this.#activeJobId = undefined; + this.#staticResponseId = undefined; + this.#terminalResponseId = undefined; } } diff --git a/src/process/ProviderProcess.ts b/src/process/ProviderProcess.ts index b1a8183f20..8b2c9265c9 100644 --- a/src/process/ProviderProcess.ts +++ b/src/process/ProviderProcess.ts @@ -178,6 +178,10 @@ export type ProviderProcessDiagnostic = readonly signal: NodeJS.Signals | null; readonly snapshot: ProviderProcessSnapshot; } + | { + readonly type: "output-close"; + readonly snapshot: ProviderProcessSnapshot; + } | { readonly type: "error"; readonly message: string; @@ -360,6 +364,11 @@ export class ProviderProcessSupervisor { ): void => { this.#recordExit(code, signal); this.#outputClosed = true; + if (!this.#disposed) + this.#options.onDiagnostic?.({ + type: "output-close", + snapshot: this.snapshot(), + }); this.#outputClose.resolve(undefined); }; readonly #onError = (cause: Error): void => { diff --git a/src/process/ProviderProcessOutput.test.ts b/src/process/ProviderProcessOutput.test.ts index 7ced2dfb10..3a5cc5cbf0 100644 --- a/src/process/ProviderProcessOutput.test.ts +++ b/src/process/ProviderProcessOutput.test.ts @@ -430,6 +430,101 @@ it("reports a reached process memory limit only when the handle observed one", ( } }); +it("reports output closure synchronously with the final incomplete-output snapshot", async () => { + const producer = Object.assign(new EventEmitter(), { + stdout: new PassThrough(), + stderr: new PassThrough(), + exitCode: null, + signalCode: null, + kill: () => false, + }); + const events: string[] = []; + let closeSnapshot: unknown; + const owner = new ProviderProcessSupervisor( + { process: producer, ownsProcessLifetime: false }, + { + onDiagnostic: (event) => { + if (event.type === "output-close") { + events.push("output-close"); + closeSnapshot = event.snapshot; + expect(owner.observedOutputClose()).toBe(true); + } + if (event.type === "exit") + queueMicrotask(() => events.push("exit-callback-microtask")); + }, + }, + ); + try { + producer.stdout.emit("error", new Error("output ended unexpectedly")); + producer.stdout.end(); + producer.stderr.end(); + producer.emit("close", 0, null); + expect(events).toEqual(["output-close"]); + expect(closeSnapshot).toMatchObject({ outputIncomplete: true }); + await Promise.resolve(); + expect(events).toEqual(["output-close", "exit-callback-microtask"]); + await expect(owner.waitForOutputClose(0)).resolves.toBe(true); + } finally { + owner.dispose(); + producer.stdout.destroy(); + producer.stderr.destroy(); + } +}); + +it("does not emit diagnostics after disposal when process close arrives", () => { + const producer = Object.assign(new EventEmitter(), { + stdout: new PassThrough(), + stderr: new PassThrough(), + exitCode: null, + signalCode: null, + kill: () => false, + }); + const events: string[] = []; + const owner = new ProviderProcessSupervisor( + { process: producer, ownsProcessLifetime: false }, + { onDiagnostic: (event) => events.push(event.type) }, + ); + owner.dispose(); + producer.emit("close", 0, null); + expect(events).toEqual([]); + producer.stdout.destroy(); + producer.stderr.destroy(); +}); + +it("emits output-close before queued abort microtasks and promise continuations", async () => { + const producer = Object.assign(new EventEmitter(), { + stdout: new PassThrough(), + stderr: new PassThrough(), + exitCode: null, + signalCode: null, + kill: () => false, + }); + const events: string[] = []; + let owner!: ProviderProcessSupervisor; + owner = new ProviderProcessSupervisor( + { process: producer, ownsProcessLifetime: false }, + { + onDiagnostic: (event) => { + if (event.type === "exit") + queueMicrotask(() => events.push("later-abort-microtask")); + if (event.type === "output-close") { + expect(owner.observedOutputClose()).toBe(true); + events.push("output-close"); + } + }, + }, + ); + const closeWait = owner.waitForOutputClose(); + producer.stdout.end(); + producer.stderr.end(); + producer.emit("close", 0, null); + expect(events).toEqual(["output-close"]); + await closeWait; + await Promise.resolve(); + expect(events).toEqual(["output-close", "later-abort-microtask"]); + owner.dispose(); +}); + it("projects provider output loss as truncated diagnostics without inventing a byte count", () => { const producer = Object.assign(new EventEmitter(), { stdout: new PassThrough(), diff --git a/tests/boundary/javascript/workerOrderAudit.test.ts b/tests/boundary/javascript/workerOrderAudit.test.ts index cb73eb1c8c..233147ea0e 100644 --- a/tests/boundary/javascript/workerOrderAudit.test.ts +++ b/tests/boundary/javascript/workerOrderAudit.test.ts @@ -20,6 +20,27 @@ const ready = 'process.stdout.write(JSON.stringify({kind:"ready",heap_limit_bytes:require("node:v8").getHeapStatistics().heap_size_limit})+"\\n");'; const phaseAfterRequest = 'process.stdin.once("data", chunk => { const request=JSON.parse(String(chunk)); process.stdout.write(JSON.stringify({kind:"phase",id:request.id,phase:"parse"})+"\\n");'; +const staticCheckpointFacts = { + parse_status: "complete", + parse_error_count: 0, + visited_ast_nodes: 1, + references: [], + endpoints: [], + storage: [], + bundler_registrations: [], + role_paths: [], + source_map_urls: [], + vendors: [], + electron: { + browser_windows: [], + context_bridge_apis: [], + ipc: [], + sender_validations: [], + utility_processes: [], + native_addon_bindings: [], + }, + limitations: [], +}; it.each([false, true])( "keeps an owned worker exit observed before caller abort as the primary result (memory limit: %s)", @@ -166,28 +187,7 @@ it("clears a consumed resource frame before the next reusable worker job", async }); it("keeps coalesced static and resource frames in FIFO order and reuses the worker", async () => { - const staticFacts = { - parse_status: "complete", - parse_error_count: 0, - visited_ast_nodes: 1, - references: [], - endpoints: [], - storage: [], - bundler_registrations: [], - role_paths: [], - source_map_urls: [], - vendors: [], - electron: { - browser_windows: [], - context_bridge_apis: [], - ipc: [], - sender_validations: [], - utility_processes: [], - native_addon_bindings: [], - }, - limitations: [], - }; - const encodedStatic = JSON.stringify(staticFacts); + const encodedStatic = JSON.stringify(staticCheckpointFacts); let launchCount = 0; const worker = new JavaScriptAnalysisWorker( { heapMb: 128, timeoutMs: 30_000 }, @@ -225,28 +225,7 @@ it("keeps coalesced static and resource frames in FIFO order and reuses the work it("retains coalesced static evidence when abort follows a resource frame", async () => { const controller = new AbortController(); - const staticFacts = { - parse_status: "complete", - parse_error_count: 0, - visited_ast_nodes: 1, - references: [], - endpoints: [], - storage: [], - bundler_registrations: [], - role_paths: [], - source_map_urls: [], - vendors: [], - electron: { - browser_windows: [], - context_bridge_apis: [], - ipc: [], - sender_validations: [], - utility_processes: [], - native_addon_bindings: [], - }, - limitations: [], - }; - const encodedStatic = JSON.stringify(staticFacts); + const encodedStatic = JSON.stringify(staticCheckpointFacts); const worker = new JavaScriptAnalysisWorker( { heapMb: 128, timeoutMs: 30_000 }, async (request) => { @@ -284,6 +263,249 @@ it("retains coalesced static evidence when abort follows a resource frame", asyn } }); +it("selects output EOF before a later abort when the current job has no terminal frame", async () => { + const controller = new AbortController(); + const worker = new JavaScriptAnalysisWorker( + { heapMb: 128, timeoutMs: 30_000 }, + async (request) => { + const launched = await spawnOwnedProviderProcess({ + ...request, + arguments: [ + "-e", + `${ready} process.stdin.once("data", () => process.exit(0));`, + ], + }); + launched.process.once("close", () => { + queueMicrotask(() => + controller.abort({ requested: "abort after EOF" }), + ); + }); + return launched; + }, + ); + try { + const result = await worker.analyze( + file, + { nodes: 100, unknowns: 100 }, + controller.signal, + ); + expect(controller.signal.aborted).toBe(true); + expect(result.ok).toBe(false); + if (!result.ok) + expect(result.error.error).toMatchObject({ + _tag: "AnalysisOutputError", + reason: + "JavaScript worker closed output before the operation received its terminal response", + }); + } finally { + await worker.close(); + } +}); + +it("selects startup EOF before an abort queued by the same close event", async () => { + const controller = new AbortController(); + const worker = new JavaScriptAnalysisWorker( + { heapMb: 128, timeoutMs: 30_000 }, + async (request) => { + const launched = await spawnOwnedProviderProcess({ + ...request, + arguments: ["-e", "process.exit(0);"], + }); + launched.process.once("close", () => { + queueMicrotask(() => + controller.abort({ requested: "abort after startup EOF" }), + ); + }); + return launched; + }, + ); + try { + const result = await worker.analyze( + file, + { nodes: 100, unknowns: 100 }, + controller.signal, + ); + expect(controller.signal.aborted).toBe(true); + expect(result.ok).toBe(false); + if (!result.ok) + expect(result.error.error).toMatchObject({ + _tag: "AnalysisOutputError", + reason: + "JavaScript worker closed output before the operation received its terminal response", + }); + } finally { + await worker.close(); + } +}); + +it("keeps a parsed terminal failure primary over startup-close abort", async () => { + const controller = new AbortController(); + const worker = new JavaScriptAnalysisWorker( + { heapMb: 128, timeoutMs: 30_000 }, + async (request) => { + const launched = await spawnOwnedProviderProcess({ + ...request, + arguments: [ + "-e", + `${ready} process.stdin.once("data", chunk => { const id=JSON.parse(String(chunk)).id; process.stdout.write(JSON.stringify({kind:"failure",id,reason:"output",resource:null,limits:null,message:"terminal failure before EOF"})+"\\n",()=>process.exit(0)); });`, + ], + }); + launched.process.once("close", () => { + queueMicrotask(() => + controller.abort({ requested: "abort after terminal EOF" }), + ); + }); + return launched; + }, + ); + try { + const result = await worker.analyze( + file, + { nodes: 100, unknowns: 100 }, + controller.signal, + ); + expect(controller.signal.aborted).toBe(true); + expect(result.ok).toBe(false); + if (!result.ok) + expect(result.error.error).toMatchObject({ + _tag: "AnalysisOutputError", + reason: "app.mjs during source-transfer: terminal failure before EOF", + }); + } finally { + await worker.close(); + } +}); + +it.each([false, true])( + "admits a valid complete response with static facts before EOF cancellation (%s)", + async (abortAfterTerminal) => { + const controller = new AbortController(); + const reason = { requested: "abort after valid terminal frame" }; + const encodedStatic = JSON.stringify(staticCheckpointFacts); + const worker = new JavaScriptAnalysisWorker( + { heapMb: 128, timeoutMs: 30_000 }, + async (request) => { + const launched = await spawnOwnedProviderProcess({ + ...request, + arguments: [ + "-e", + `${ready} const fs=require("node:fs"); const {createHash}=require("node:crypto"); const staticFacts=JSON.parse(${JSON.stringify(encodedStatic)}); let id; require("node:readline").createInterface({input:process.stdin}).on("line", line => { const request=JSON.parse(line); if(request.file){ id=request.id; const text=JSON.stringify(staticFacts); fs.writeFileSync(id+".static.json",text); process.stdout.write(JSON.stringify({kind:"static",id,bytes:Buffer.byteLength(text),sha256:createHash("sha256").update(text).digest("hex")})+"\\n"); } else { process.stdout.write(JSON.stringify({kind:"complete",id,descriptor:null})+"\\n",()=>process.exit(0)); } });`, + ], + }); + if (abortAfterTerminal) + launched.process.once("close", () => + queueMicrotask(() => controller.abort(reason)), + ); + return launched; + }, + ); + try { + const result = await worker.analyze( + file, + { nodes: 100, unknowns: 100 }, + controller.signal, + ); + expect(controller.signal.aborted).toBe(abortAfterTerminal); + if (abortAfterTerminal) { + expect(result.ok).toBe(false); + if (!result.ok) { + expect(result.error.error._tag).toBe("AnalysisCancelledError"); + expect(result.error.error.cause).toBe(reason); + expect(result.error.javascript?.parse_status).toBe("complete"); + } + } else { + expect(result).toMatchObject({ + ok: true, + value: { javascript: { parse_status: "complete" } }, + }); + } + } finally { + await worker.close(); + } + }, +); + +it.each([false, true])( + "keeps a complete-before-static protocol failure ahead of later abort (%s)", + async (abortAfterComplete) => { + const controller = new AbortController(); + const worker = new JavaScriptAnalysisWorker( + { heapMb: 128, timeoutMs: 30_000 }, + async (request) => { + const launched = await spawnOwnedProviderProcess({ + ...request, + arguments: [ + "-e", + `${ready} require("node:readline").createInterface({input:process.stdin}).once("line", line => { const {id}=JSON.parse(line); process.stdout.write(JSON.stringify({kind:"complete",id,descriptor:null})+"\\n"); setInterval(()=>{},1000); });`, + ], + }); + if (abortAfterComplete) { + const stdout = launched.process.stdout; + let readySeen = false; + stdout?.on("data", (chunk: Buffer | string) => { + if (!readySeen && String(chunk).includes('"kind":"ready"')) { + readySeen = true; + stdout.on("data", (responseChunk: Buffer | string) => { + if (String(responseChunk).includes('"kind":"complete"')) + queueMicrotask(() => controller.abort()); + }); + } + }); + } + return launched; + }, + ); + try { + const result = await worker.analyze( + file, + { nodes: 100, unknowns: 100 }, + controller.signal, + ); + expect(controller.signal.aborted).toBe(abortAfterComplete); + expect(result.ok).toBe(false); + if (!result.ok) { + expect(result.error.error).toMatchObject({ + _tag: "AnalysisOutputError", + reason: "JavaScript worker completed before a static checkpoint", + }); + } + } finally { + await worker.close(); + } + }, +); + +it("includes the owned process output-loss marker in captured truncation", async () => { + const worker = new JavaScriptAnalysisWorker( + { heapMb: 128, timeoutMs: 30_000 }, + async (request) => { + const launched = await spawnOwnedProviderProcess({ + ...request, + arguments: [ + "-e", + `${ready} process.stdin.once("data", () => process.exit(0));`, + ], + }); + Object.defineProperty(launched.process, "outputIncomplete", { + value: true, + }); + return launched; + }, + ); + try { + const result = await worker.analyze(file, { nodes: 100, unknowns: 100 }); + expect(result.ok).toBe(false); + if (!result.ok) { + expect(result.error.error._tag).toBe("AnalysisOutputError"); + expect(result.error.error.capturedOutput).toMatchObject({ + truncated: true, + }); + } + } finally { + await worker.close(); + } +}); + it("selects a control-frame budget failure observed before caller abort", async () => { const controller = new AbortController(); let stdoutBytes = 0;