diff --git a/src/browser/WebBundleAnalysisWorker.test.ts b/src/browser/WebBundleAnalysisWorker.test.ts index 6d405f3e07..ee1cf292fa 100644 --- a/src/browser/WebBundleAnalysisWorker.test.ts +++ b/src/browser/WebBundleAnalysisWorker.test.ts @@ -26,7 +26,10 @@ import { spawnOwnedProviderProcess, type ProviderProcessHandle, } from "../process/ProviderProcess.js"; -import { WebBundleAnalysisWorker } from "./WebBundleAnalysisWorker.js"; +import { + WebBundleAnalysisWorker, + WebBundleWorkerCleanup, +} from "./WebBundleAnalysisWorker.js"; import { WEB_BUNDLE_WORKER_LIMITS } from "./WebBundleAnalysisWorkerProtocol.js"; const makeScript = (text: string): IncludedScript => ({ @@ -268,7 +271,7 @@ const launcherFor = ( }; }; -const failureOf = ( +const failureOf = ( result: Awaited>, ): AnalysisError => { if (result.ok) throw new Error("Expected web worker failure"); @@ -1034,3 +1037,66 @@ it("preserves literal replacement characters in valid UTF-8 worker facts", async path: root, }); }); + +it("retries independent browser workers once while an earlier exact owner remains stuck", async () => { + const cleanup = new WebBundleWorkerCleanup(); + const failures = [true, true]; + const attempts = [0, 0]; + const roots: string[] = []; + let spawning = 0; + let releaseSpawns: () => void = () => {}; + const bothSpawning = new Promise((resolve) => { + releaseSpawns = resolve; + }); + const workers = [0, 1].map((index) => { + const launch = launcherFor({ kind: "protocol", failure: "malformed" }); + return new WebBundleAnalysisWorker( + 128, + async (request) => { + const child = await launch(request); + roots[index] = request.cwd!; + if (++spawning === 2) releaseSpawns(); + await bothSpawning; + return { + ...child, + cleanup: async () => { + attempts[index]! += 1; + return failures[index] + ? { + cleaned: false, + reason: `exact worker ${index} cleanup blocked`, + } + : child.cleanup!(); + }, + }; + }, + cleanup, + ); + }); + try { + const results = await Promise.all( + workers.map((worker) => + worker.analyze(makeScript("fetch('/mixed-cleanup');"), 10_000), + ), + ); + const errors = results.map(failureOf); + expect(errors.every((error) => error.cleanupIncomplete)).toBe(true); + expect(attempts).toEqual([1, 1]); + failures[1] = false; + await expect(cleanup.retry()).rejects.toThrow(); + expect(attempts).toEqual([2, 2]); + expect(errors[0]?.cleanupIncomplete).toBe(true); + expect(errors[1]?.cleanupIncomplete).toBe(false); + await access(roots[0]!); + await expect(access(roots[1]!)).rejects.toMatchObject({ code: "ENOENT" }); + failures[0] = false; + await cleanup.retry(); + expect(attempts).toEqual([3, 2]); + expect(errors[0]?.cleanupIncomplete).toBe(false); + await expect(access(roots[0]!)).rejects.toMatchObject({ code: "ENOENT" }); + } finally { + failures.fill(false); + await Promise.all(workers.map((worker) => worker.close())); + await cleanup.retry(); + } +}); diff --git a/src/browser/WebBundleAnalysisWorker.ts b/src/browser/WebBundleAnalysisWorker.ts index 34de81b81e..4e0da8e59e 100644 --- a/src/browser/WebBundleAnalysisWorker.ts +++ b/src/browser/WebBundleAnalysisWorker.ts @@ -7,6 +7,7 @@ import { join } from "node:path"; import { fileURLToPath } from "node:url"; import { StringDecoder } from "node:string_decoder"; import { getHeapStatistics } from "node:v8"; +import { ArtifactReaderFailure } from "../artifacts/ArtifactReader.js"; import { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js"; import { AnalysisCleanupLease, @@ -16,6 +17,8 @@ import { artifactFileCompletion, ArtifactFileCleanupFailure, artifactFileCancellationCause, + artifactFilePrimaryFacts, + assertArtifactReadActive, withArtifactFileHandle, } from "../artifacts/ArtifactFileOperation.js"; import { @@ -125,7 +128,6 @@ export interface WebBundleScriptAnalysisFailure { /** Sequential inert-source analysis; only bounded completed facts cross into the host. */ export class WebBundleAnalysisWorker { - readonly #workerCleanup: WebBundleWorkerCleanup; readonly #runtime: OwnedWorkerRuntime; readonly #files = new ArtifactResourceScope(); readonly #cleanup: AnalysisCleanupLease; @@ -184,7 +186,6 @@ export class WebBundleAnalysisWorker { launcher: typeof spawnOwnedProviderProcess = spawnOwnedProviderProcess, workerCleanup: WebBundleWorkerCleanup = processWebBundleCleanup, ) { - this.#workerCleanup = workerCleanup; if ( !Number.isSafeInteger(heapMb) || heapMb < MIN_WEB_BUNDLE_HEAP_MB || @@ -285,6 +286,7 @@ export class WebBundleAnalysisWorker { let completed: WebBundleScriptAnalysis | null = null; let continuation: "result_capacity_exhausted" | undefined; let resultDescriptor: WebBundleResultDescriptor | undefined; + let artifactFailure = false; const failed = ( error: AnalysisError, ): Result => @@ -321,9 +323,18 @@ export class WebBundleAnalysisWorker { if (root === undefined || supervisor === undefined) throw new TypeError("Web analyzer resources were not acquired"); supervisor.resetOutput(); - const source = await this.#files.run(() => - this.#writeSource(join(root, `${id}.source`), script, deadline!.signal), - ); + const source = await this.#files + .run(() => + this.#writeSource( + join(root, `${id}.source`), + script, + deadline!.signal, + ), + ) + .catch((cause: unknown) => { + artifactFailure = true; + throw cause; + }); const request = webBundleWorkerRequestSchema.parse({ kind: "analyze", ...(precision === undefined ? {} : { semantic_precision: precision }), @@ -383,15 +394,20 @@ export class WebBundleAnalysisWorker { "Completed script facts exceed the host's remaining reserved result capacity", { result_budget_bytes: resultBudget }, ); - const stored = await this.#files.run(() => - readStableArtifact( - join(root, `${id}.result`), - descriptor.bytes, - deadline!.signal, - undefined, - this.#files, - ), - ); + const stored = await this.#files + .run(() => + readStableArtifact( + join(root, `${id}.result`), + descriptor.bytes, + deadline!.signal, + undefined, + this.#files, + ), + ) + .catch((cause: unknown) => { + artifactFailure = true; + throw cause; + }); const facts = this.#validateCompletedResult(stored, descriptor, script); this.#retainedResultBytes += descriptor.bytes; completed = facts; @@ -411,15 +427,29 @@ export class WebBundleAnalysisWorker { // The physical read completed, but only validated facts are retained. } } - const cancellationCause = completion?.cancellation; + const artifactPrimary = artifactFailure + ? artifactFilePrimaryFacts(cause) + : undefined; + const cancellationCause = + completion?.cancellation ?? + (artifactPrimary?.origin === "owned-or-classified" && + artifactPrimary.cause instanceof ArtifactReaderFailure && + artifactPrimary.cause.reason === "cancelled" + ? { reason: artifactFileCancellationCause(artifactPrimary.cause) } + : undefined); + // File outcomes own their cancellation facts. A later signal cannot + // replace a rejected physical read/write that survived descriptor close. + const interruptionApplies = + !artifactFailure || cancellationCause !== undefined; const interrupted = - deadline?.interruption === "timeout" + interruptionApplies && deadline?.interruption === "timeout" ? scriptTimeout(timeoutMs, cause) : cancellationCause !== undefined ? new AnalysisCancelledError(OPERATION, { cause: cancellationCause.reason, }) - : deadline?.interruption === "cancelled" || signal?.aborted + : interruptionApplies && + (deadline?.interruption === "cancelled" || signal?.aborted) ? new AnalysisCancelledError(OPERATION, { cause: signal?.aborted ? signal.reason @@ -432,21 +462,28 @@ export class WebBundleAnalysisWorker { : this.#projectObservedFailure(); if (observedFailure !== undefined && interrupted !== undefined) observedFailure.retainSecondaryFailure(interrupted); + const primaryCause = + artifactPrimary === undefined ? cause : artifactPrimary.cause; + const primary = + interrupted !== undefined && observedFailure !== undefined + ? observedFailure + : (interrupted ?? + (primaryCause instanceof RuntimeControlUnavailableError + ? runtimeControlCapabilityError(PROVIDER, OPERATION, primaryCause) + : primaryCause instanceof AnalysisError + ? primaryCause + : new AnalysisOutputError( + OPERATION, + "Owned web-script analysis failed", + { cause: primaryCause }, + ))); const failure = - cause instanceof ArtifactFileCleanupFailure && completion !== undefined - ? this.#cleanupFailure(cause, interrupted) - : interrupted !== undefined && observedFailure !== undefined - ? observedFailure - : (interrupted ?? - (cause instanceof RuntimeControlUnavailableError - ? runtimeControlCapabilityError(PROVIDER, OPERATION, cause) - : cause instanceof AnalysisError - ? cause - : new AnalysisOutputError( - OPERATION, - "Owned web-script analysis failed", - { cause }, - ))); + cause instanceof ArtifactFileCleanupFailure + ? this.#cleanupFailure( + cause, + completion === undefined ? primary : interrupted, + ) + : primary; result = err(failure); try { const stopped = await this.#runtime.stop(); @@ -546,7 +583,7 @@ export class WebBundleAnalysisWorker { signal, async (handle) => { for (const part of bufferedJsonParts(jsonParts(script))) { - signal.throwIfAborted(); + assertArtifactReadActive(signal); const chunk = Buffer.from(part, "utf8"); if (chunk.length > WEB_BUNDLE_WORKER_LIMITS.sourceBytes - bytes) throw new AnalysisResourceConstraintError( @@ -835,7 +872,6 @@ export class WebBundleAnalysisWorker { } #cleanupFailure(cause: unknown, primary?: AnalysisError): AnalysisError { - this.#workerCleanup.retain(this); this.#runtime.retainForCleanup(); const cleanup = new ProviderCleanupError( PROVIDER, @@ -868,9 +904,7 @@ export class WebBundleAnalysisWorker { await this.#files.close(); await this.#runtime.close(); this.#reset(); - this.#workerCleanup.release(this); } catch (cause) { - this.#workerCleanup.retain(this); this.#runtime.retainForCleanup(); throw cause; } @@ -879,23 +913,13 @@ export class WebBundleAnalysisWorker { /** Web-analyzer workers and their runtimes retained by one host. */ export class WebBundleWorkerCleanup { - readonly #workers = new Set(); readonly runtimes: WorkerCleanupRegistry; constructor(runtimes?: WorkerCleanupRegistry) { this.runtimes = runtimes ?? new WorkerCleanupRegistry(); } - retain(worker: WebBundleAnalysisWorker): void { - this.#workers.add(worker); - } - - release(worker: WebBundleAnalysisWorker): void { - this.#workers.delete(worker); - } - async retry(): Promise { - for (const worker of [...this.#workers]) await worker.close(); await this.runtimes.retry(); } } diff --git a/src/browser/WebBundleAnalysisWorkerReadCompletion.test.ts b/src/browser/WebBundleAnalysisWorkerReadCompletion.test.ts index 08f009da7e..384bcdc53a 100644 --- a/src/browser/WebBundleAnalysisWorkerReadCompletion.test.ts +++ b/src/browser/WebBundleAnalysisWorkerReadCompletion.test.ts @@ -4,6 +4,9 @@ import { expect, it, vi } from "vitest"; const closeFailure = vi.hoisted(() => ({ enabled: false, + readFailure: undefined as Error | undefined, + onReadFailure: undefined as (() => void) | undefined, + onResultClose: undefined as (() => void) | undefined, resultCloseAttempts: 0, root: undefined as string | undefined, rootRemovedBeforeResultClose: false, @@ -25,6 +28,20 @@ vi.mock("node:fs/promises", async (importOriginal) => { open: async (...args: Parameters) => { const handle = await actual.open(...args); const path = args[0]; + if ( + closeFailure.readFailure !== undefined && + String(path).endsWith(".result") + ) { + const failure = closeFailure.readFailure; + closeFailure.readFailure = undefined; + Object.defineProperty(handle, "read", { + configurable: true, + value: async () => { + closeFailure.onReadFailure?.(); + throw failure; + }, + }); + } if (closeFailure.enabled && String(path).endsWith(".result")) { closeFailure.enabled = false; const close = handle.close.bind(handle); @@ -41,6 +58,19 @@ vi.mock("node:fs/promises", async (importOriginal) => { }, }); } + if ( + closeFailure.onResultClose !== undefined && + String(path).endsWith(".result") + ) { + const close = handle.close.bind(handle); + Object.defineProperty(handle, "close", { + configurable: true, + value: async () => { + await close(); + closeFailure.onResultClose?.(); + }, + }); + } return handle; }, }; @@ -48,6 +78,10 @@ vi.mock("node:fs/promises", async (importOriginal) => { import { createWebTextArtifact } from "../domain/webContentArtifact.js"; import type { IncludedScript } from "../domain/webBundleAnalyzerState.js"; +import { + AnalysisOutputError, + AnalysisCancelledError, +} from "../domain/analysisErrorCore.js"; import { ProviderCleanupError } from "../domain/providerCleanupError.js"; import { cleanupOwnedProcessGroup } from "../process/ProcessOwnership.js"; import { @@ -171,3 +205,76 @@ it("retains completed facts and resolves a result-file lease through another cle await OwnedWorkerRuntime.retryPending().catch(() => undefined); } }); + +it.each([false, true])( + "keeps a physical result read failure primary before late cancellation and close failure: %s", + async (failClose) => { + const worker = new WebBundleAnalysisWorker( + WEB_BUNDLE_WORKER_LIMITS.heapMb, + ownedSourceLauncher(), + ); + const physical = new Error("controlled physical result read failed"); + const controller = new AbortController(); + const reason = { source: "after physical read began" }; + closeFailure.enabled = failClose; + closeFailure.readFailure = physical; + closeFailure.onReadFailure = () => controller.abort(reason); + try { + const result = await worker.analyze( + makeScript("fetch('/physical-failure');"), + 10_000, + controller.signal, + ); + if (result.ok) throw new Error("Expected physical read failure"); + expect(result.error.completed).toBeNull(); + const primary = result.error.error.cleanupPrimary ?? result.error.error; + expect(primary).toBeInstanceOf(AnalysisOutputError); + expect(primary.cause).toBe(physical); + expect(primary.secondaryFailures).toHaveLength(1); + expect(primary.secondaryFailures[0]).toBeInstanceOf( + AnalysisCancelledError, + ); + expect(primary.secondaryFailures[0]?.cause).toBe(reason); + expect(result.error.error.cleanupIncomplete).toBe(failClose); + await worker.close(); + expect(result.error.error.cleanupIncomplete).toBe(false); + } finally { + closeFailure.enabled = false; + closeFailure.readFailure = undefined; + closeFailure.onReadFailure = undefined; + await worker.close(); + } + }, +); + +it("retains validated completed facts when cancellation arrives at result descriptor close", async () => { + const worker = new WebBundleAnalysisWorker( + WEB_BUNDLE_WORKER_LIMITS.heapMb, + ownedSourceLauncher(), + ); + const controller = new AbortController(); + const reason = { source: "completed result descriptor closed" }; + closeFailure.onResultClose = () => controller.abort(reason); + try { + const result = await worker.analyze( + makeScript("fetch('/completed-before-abort');"), + 10_000, + controller.signal, + ); + if (result.ok) throw new Error("Expected finalization cancellation"); + expect(result.error.error).toBeInstanceOf(AnalysisCancelledError); + expect(result.error.error.cause).toBe(reason); + expect(result.error.completed).toMatchObject({ + status: "analyzed", + observations: { + endpoints: [ + expect.objectContaining({ value: "/completed-before-abort" }), + ], + }, + }); + expect(result.error.error.cleanupIncomplete).toBe(false); + } finally { + closeFailure.onResultClose = undefined; + await worker.close(); + } +}); diff --git a/src/javascript/analysis/JavaScriptAnalysisWorker.ts b/src/javascript/analysis/JavaScriptAnalysisWorker.ts index e01df48ad2..4d79d41c38 100644 --- a/src/javascript/analysis/JavaScriptAnalysisWorker.ts +++ b/src/javascript/analysis/JavaScriptAnalysisWorker.ts @@ -401,12 +401,15 @@ export class JavaScriptAnalysisWorker { OPERATION, "JavaScript worker emitted a duplicate static checkpoint", ); + const previousPhase = this.#phase; + this.#phase = "checkpoint-read"; javascript = await this.#readCheckpoint( id, file.path, sourceIndex, response, ); + this.#phase = previousPhase; const exhausted = this.#parentCapacityFailure(file.path); if (exhausted !== undefined) throw exhausted; offeredResultBudget = @@ -1276,6 +1279,7 @@ export class JavaScriptAnalysisWorker { signal: AbortSignal | undefined, ): AnalysisError { if ( + this.#phase !== "checkpoint-read" && deadline?.interruption === "cancelled" && cause === deadline.signal.reason ) @@ -1299,8 +1303,13 @@ export class JavaScriptAnalysisWorker { cause.bindCleanupFailure(error); return error; } - // Ownership preparation is abortable before the worker deadline exists. - if (deadline?.interruption !== undefined) + // Checkpoint reads do not receive an AbortSignal. A rejected physical + // read or validation owns its failure even if the caller cancelled later. + // Ownership preparation remains abortable before the deadline exists. + if ( + this.#phase !== "checkpoint-read" && + deadline?.interruption !== undefined + ) return this.#interruptionFailure(path, deadline); if ( deadline === undefined && diff --git a/tests/boundary/javascript/workerFailureComposition.test.ts b/tests/boundary/javascript/workerFailureComposition.test.ts index 444b95ef30..ba49a90f20 100644 --- a/tests/boundary/javascript/workerFailureComposition.test.ts +++ b/tests/boundary/javascript/workerFailureComposition.test.ts @@ -4,6 +4,8 @@ import { afterEach, expect, it, vi } from "vitest"; const faults = vi.hoisted(() => ({ lowHeadroom: false, + onCheckpointRead: undefined as (() => void) | undefined, + checkpointReadFailure: undefined as Error | undefined, sourceRemoval: false, removals: 0, onSourceRemoval: undefined as (() => void) | undefined, @@ -30,6 +32,17 @@ vi.mock("node:fs/promises", async (original) => { const actual = await original(); return { ...actual, + readFile: async (...args: Parameters) => { + if ( + String(args[0]).includes("rea-javascript-analysis-") && + String(args[0]).endsWith(".static.json") + ) { + faults.onCheckpointRead?.(); + if (faults.checkpointReadFailure !== undefined) + throw faults.checkpointReadFailure; + } + return actual.readFile(...args); + }, rm: async (...args: Parameters) => { if ( String(args[0]).includes("rea-javascript-analysis-") && @@ -76,6 +89,8 @@ const ready = afterEach(() => { faults.lowHeadroom = false; + faults.onCheckpointRead = undefined; + faults.checkpointReadFailure = undefined; faults.sourceRemoval = false; faults.removals = 0; faults.onSourceRemoval = undefined; @@ -305,3 +320,64 @@ it.each([ await expect(access(runtimeRoot)).rejects.toMatchObject({ code: "ENOENT" }); }, ); + +it.each([ + { readFailed: false, deniedCleanup: false }, + { readFailed: true, deniedCleanup: false }, + { readFailed: true, deniedCleanup: true }, +])( + "preserves checkpoint physical outcome before late cancellation (read failed: $readFailed; cleanup denied: $deniedCleanup)", + async ({ readFailed, deniedCleanup }) => { + const controller = new AbortController(); + const cancellation = { source: "checkpoint read began" }; + const physical = new Error("controlled physical checkpoint read failed"); + faults.onCheckpointRead = () => controller.abort(cancellation); + faults.checkpointReadFailure = readFailed ? physical : undefined; + faults.sourceRemoval = deniedCleanup; + let retainedFailure: ProviderCleanupError | undefined; + const checkpoint = JSON.stringify(staticCheckpointFacts); + const worker = new JavaScriptAnalysisWorker( + { heapMb: 128, timeoutMs: 10_000 }, + async (request) => + spawnOwnedProviderProcess({ + ...request, + arguments: [ + "-e", + `${ready}const fs=require("node:fs");const {createHash}=require("node:crypto");const checkpoint=${JSON.stringify(checkpoint)};require("node:readline").createInterface({input:process.stdin}).on("line",line=>{const request=JSON.parse(line);if(request.file){fs.writeFileSync(request.id+".static.json",checkpoint);process.stdout.write(JSON.stringify({kind:"static",id:request.id,bytes:Buffer.byteLength(checkpoint),sha256:createHash("sha256").update(checkpoint).digest("hex")})+"\\n");}});`, + ], + }), + ); + try { + const result = await worker.analyze( + file, + { nodes: 100, unknowns: 100 }, + controller.signal, + ); + if (result.ok) + throw new Error("Expected cancellation or physical read failure"); + const failure = result.error.error; + const primary = failure.cleanupPrimary ?? failure; + expect(primary._tag).toBe( + readFailed ? "ProviderAdapterError" : "AnalysisCancelledError", + ); + expect(primary.cause).toBe(readFailed ? physical : cancellation); + expect(failure.cleanupIncomplete).toBe(deniedCleanup); + if (failure instanceof ProviderCleanupError) retainedFailure = failure; + if (readFailed) { + expect(result.error.javascript).toBeNull(); + expect(primary.secondaryFailures).toEqual([ + expect.objectContaining({ + _tag: "AnalysisCancelledError", + cause: cancellation, + }), + ]); + } else expect(result.error.javascript?.parse_status).toBe("complete"); + } finally { + faults.sourceRemoval = false; + await worker.close(); + } + expect(retainedFailure?.cleanupIncomplete).toBe( + deniedCleanup ? false : undefined, + ); + }, +);