fix(workers): retain cancellation through final cleanup

This commit is contained in:
morluto committed 2026-10-27 12:56:10 +08:00
1 parent 4572f31589
commit c66729eca9
6 files changed
+267 -60

No files matched your search

+14 -5
View File
@@ -1,3 +1,4 @@
import { analysisErrorWithCancellation } from "../domain/analysisErrorCancellation.js";
import { WorkerFrameSignal } from "../process/WorkerFrameSignal.js";
import { createHash, randomUUID } from "node:crypto";
import { open, rm } from "node:fs/promises";
@@ -463,13 +464,21 @@ export class WebBundleAnalysisWorker {
await rm(join(root, `${id}.result`), { force: true });
}
} catch (cause) {
return failed(
this.#cleanupFailure(cause, result.ok ? undefined : result.error),
);
const previous = result.ok ? undefined : result.error;
const primary = signal?.aborted
? analysisErrorWithCancellation(
previous,
new AnalysisCancelledError(OPERATION, { cause: signal.reason }),
)
: previous;
return failed(this.#cleanupFailure(cause, primary));
}
if (signal?.aborted && result.ok)
if (signal?.aborted)
return failed(
new AnalysisCancelledError(OPERATION, { cause: signal.reason }),
analysisErrorWithCancellation(
result.ok ? undefined : result.error,
new AnalysisCancelledError(OPERATION, { cause: signal.reason }),
),
);
return result.ok ? result : failed(result.error);
}
+24
View File
@@ -0,0 +1,24 @@
import {
primaryAnalysisError,
type AnalysisError,
} from "./analysisErrorBase.js";
import { AnalysisCancelledError } from "./analysisErrorCore.js";
/** Keep the established failure while retaining a later caller cancellation once. */
export const analysisErrorWithCancellation = (
failure: AnalysisError | undefined,
cancellation: AnalysisCancelledError,
): AnalysisError => {
if (failure === undefined) return cancellation;
const primary = primaryAnalysisError(failure);
if (
!(primary instanceof AnalysisCancelledError) &&
!primary.secondaryFailures.some(
(secondary) =>
secondary instanceof AnalysisCancelledError &&
secondary.cause === cancellation.cause,
)
)
primary.retainSecondaryFailure(cancellation);
return failure;
};
@@ -1,3 +1,4 @@
import { analysisErrorWithCancellation } from "../../domain/analysisErrorCancellation.js";
import { safeParseUtf8Json } from "../../domain/safeJson.js";
import { analysisSettingLimit } from "../../domain/analysisSettingDefinitions.js";
import { WorkerFrameSignal } from "../../process/WorkerFrameSignal.js";
@@ -541,7 +542,17 @@ export class JavaScriptAnalysisWorker {
await rm(join(rootPath, `${id}.${suffix}`), { force: true });
} catch (cause: unknown) {
this.#runtime.retainForCleanup();
const error = this.#cleanupError(message(cause));
const previous = result.ok ? undefined : result.error.error;
const primary = signal?.aborted
? analysisErrorWithCancellation(
previous,
new AnalysisCancelledError(OPERATION, {
cause: signal.reason,
capturedOutput: this.#captured(),
}),
)
: previous;
const error = this.#cleanupError(message(cause), primary);
this.#runtime.bindCleanupFailure(error);
result = err({
javascript,
@@ -551,12 +562,15 @@ export class JavaScriptAnalysisWorker {
});
}
}
if (result.ok && signal?.aborted === true) {
const error = new AnalysisCancelledError(OPERATION, {
cause: signal.reason,
capturedOutput: this.#captured(),
});
this.#lastFailure = error;
if (signal?.aborted === true) {
const error = analysisErrorWithCancellation(
result.ok ? undefined : result.error.error,
new AnalysisCancelledError(OPERATION, {
cause: signal.reason,
capturedOutput: this.#captured(),
}),
);
if (result.ok) this.#lastFailure = error;
result = err({ error, javascript, module, projection });
}
if (this.#applicationProjectionFailure !== undefined)
@@ -0,0 +1,119 @@
import { execFile } from "node:child_process";
import { promisify } from "node:util";
import { expect, it } from "vitest";
it.each([
[false, false],
[true, false],
[false, true],
[true, true],
])(
"retains completed web worker facts when cancellation arrives during final job cleanup (cleanup failure: %s; prior failure: %s)",
async (failCleanup, priorFailure) => {
const loaderUrl = new URL(
"../../../scripts/lib/verifier-source-loader.mjs",
import.meta.url,
).href;
const workerUrl = new URL(
"../../../src/browser/WebBundleAnalysisWorker.ts",
import.meta.url,
).href;
const processUrl = new URL(
"../../../src/process/ProviderProcess.ts",
import.meta.url,
).href;
// Isolate the filesystem seam in a Node process. Both analyses use the real
// analyzer. Failed final removal retains the real file for owning close retry.
const script = `
import fs from 'node:fs';
import { createHash } from 'node:crypto';
import { syncBuiltinESMExports } from 'node:module';
const { importVerifierSource } = await import(${JSON.stringify(loaderUrl)});
const controller = new AbortController();
const reason = new Error('cancel during final worker cleanup');
const nativeRm = fs.promises.rm;
let cancelOnCleanup = false;
let completedRemovals = 0;
fs.promises.rm = async (path, options) => {
const finalJobRemoval = String(path).includes('rea-web-bundle-analysis-') && String(path).endsWith('.result');
if (finalJobRemoval && cancelOnCleanup && ${JSON.stringify(failCleanup)}) {
controller.abort(reason);
throw new Error('fixture final removal denied');
}
await nativeRm(path, options);
if (finalJobRemoval) {
completedRemovals += 1;
if (cancelOnCleanup) controller.abort(reason);
}
};
syncBuiltinESMExports();
const { WebBundleAnalysisWorker } = await importVerifierSource(new URL(${JSON.stringify(workerUrl)}));
const { spawnOwnedProviderProcess } = await importVerifierSource(new URL(${JSON.stringify(processUrl)}));
const { createWebTextArtifact } = await importVerifierSource(new URL('../../../src/domain/webContentArtifact.ts', ${JSON.stringify(import.meta.url)}));
const value = 'export const observed = 7;';
const source = path => ({ script_key: 'scr_' + createHash('sha256').update(value).digest('hex'),
frame_id: 'frame-1', url: 'https://app.example.test/' + path, origin: 'https://app.example.test',
cdp_hash: null, length: Buffer.byteLength(value), is_module: true, language: 'JavaScript', source_map_url: null,
resource_reconciliation: {status: 'unmatched', reason: 'no_exact_sanitized_url'},
source: {included: true, artifact: createWebTextArtifact(value, 'application/javascript')} });
let root;
let child;
let launches = 0;
const worker = new WebBundleAnalysisWorker(128, async request => {
root = request.cwd;
const spawned = await spawnOwnedProviderProcess(${JSON.stringify(priorFailure)} ? {
...request,
arguments: ['-e', 'const send = value => process.stdout.write(JSON.stringify(value) + "\\n"); send({kind:"ready",heap_limit_bytes:require("node:v8").getHeapStatistics().heap_size_limit}); require("node:readline").createInterface({input:process.stdin}).on("line", line => { const request=JSON.parse(line); send({kind:"failure",id:request.id,reason:"output",message:"fixture output failure"}); });'],
} : request);
child = spawned.process;
launches += 1;
return spawned;
});
let observed;
try {
const control = await worker.analyze(source('control.mjs'), 5000);
cancelOnCleanup = true;
const cancelled = await worker.analyze(source('cancelled.mjs'), 5000, controller.signal);
const facts = cancelled.ok ? cancelled.value : cancelled.error;
const primary = cancelled.ok ? undefined : (cancelled.error.error.cleanupPrimary ?? cancelled.error.error);
const cancellation = primary?._tag === 'AnalysisCancelledError' ? primary : primary?.secondaryFailures.find(failure => failure._tag === 'AnalysisCancelledError');
observed = { primaryOutput: primary?._tag === 'AnalysisOutputError', secondaryCount: primary?.secondaryFailures.length ?? 0, controlOk: control.ok, completedRemovals, launches,
cancelled: cancellation !== undefined,
exactReason: cancellation?.cause === reason,
cleanupIncomplete: !cancelled.ok && cancelled.error.error.cleanupIncomplete,
hasFacts: facts.status === 'analyzed' || facts.completed?.status === 'analyzed' };
} finally {
fs.promises.rm = nativeRm;
syncBuiltinESMExports();
await worker.close();
}
const rootRemoved = await fs.promises.stat(root).then(() => false, cause => {
if (cause.code !== 'ENOENT') throw cause;
return true;
});
process.stdout.write(JSON.stringify({ ...observed, rootRemoved,
childExited: child.exitCode !== null || child.signalCode !== null }));
`;
const { stdout } = await promisify(execFile)(
process.execPath,
["--input-type=module", "--eval", script],
{ timeout: 20_000 },
);
const observed: unknown = JSON.parse(stdout);
expect(observed).toEqual({
primaryOutput: priorFailure,
secondaryCount: priorFailure ? 1 : 0,
controlOk: !priorFailure,
completedRemovals: failCleanup ? 1 : 2,
launches: priorFailure ? 2 : 1,
cancelled: true,
exactReason: true,
cleanupIncomplete: failCleanup,
hasFacts: !priorFailure,
rootRemoved: true,
childExited: true,
});
},
);
@@ -3,22 +3,24 @@ import { promisify } from "node:util";
import { expect, it } from "vitest";
it("retains completed worker facts when cancellation arrives during final job cleanup", async () => {
const loaderUrl = new URL(
"../../../scripts/lib/verifier-source-loader.mjs",
import.meta.url,
).href;
const workerUrl = new URL(
"../../../src/javascript/analysis/JavaScriptAnalysisWorker.ts",
import.meta.url,
).href;
const processUrl = new URL(
"../../../src/process/ProviderProcess.ts",
import.meta.url,
).href;
// Isolate the filesystem seam in a Node process. Both analyses use the real
// analyzer and remove their actual files before the cancellation is delivered.
const script = `
it.each([false, true])(
"retains completed worker facts when cancellation arrives during final job cleanup (failure: %s)",
async (failCleanup) => {
const loaderUrl = new URL(
"../../../scripts/lib/verifier-source-loader.mjs",
import.meta.url,
).href;
const workerUrl = new URL(
"../../../src/javascript/analysis/JavaScriptAnalysisWorker.ts",
import.meta.url,
).href;
const processUrl = new URL(
"../../../src/process/ProviderProcess.ts",
import.meta.url,
).href;
// Isolate the filesystem seam in a Node process. Both analyses use the real
// analyzer. Failed final removal retains the real file for owning close retry.
const script = `
import fs from 'node:fs';
import { createHash } from 'node:crypto';
import { syncBuiltinESMExports } from 'node:module';
@@ -29,8 +31,13 @@ it("retains completed worker facts when cancellation arrives during final job cl
let cancelOnCleanup = false;
let completedRemovals = 0;
fs.promises.rm = async (path, options) => {
const finalJobRemoval = String(path).includes('rea-javascript-analysis-') && String(path).endsWith('.records');
if (finalJobRemoval && cancelOnCleanup && ${JSON.stringify(failCleanup)}) {
controller.abort(reason);
throw new Error('fixture final removal denied');
}
await nativeRm(path, options);
if (String(path).includes('rea-javascript-analysis-') && String(path).endsWith('.records')) {
if (finalJobRemoval) {
completedRemovals += 1;
if (cancelOnCleanup) controller.abort(reason);
}
@@ -60,8 +67,9 @@ it("retains completed worker facts when cancellation arrives during final job cl
const cancelled = await worker.analyze(source('cancelled.mjs'), { nodes: 100, unknowns: 100 }, controller.signal);
const facts = cancelled.ok ? cancelled.value : cancelled.error;
observed = { controlOk: control.ok, controlReusable, completedRemovals, launches,
cancelled: !cancelled.ok && cancelled.error.error._tag === 'AnalysisCancelledError',
exactReason: !cancelled.ok && cancelled.error.error.cause === reason,
cancelled: !cancelled.ok && (cancelled.error.error.cleanupPrimary ?? cancelled.error.error)._tag === 'AnalysisCancelledError',
exactReason: !cancelled.ok && (cancelled.error.error.cleanupPrimary ?? cancelled.error.error).cause === reason,
cleanupIncomplete: !cancelled.ok && cancelled.error.error.cleanupIncomplete,
parseStatus: facts.javascript?.parse_status, hasModule: facts.module !== null,
hasProjection: facts.projection !== null, reusable: worker.canReuse() };
} finally {
@@ -76,24 +84,26 @@ it("retains completed worker facts when cancellation arrives during final job cl
process.stdout.write(JSON.stringify({ ...observed, rootRemoved,
childExited: child.exitCode !== null || child.signalCode !== null }));
`;
const { stdout } = await promisify(execFile)(
process.execPath,
["--input-type=module", "--eval", script],
{ timeout: 20_000 },
);
const observed: unknown = JSON.parse(stdout);
expect(observed).toEqual({
controlOk: true,
controlReusable: true,
completedRemovals: 2,
launches: 1,
cancelled: true,
exactReason: true,
parseStatus: "complete",
hasModule: true,
hasProjection: true,
reusable: false,
rootRemoved: true,
childExited: true,
});
});
const { stdout } = await promisify(execFile)(
process.execPath,
["--input-type=module", "--eval", script],
{ timeout: 20_000 },
);
const observed: unknown = JSON.parse(stdout);
expect(observed).toEqual({
controlOk: true,
controlReusable: true,
completedRemovals: failCleanup ? 1 : 2,
launches: 1,
cancelled: true,
exactReason: true,
cleanupIncomplete: failCleanup,
parseStatus: "complete",
hasModule: true,
hasProjection: true,
reusable: false,
rootRemoved: true,
childExited: true,
});
},
);
@@ -31,13 +31,17 @@ vi.mock("node:fs/promises", async (original) => {
return {
...actual,
rm: async (...args: Parameters<typeof actual.rm>) => {
if (
String(args[0]).includes("rea-javascript-analysis-") &&
String(args[0]).endsWith(".source")
)
faults.onSourceRemoval?.();
if (
faults.sourceRemoval &&
String(args[0]).includes("rea-javascript-analysis-") &&
String(args[0]).endsWith(".source")
) {
faults.removals += 1;
faults.onSourceRemoval?.();
throw Object.assign(
new Error("controlled source-file removal failure"),
{ code: "EACCES" },
@@ -77,9 +81,20 @@ afterEach(() => {
faults.onSourceRemoval = undefined;
});
it.each(["io", "output"] as const)(
"keeps worker %s failure classification distinct at the protocol boundary",
async (reason) => {
it.each([
["io", false, false],
["output", false, false],
["output", true, true],
["output", false, true],
] as const)(
"keeps worker %s failure classification distinct at the protocol boundary (cleanup: %s; cancellation: %s)",
async (reason, deniedCleanup, cancelCleanup) => {
const controller = new AbortController();
const cancellation = new Error("cancel during failed job cleanup");
faults.sourceRemoval = deniedCleanup;
faults.onSourceRemoval = cancelCleanup
? () => controller.abort(cancellation)
: undefined;
const worker = new JavaScriptAnalysisWorker(
{ heapMb: 128, timeoutMs: 30_000 },
async (request) =>
@@ -92,13 +107,26 @@ it.each(["io", "output"] as const)(
}),
);
try {
const result = await worker.analyze(file, { nodes: 100, unknowns: 100 });
const result = await worker.analyze(
file,
{ nodes: 100, unknowns: 100 },
controller.signal,
);
expect(result.ok).toBe(false);
if (result.ok) throw new Error("Expected worker failure");
expect(result.error.error._tag).toBe(
const failure = result.error.error.cleanupPrimary ?? result.error.error;
expect(result.error.error.cleanupIncomplete).toBe(deniedCleanup);
if (cancelCleanup) {
expect(failure.secondaryFailures).toHaveLength(1);
expect(failure.secondaryFailures[0]).toMatchObject({
_tag: "AnalysisCancelledError",
cause: cancellation,
});
}
expect(failure._tag).toBe(
reason === "io" ? "ProviderAdapterError" : "AnalysisOutputError",
);
const projected = projectAnalysisError(result.error.error);
const projected = projectAnalysisError(failure);
expect(projected.code).toBe(
reason === "io" ? "execution_failure" : "unreadable_output",
);
@@ -112,6 +140,7 @@ it.each(["io", "output"] as const)(
expect(projected.message).toContain("storage failed");
}
} finally {
faults.sourceRemoval = false;
await worker.close();
}
},
@@ -124,7 +153,9 @@ it.each([false, true])(
faults.sourceRemoval = removalFailure;
const controller = new AbortController();
const reason = new Error("cancel during denied source-file cleanup");
faults.onSourceRemoval = () => controller.abort(reason);
faults.onSourceRemoval = removalFailure
? () => controller.abort(reason)
: undefined;
const sourceMapUrls = [
{
declared_url: "app.map",