mirror of
https://github.com/morluto/rea.git
synced 2026-11-12 09:52:27 +00:00
fix(recovery): preserve bounded Wakaru failure and cleanup facts (#10087)
* fix(javascript): retain Wakaru diagnostics and cleanup ownership * fix(recovery): preserve launch failure with late cancellation * fix(recovery): retain cancellation through provider cleanup
This commit is contained in:
1 parent
8a2eb4ab61
commit
a16f5f4b5d
5 files changed
+650
-37
No files matched your search
@@ -1,4 +1,9 @@
|
||||
import { EventEmitter } from "node:events";
|
||||
import { join } from "node:path";
|
||||
import { tmpdir } from "node:os";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { spawnOwnedProviderProcess } from "../../process/ProviderProcess.js";
|
||||
import { execFileOutput } from "../../process/ExecFileOutput.js";
|
||||
import { PassThrough } from "node:stream";
|
||||
import { expect, it } from "vitest";
|
||||
import {
|
||||
@@ -6,6 +11,7 @@ import {
|
||||
AnalysisTimeoutError,
|
||||
} from "../../domain/analysisErrorCore.js";
|
||||
import { projectAnalysisError } from "../../domain/analysisErrorProjection.js";
|
||||
import { primaryAnalysisError } from "../../domain/analysisErrorBase.js";
|
||||
import {
|
||||
resolveWakaruCommand,
|
||||
runWakaruCommand,
|
||||
@@ -90,9 +96,10 @@ it.each([null, false, { request: "cancel launch" }])(
|
||||
environment: {},
|
||||
deadline: performance.now() + 5_000,
|
||||
signal: controller.signal,
|
||||
launcher: async () => {
|
||||
launcher: async (options) => {
|
||||
controller.abort(reason);
|
||||
throw new Error("launcher interrupted");
|
||||
options.signal?.throwIfAborted();
|
||||
throw new Error("Expected launcher cancellation");
|
||||
},
|
||||
}),
|
||||
).rejects.toMatchObject({
|
||||
@@ -102,6 +109,67 @@ it.each([null, false, { request: "cancel launch" }])(
|
||||
},
|
||||
);
|
||||
|
||||
it("preserves cancellation reported by the ownership preparation command", async () => {
|
||||
const controller = new AbortController();
|
||||
const reason = { caller: "cancel ownership preparation" };
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
await expect(
|
||||
runWakaruCommand({
|
||||
command: "/selected/wakaru",
|
||||
limiter: "/unused/prlimit",
|
||||
arguments: ["--version"],
|
||||
cwd: tmpdir(),
|
||||
environment: {},
|
||||
deadline: performance.now() + 10_000,
|
||||
signal: controller.signal,
|
||||
launcher: async () => {
|
||||
timer = setTimeout(() => controller.abort(reason), 100);
|
||||
await execFileOutput(
|
||||
process.execPath,
|
||||
["-e", "setInterval(() => {}, 1000)"],
|
||||
{
|
||||
signal: controller.signal,
|
||||
stopSignal: "SIGTERM",
|
||||
},
|
||||
);
|
||||
throw new Error("Expected preparation cancellation");
|
||||
},
|
||||
}),
|
||||
).rejects.toMatchObject({ _tag: "AnalysisCancelledError", cause: reason });
|
||||
} finally {
|
||||
if (timer !== undefined) clearTimeout(timer);
|
||||
}
|
||||
});
|
||||
|
||||
it("preserves a real launch failure when the caller cancels after it", async () => {
|
||||
const controller = new AbortController();
|
||||
const reason = { caller: "cancel after launch failure" };
|
||||
const missing = join(tmpdir(), `rea-missing-wakaru-${randomUUID()}`);
|
||||
const failure: unknown = await runWakaruCommand({
|
||||
command: "/selected/wakaru",
|
||||
limiter: missing,
|
||||
arguments: ["--version"],
|
||||
cwd: tmpdir(),
|
||||
environment: {},
|
||||
deadline: performance.now() + 30_000,
|
||||
signal: controller.signal,
|
||||
launcher: async (options) => {
|
||||
try {
|
||||
return await spawnOwnedProviderProcess(options);
|
||||
} catch (cause) {
|
||||
controller.abort(reason);
|
||||
throw cause;
|
||||
}
|
||||
},
|
||||
}).catch((cause: unknown) => cause);
|
||||
expect(failure).toMatchObject({
|
||||
_tag: "ProviderAdapterError",
|
||||
cause: { code: "ENOENT" },
|
||||
secondaryFailures: [{ _tag: "AnalysisCancelledError", cause: reason }],
|
||||
});
|
||||
});
|
||||
|
||||
const commandFixture = (
|
||||
emit: (producer: ReturnType<typeof createProducer>) => void,
|
||||
) => {
|
||||
@@ -169,12 +237,67 @@ it("checks the diagnostic byte limit against output delivered after exit", async
|
||||
producer.emit("close", 0, null);
|
||||
});
|
||||
});
|
||||
await expect(runWakaruCommand(context)).rejects.toMatchObject({
|
||||
const failure = await runWakaruCommand(context).catch(
|
||||
(cause: unknown) => cause,
|
||||
);
|
||||
expect(failure).toMatchObject({
|
||||
_tag: "AnalysisOutputError",
|
||||
operation: "recover_javascript_sources",
|
||||
capturedOutput: {
|
||||
truncated: true,
|
||||
exit_code: 0,
|
||||
},
|
||||
});
|
||||
expect(
|
||||
(failure as { capturedOutput: { stderr: string } }).capturedOutput.stderr
|
||||
.length,
|
||||
).toBe(RECOVERY_LIMITS.reportBytes);
|
||||
});
|
||||
|
||||
it.each([false, true])(
|
||||
"retains output and exact cancellation during final cleanup (cleanup fails: %s)",
|
||||
async (cleanupFails) => {
|
||||
const reason = { caller: "cancel during final cleanup" };
|
||||
const controller = new AbortController();
|
||||
const { context } = commandFixture((producer) => {
|
||||
producer.stdout.end("finished output");
|
||||
producer.stderr.end("finished diagnostic");
|
||||
producer.emit("exit", 0, null);
|
||||
producer.emit("close", 0, null);
|
||||
});
|
||||
const launcher = context.launcher;
|
||||
const failure = await runWakaruCommand({
|
||||
...context,
|
||||
signal: controller.signal,
|
||||
launcher: async (options) => ({
|
||||
...(await launcher(options)),
|
||||
cleanup: async () => {
|
||||
controller.abort(reason);
|
||||
return cleanupFails
|
||||
? { cleaned: false as const, reason: "owned cleanup refused" }
|
||||
: { cleaned: true as const, signaled: false };
|
||||
},
|
||||
}),
|
||||
}).catch((cause: unknown) => cause);
|
||||
expect(failure).toMatchObject({
|
||||
cleanupIncomplete: cleanupFails,
|
||||
capturedOutput: {
|
||||
stdout: "finished output",
|
||||
stderr: "finished diagnostic",
|
||||
exit_code: 0,
|
||||
},
|
||||
});
|
||||
const primary = primaryAnalysisError(
|
||||
failure as Parameters<typeof primaryAnalysisError>[0],
|
||||
);
|
||||
expect(primary).toMatchObject({
|
||||
_tag: "AnalysisCancelledError",
|
||||
cause: reason,
|
||||
});
|
||||
expect(primary.cause).toBe(reason);
|
||||
},
|
||||
);
|
||||
|
||||
it("enforces the shared deadline while exited output streams remain open", async () => {
|
||||
const { context, producer } = commandFixture((process) => {
|
||||
process.emit("exit", 0, null);
|
||||
|
||||
@@ -13,7 +13,10 @@ import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js";
|
||||
import type { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js";
|
||||
import { ProviderAdapterError } from "../../domain/providerAdapterError.js";
|
||||
import { ProviderCleanupError } from "../../domain/providerCleanupError.js";
|
||||
import { AnalysisError } from "../../domain/analysisErrorBase.js";
|
||||
import { analysisErrorWithCancellation } from "../../domain/analysisErrorCancellation.js";
|
||||
import { prlimitArgumentVector } from "../../process/LinuxResourceLimits.js";
|
||||
import { ExecFileOutputCancelledError } from "../../process/ExecFileOutput.js";
|
||||
import { cleanupOwnedProcessGroup } from "../../process/ProcessOwnership.js";
|
||||
import {
|
||||
ProviderProcessSupervisor,
|
||||
@@ -33,6 +36,17 @@ const OPERATION = "recover_javascript_sources";
|
||||
const WAKARU_REQUIREMENT =
|
||||
"Provide an absolute REA_WAKARU_COMMAND pointing to a Wakaru ^1.13.0 Linux x64 binary (verified with 1.14.0); no tool is installed by REA. See docs/guides/javascript/javascript-recovery.md#install-wakaru.";
|
||||
|
||||
const commandCapturedOutput = (
|
||||
snapshot: ReturnType<ProviderProcessSupervisor["snapshot"]>,
|
||||
) => ({
|
||||
stdout: snapshot.stdout.text,
|
||||
stderr: snapshot.stderr.text,
|
||||
exit_code: snapshot.exitCode ?? null,
|
||||
signal: snapshot.signal ?? null,
|
||||
truncated:
|
||||
snapshot.diagnosticTruncated === true || snapshot.outputIncomplete === true,
|
||||
});
|
||||
|
||||
/** Wakaru recovery runs only on Linux x64; discovery and execution share this check. */
|
||||
export const wakaruHostSupported = (
|
||||
platform: NodeJS.Platform,
|
||||
@@ -64,7 +78,14 @@ export class WakaruCleanupFailure extends ProviderCleanupError {
|
||||
exit_code: snapshot.exitCode ?? null,
|
||||
signal: snapshot.signal ?? null,
|
||||
},
|
||||
{ operation: OPERATION, cause: previousError },
|
||||
{
|
||||
operation: OPERATION,
|
||||
cause: previousError,
|
||||
...(previousError instanceof AnalysisError
|
||||
? { cleanupPrimary: previousError }
|
||||
: {}),
|
||||
capturedOutput: commandCapturedOutput(snapshot),
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -158,11 +179,15 @@ export const runWakaruCommand = async (context: WakaruCommandContext) => {
|
||||
...(context.signal === undefined ? {} : { signal: context.signal }),
|
||||
env: { TMPDIR: context.cwd, RAYON_NUM_THREADS: "1", NO_COLOR: "1" },
|
||||
}).catch((cause: unknown) => {
|
||||
if (context.signal?.aborted === true)
|
||||
throw new AnalysisCancelledError(OPERATION, {
|
||||
cause: context.signal.reason,
|
||||
});
|
||||
throw new ProviderAdapterError("wakaru", OPERATION, {
|
||||
const signal = context.signal;
|
||||
if (
|
||||
signal?.aborted === true &&
|
||||
(Object.is(cause, signal.reason) ||
|
||||
(cause instanceof ExecFileOutputCancelledError &&
|
||||
Object.is(cause.cause, signal.reason)))
|
||||
)
|
||||
throw new AnalysisCancelledError(OPERATION, { cause: signal.reason });
|
||||
const failure = new ProviderAdapterError("wakaru", OPERATION, {
|
||||
cause,
|
||||
diagnostics: {
|
||||
reason: "Failed to launch configured Wakaru through util-linux prlimit",
|
||||
@@ -170,6 +195,12 @@ export const runWakaruCommand = async (context: WakaruCommandContext) => {
|
||||
detail: recoveryFailureMessage(cause),
|
||||
},
|
||||
});
|
||||
throw signal?.aborted === true
|
||||
? analysisErrorWithCancellation(
|
||||
failure,
|
||||
new AnalysisCancelledError(OPERATION, { cause: signal.reason }),
|
||||
)
|
||||
: failure;
|
||||
});
|
||||
let oversized = false;
|
||||
const supervisor = new ProviderProcessSupervisor(
|
||||
@@ -180,6 +211,10 @@ export const runWakaruCommand = async (context: WakaruCommandContext) => {
|
||||
spawned.cleanup ?? (() => cleanupOwnedProcessGroup(spawned.ownership)),
|
||||
},
|
||||
{
|
||||
maxDiagnosticBytes: {
|
||||
stdout: RECOVERY_LIMITS.reportBytes,
|
||||
stderr: RECOVERY_LIMITS.reportBytes,
|
||||
},
|
||||
onDiagnostic: (event) => {
|
||||
if (
|
||||
event.type === "output" &&
|
||||
@@ -222,6 +257,15 @@ export const runWakaruCommand = async (context: WakaruCommandContext) => {
|
||||
}
|
||||
const stopped = await supervisor.stop();
|
||||
const snapshot = supervisor.snapshot();
|
||||
if (context.signal?.aborted === true) {
|
||||
const cancelled = new AnalysisCancelledError(OPERATION, {
|
||||
cause: context.signal.reason,
|
||||
});
|
||||
if (failure === undefined || failure instanceof AnalysisError)
|
||||
failure = analysisErrorWithCancellation(failure, cancelled);
|
||||
}
|
||||
if (failure instanceof AnalysisError)
|
||||
failure.retainCapturedOutput(commandCapturedOutput(snapshot));
|
||||
if (stopped.status === "incomplete")
|
||||
throw new WakaruCleanupFailure(
|
||||
supervisor,
|
||||
@@ -232,10 +276,6 @@ export const runWakaruCommand = async (context: WakaruCommandContext) => {
|
||||
snapshot,
|
||||
);
|
||||
if (failure !== undefined) throw failure;
|
||||
if (context.signal?.aborted === true)
|
||||
throw new AnalysisCancelledError(OPERATION, {
|
||||
cause: context.signal.reason,
|
||||
});
|
||||
return {
|
||||
exit_code: snapshot.exitCode ?? null,
|
||||
signal: snapshot.signal ?? null,
|
||||
|
||||
@@ -12,8 +12,10 @@ import { AnalysisCleanupLease } from "../../artifacts/AnalysisCleanupLease.js";
|
||||
import { SafeOutputTree } from "../../artifacts/SafeOutputTree.js";
|
||||
import { SafeOutputTreeCreationFailure } from "../../artifacts/SafeOutputTreeCreationFailure.js";
|
||||
import { AnalysisError } from "../../domain/analysisErrorBase.js";
|
||||
import { analysisErrorWithCancellation } from "../../domain/analysisErrorCancellation.js";
|
||||
import {
|
||||
AnalysisCancelledError,
|
||||
AnalysisCapabilityUnavailableError,
|
||||
AnalysisOutputError,
|
||||
} from "../../domain/analysisErrorCore.js";
|
||||
import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js";
|
||||
@@ -75,28 +77,9 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
|
||||
input: JavaScriptRecoveryInput,
|
||||
options?: ExecutionOptions,
|
||||
): Promise<Result<AnalysisExecution, AnalysisError>> {
|
||||
if (this.#closing)
|
||||
return Promise.resolve(
|
||||
err(
|
||||
new ProviderCleanupError(
|
||||
"wakaru",
|
||||
[],
|
||||
{ reason: "Provider is closing" },
|
||||
{ operation: OPERATION },
|
||||
),
|
||||
),
|
||||
);
|
||||
if (this.#closing) return Promise.resolve(err(this.#closedError()));
|
||||
const operation = this.#tail.then(() =>
|
||||
this.#closing
|
||||
? err(
|
||||
new ProviderCleanupError(
|
||||
"wakaru",
|
||||
[],
|
||||
{ reason: "Provider is closing" },
|
||||
{ operation: OPERATION },
|
||||
),
|
||||
)
|
||||
: this.#recover(input, options),
|
||||
this.#closing ? err(this.#closedError()) : this.#recover(input, options),
|
||||
);
|
||||
this.#tail = operation.then(
|
||||
() => undefined,
|
||||
@@ -105,6 +88,14 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
|
||||
return operation;
|
||||
}
|
||||
|
||||
#closedError(): AnalysisCapabilityUnavailableError {
|
||||
return new AnalysisCapabilityUnavailableError(
|
||||
"wakaru",
|
||||
OPERATION,
|
||||
"Provider is closing",
|
||||
);
|
||||
}
|
||||
|
||||
async #recover(
|
||||
input: JavaScriptRecoveryInput,
|
||||
options?: ExecutionOptions,
|
||||
@@ -172,7 +163,7 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
|
||||
? failed.cause
|
||||
: undefined;
|
||||
const primary = reportFailure?.failure ?? failed.cause;
|
||||
const failure =
|
||||
let failure =
|
||||
primary instanceof ArtifactReaderFailure &&
|
||||
primary.reason === "cancelled" &&
|
||||
primary.cleanup === undefined
|
||||
@@ -258,6 +249,29 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
|
||||
if (root !== undefined) residuals.push(root);
|
||||
root = undefined;
|
||||
}
|
||||
// Cleanup can settle after the caller cancels. Preserve the established
|
||||
// physical outcome and record the exact later cancellation on its primary.
|
||||
const signal = options?.signal;
|
||||
if (signal?.aborted === true) {
|
||||
const analysisFailure =
|
||||
failure instanceof AnalysisError
|
||||
? failure
|
||||
: new ProviderAdapterError("wakaru", OPERATION, {
|
||||
cause: failure,
|
||||
diagnostics: { reason: recoveryFailureMessage(failure) },
|
||||
...(failure instanceof ArtifactReaderFailure &&
|
||||
failure.partialObservation !== undefined
|
||||
? { partialObservation: failure.partialObservation }
|
||||
: {}),
|
||||
...(reportFailure === undefined
|
||||
? {}
|
||||
: { capturedOutput: reportFailure.capturedOutput }),
|
||||
});
|
||||
failure = analysisErrorWithCancellation(
|
||||
analysisFailure,
|
||||
new AnalysisCancelledError(OPERATION, { cause: signal.reason }),
|
||||
);
|
||||
}
|
||||
if (residuals.length > 0) {
|
||||
const error = new ProviderCleanupError(
|
||||
"wakaru",
|
||||
@@ -275,6 +289,12 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
|
||||
{
|
||||
operation: OPERATION,
|
||||
cause: failure,
|
||||
...(failure instanceof AnalysisError
|
||||
? {
|
||||
cleanupPrimary: failure,
|
||||
capturedOutput: failure.capturedOutput,
|
||||
}
|
||||
: {}),
|
||||
...(reportFailure === undefined
|
||||
? {}
|
||||
: { capturedOutput: reportFailure.capturedOutput }),
|
||||
@@ -287,7 +307,8 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
|
||||
),
|
||||
},
|
||||
}
|
||||
: failure instanceof ArtifactReaderFailure &&
|
||||
: (failure instanceof AnalysisError ||
|
||||
failure instanceof ArtifactReaderFailure) &&
|
||||
failure.partialObservation !== undefined
|
||||
? { partialObservation: failure.partialObservation }
|
||||
: {}),
|
||||
@@ -295,6 +316,8 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
|
||||
);
|
||||
const lease = this.#pendingCleanup?.lease;
|
||||
if (lease !== undefined) {
|
||||
if (failure instanceof WakaruCleanupFailure)
|
||||
lease.bindCleanupError(failure);
|
||||
lease.bindCleanupError(error);
|
||||
error.bindCleanupResourceOwner(lease);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
import { expect, it } from "vitest";
|
||||
import { AnalysisCancelledError } from "../../../src/domain/analysisErrorCore.js";
|
||||
import { primaryAnalysisError } from "../../../src/domain/analysisErrorBase.js";
|
||||
import { runWakaruCommand } from "../../../src/javascript/recovery/WakaruCommand.js";
|
||||
import { RECOVERY_LIMITS } from "../../../src/javascript/recovery/WakaruRelease.js";
|
||||
import {
|
||||
spawnOwnedProviderProcess,
|
||||
type SpawnedOwnedProviderProcess,
|
||||
} from "../../../src/process/ProviderProcess.js";
|
||||
import { cleanupOwnedProcessGroup } from "../../../src/process/ProcessOwnership.js";
|
||||
|
||||
// Only this authored Node diagnostic producer runs: no Wakaru, prlimit, or
|
||||
// selected/recovered application source is executed by this fixture.
|
||||
it.skipIf(process.platform === "win32").each([
|
||||
{
|
||||
bytes: RECOVERY_LIMITS.reportBytes - 1,
|
||||
overflow: false,
|
||||
lateCancel: false,
|
||||
},
|
||||
{ bytes: RECOVERY_LIMITS.reportBytes, overflow: false, lateCancel: false },
|
||||
{ bytes: RECOVERY_LIMITS.reportBytes + 1, overflow: true, lateCancel: true },
|
||||
{ bytes: 17, overflow: false, lateCancel: true },
|
||||
])(
|
||||
"preserves per-stream allowance and physical nonzero exit for $bytes diagnostic bytes (late cancellation: $lateCancel)",
|
||||
async ({ bytes, overflow, lateCancel }) => {
|
||||
const controller = new AbortController();
|
||||
const reason = { caller: "late abort after owned fixture cleanup" };
|
||||
let spawned: SpawnedOwnedProviderProcess | undefined;
|
||||
try {
|
||||
const result = await runWakaruCommand({
|
||||
command: "/unused/wakaru",
|
||||
limiter: "/unused/prlimit",
|
||||
arguments: [],
|
||||
cwd: process.cwd(),
|
||||
environment: {},
|
||||
deadline: performance.now() + 10_000,
|
||||
signal: controller.signal,
|
||||
launcher: async (options) => {
|
||||
spawned = await spawnOwnedProviderProcess({
|
||||
...options,
|
||||
command: process.execPath,
|
||||
arguments: [
|
||||
"-e",
|
||||
`process.stderr.write(Buffer.alloc(${bytes}, 97), () => { process.stdout.write('owned fixture stdout'); process.exitCode = 17; });`,
|
||||
],
|
||||
});
|
||||
const owned = spawned;
|
||||
return {
|
||||
...owned,
|
||||
cleanup: async () => {
|
||||
const cleanup = await cleanupOwnedProcessGroup(owned.ownership);
|
||||
if (lateCancel) controller.abort(reason);
|
||||
return cleanup;
|
||||
},
|
||||
};
|
||||
},
|
||||
}).then(
|
||||
(value) => ({ ok: true as const, value }),
|
||||
(error: unknown) => ({ ok: false as const, error }),
|
||||
);
|
||||
if (!overflow && !lateCancel) {
|
||||
if (!result.ok) throw result.error;
|
||||
expect(result.value).toMatchObject({
|
||||
exit_code: 17,
|
||||
signal: null,
|
||||
stdout: "owned fixture stdout",
|
||||
});
|
||||
expect(Buffer.byteLength(result.value.stderr)).toBe(bytes);
|
||||
return;
|
||||
}
|
||||
if (!overflow) {
|
||||
if (result.ok) throw new Error("Expected late caller cancellation");
|
||||
const primary = primaryAnalysisError(
|
||||
result.error as Parameters<typeof primaryAnalysisError>[0],
|
||||
);
|
||||
expect(primary).toMatchObject({
|
||||
_tag: "AnalysisCancelledError",
|
||||
capturedOutput: {
|
||||
exit_code: 17,
|
||||
signal: null,
|
||||
stdout: "owned fixture stdout",
|
||||
truncated: false,
|
||||
},
|
||||
cleanupIncomplete: false,
|
||||
});
|
||||
expect(primary.cause).toBe(reason);
|
||||
expect(Buffer.byteLength(primary.capturedOutput!.stderr)).toBe(bytes);
|
||||
return;
|
||||
}
|
||||
expect(result).toMatchObject({
|
||||
ok: false,
|
||||
error: {
|
||||
_tag: "AnalysisOutputError",
|
||||
capturedOutput: { truncated: true },
|
||||
cleanupIncomplete: false,
|
||||
},
|
||||
});
|
||||
if (result.ok) throw new Error("Expected diagnostic overflow refusal");
|
||||
const primary = primaryAnalysisError(
|
||||
result.error as Parameters<typeof primaryAnalysisError>[0],
|
||||
);
|
||||
expect(Buffer.byteLength(primary.capturedOutput!.stderr)).toBe(
|
||||
RECOVERY_LIMITS.reportBytes,
|
||||
);
|
||||
expect(
|
||||
primary.secondaryFailures.some(
|
||||
(failure) =>
|
||||
failure instanceof AnalysisCancelledError &&
|
||||
failure.cause === reason,
|
||||
),
|
||||
).toBe(true);
|
||||
} finally {
|
||||
if (spawned !== undefined) {
|
||||
const cleanup = await cleanupOwnedProcessGroup(spawned.ownership);
|
||||
expect(cleanup.cleaned).toBe(true);
|
||||
}
|
||||
}
|
||||
},
|
||||
15_000,
|
||||
);
|
||||
@@ -1,23 +1,75 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { access, writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { Readable } from "node:stream";
|
||||
import { Readable, PassThrough } from "node:stream";
|
||||
import { EventEmitter } from "node:events";
|
||||
import { expect, it, vi } from "vitest";
|
||||
import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
|
||||
|
||||
const publicationCancellation = vi.hoisted(() => ({
|
||||
controller: undefined as AbortController | undefined,
|
||||
directorySyncs: 0,
|
||||
commitController: undefined as AbortController | undefined,
|
||||
commitParent: undefined as string | undefined,
|
||||
failPublishedClose: false,
|
||||
publishedCloseAttempts: 0,
|
||||
cleanupController: undefined as AbortController | undefined,
|
||||
cleanupRoot: undefined as string | undefined,
|
||||
physicalFailure: undefined as Error | undefined,
|
||||
failRootCleanup: false,
|
||||
reportCleanup: false,
|
||||
allowCleanup: false,
|
||||
commandFailure: undefined as
|
||||
| "cancelled"
|
||||
| "timeout"
|
||||
| "output-limit"
|
||||
| undefined,
|
||||
}));
|
||||
|
||||
vi.mock("node:fs/promises", async (importOriginal) => {
|
||||
const filesystem = await importOriginal<typeof import("node:fs/promises")>();
|
||||
return {
|
||||
...filesystem,
|
||||
rm: async (...args: Parameters<typeof filesystem.rm>) => {
|
||||
if (String(args[0]) === publicationCancellation.cleanupRoot) {
|
||||
publicationCancellation.cleanupController?.abort({
|
||||
stage: "workspace cleanup",
|
||||
});
|
||||
if (publicationCancellation.failRootCleanup)
|
||||
throw Object.assign(new Error("workspace cleanup denied"), {
|
||||
code: "EACCES",
|
||||
});
|
||||
}
|
||||
return filesystem.rm(...args);
|
||||
},
|
||||
open: async (...args: Parameters<typeof filesystem.open>) => {
|
||||
const handle = await filesystem.open(...args);
|
||||
if (
|
||||
publicationCancellation.commitController !== undefined &&
|
||||
String(args[0]) === publicationCancellation.commitParent
|
||||
) {
|
||||
const sync = handle.sync.bind(handle);
|
||||
const close = handle.close.bind(handle);
|
||||
let synced = false;
|
||||
handle.sync = async () => {
|
||||
await sync();
|
||||
synced = true;
|
||||
};
|
||||
handle.close = async () => {
|
||||
if (synced) {
|
||||
publicationCancellation.publishedCloseAttempts += 1;
|
||||
publicationCancellation.commitController?.abort({
|
||||
stage: "published parent close",
|
||||
});
|
||||
if (
|
||||
publicationCancellation.failPublishedClose &&
|
||||
publicationCancellation.publishedCloseAttempts === 1
|
||||
)
|
||||
throw Object.assign(new Error("close denied"), { code: "EIO" });
|
||||
}
|
||||
await close();
|
||||
};
|
||||
}
|
||||
if (publicationCancellation.controller !== undefined) {
|
||||
const stats = await handle.stat();
|
||||
if (stats.isDirectory()) {
|
||||
@@ -74,6 +126,73 @@ vi.mock(
|
||||
prepareWakaruExecution: async (
|
||||
context: Parameters<typeof actual.prepareWakaruExecution>[0],
|
||||
) => {
|
||||
if (publicationCancellation.physicalFailure !== undefined) {
|
||||
publicationCancellation.cleanupRoot = context.root;
|
||||
throw publicationCancellation.physicalFailure;
|
||||
}
|
||||
if (publicationCancellation.commandFailure !== undefined) {
|
||||
const { runWakaruCommand } =
|
||||
await import("../../../src/javascript/recovery/WakaruCommand.js");
|
||||
const { RECOVERY_LIMITS } =
|
||||
await import("../../../src/javascript/recovery/WakaruRelease.js");
|
||||
const kind = publicationCancellation.commandFailure;
|
||||
return runWakaruCommand({
|
||||
command: "/unused/wakaru",
|
||||
limiter: "/unused/prlimit",
|
||||
arguments: [],
|
||||
cwd: context.root,
|
||||
environment: {},
|
||||
deadline: performance.now() + (kind === "timeout" ? 10 : 5000),
|
||||
...(context.options?.signal === undefined
|
||||
? {}
|
||||
: { signal: context.options.signal }),
|
||||
launcher: async (options) => {
|
||||
const child = Object.assign(new EventEmitter(), {
|
||||
stdout: new PassThrough(),
|
||||
stderr: new PassThrough(),
|
||||
exitCode: null,
|
||||
signalCode: null,
|
||||
kill: () => false,
|
||||
});
|
||||
setTimeout(
|
||||
() => {
|
||||
child.stdout.end("finished command output");
|
||||
child.stderr.end(
|
||||
kind === "output-limit"
|
||||
? Buffer.alloc(RECOVERY_LIMITS.reportBytes + 1, 97)
|
||||
: "finished diagnostic",
|
||||
);
|
||||
child.emit("exit", 0, null);
|
||||
child.emit("close", 0, null);
|
||||
},
|
||||
kind === "timeout" ? 25 : 0,
|
||||
);
|
||||
return {
|
||||
process: child,
|
||||
ownership: {
|
||||
runId: options.runId,
|
||||
leaderPid: 12345,
|
||||
processGroupId: 12345,
|
||||
expectedParentPid: process.pid,
|
||||
},
|
||||
cleanup: async () => {
|
||||
if (kind === "cancelled")
|
||||
publicationCancellation.controller?.abort({
|
||||
caller: "stop at owned command cleanup",
|
||||
});
|
||||
return publicationCancellation.allowCleanup
|
||||
? { cleaned: true as const, signaled: false }
|
||||
: {
|
||||
cleaned: false as const,
|
||||
reason: "controlled command cleanup failure",
|
||||
};
|
||||
},
|
||||
};
|
||||
},
|
||||
}).then(() => {
|
||||
throw new Error("Expected controlled command failure");
|
||||
});
|
||||
}
|
||||
if (publicationCancellation.reportCleanup) {
|
||||
const cleanup = await context.resources.release({
|
||||
kind: "lease",
|
||||
@@ -126,6 +245,88 @@ vi.mock(
|
||||
import { AnalysisCancelledError } from "../../../src/domain/analysisErrorCore.js";
|
||||
import { projectAnalysisError } from "../../../src/domain/analysisErrorProjection.js";
|
||||
import { WakaruProvider } from "../../../src/javascript/recovery/WakaruProvider.js";
|
||||
import { JavaScriptRecoveryService } from "../../../src/application/javascript/JavaScriptRecoveryService.js";
|
||||
import {
|
||||
primaryAnalysisError,
|
||||
resolvedAnalysisError,
|
||||
} from "../../../src/domain/analysisErrorBase.js";
|
||||
|
||||
it.each(["queued at close", "after clean close"])(
|
||||
"refuses Wakaru recovery %s without claiming cleanup ownership",
|
||||
async (stage) => {
|
||||
const provider = new WakaruProvider({});
|
||||
const input = {
|
||||
path: "/unused/source.js",
|
||||
output_directory: "/unused/output",
|
||||
extraction_mode: "structural" as const,
|
||||
rewrite_level: "standard" as const,
|
||||
};
|
||||
const queued =
|
||||
stage === "queued at close" ? provider.recover(input) : undefined;
|
||||
await provider.close();
|
||||
const result = await (queued ?? provider.recover(input));
|
||||
if (result.ok) throw new Error("Expected closed-provider refusal");
|
||||
expect(projectAnalysisError(result.error).code).toBe(
|
||||
"capability_unavailable",
|
||||
);
|
||||
expect(result.error.cleanupIncomplete).toBe(false);
|
||||
expect(result.error.cleanupResources).toEqual([]);
|
||||
},
|
||||
);
|
||||
|
||||
it.each(["cancelled", "timeout", "output-limit"] as const)(
|
||||
"preserves %s command output and cleanup primary through service retry",
|
||||
async (kind) => {
|
||||
const root = await createTestTempDirectory(
|
||||
"rea-wakaru-command-composition-",
|
||||
);
|
||||
const source = join(root, "source.js");
|
||||
await writeFile(source, "module.exports = 42;");
|
||||
publicationCancellation.commandFailure = kind;
|
||||
publicationCancellation.allowCleanup = false;
|
||||
const controller = new AbortController();
|
||||
publicationCancellation.controller = controller;
|
||||
const service = new JavaScriptRecoveryService(new WakaruProvider({}));
|
||||
try {
|
||||
const result = await service.recover(
|
||||
{
|
||||
path: source,
|
||||
output_directory: join(root, "output"),
|
||||
extraction_mode: "structural",
|
||||
rewrite_level: "standard",
|
||||
},
|
||||
{ signal: controller.signal },
|
||||
);
|
||||
if (result.ok) throw new Error("Expected command cleanup failure");
|
||||
expect(result.error.capturedOutput?.stdout).toBe(
|
||||
"finished command output",
|
||||
);
|
||||
expect(result.error.cleanupIncomplete).toBe(true);
|
||||
const primary = primaryAnalysisError(result.error);
|
||||
expect(primary._tag).toBe(
|
||||
kind === "cancelled"
|
||||
? "AnalysisCancelledError"
|
||||
: kind === "timeout"
|
||||
? "AnalysisTimeoutError"
|
||||
: "AnalysisOutputError",
|
||||
);
|
||||
if (kind === "cancelled")
|
||||
expect(primary.cause).toBe(controller.signal.reason);
|
||||
publicationCancellation.allowCleanup = true;
|
||||
await service.close();
|
||||
expect(result.error.cleanupIncomplete).toBe(false);
|
||||
expect(resolvedAnalysisError(result.error)).toBe(primary);
|
||||
expect(result.error.capturedOutput?.stdout).toBe(
|
||||
"finished command output",
|
||||
);
|
||||
} finally {
|
||||
publicationCancellation.allowCleanup = true;
|
||||
publicationCancellation.commandFailure = undefined;
|
||||
publicationCancellation.controller = undefined;
|
||||
await service.close();
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it("retains report command output and cleanup ownership through provider retry", async () => {
|
||||
const root = await createTestTempDirectory("rea-wakaru-report-cleanup-");
|
||||
@@ -149,6 +350,11 @@ it("retains report command output and cleanup ownership through provider retry",
|
||||
exit_code: 1,
|
||||
signal: null,
|
||||
});
|
||||
await expect(provider.close()).rejects.toMatchObject({
|
||||
cleanupIncomplete: true,
|
||||
});
|
||||
expect(result.error.cleanupIncomplete).toBe(true);
|
||||
expect(result.error.cleanupResources).toContain("provenance descriptor");
|
||||
publicationCancellation.allowCleanup = true;
|
||||
await provider.close();
|
||||
expect(result.error.cleanupIncomplete).toBe(false);
|
||||
@@ -193,3 +399,104 @@ it.skipIf(process.platform === "win32")(
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it.each([false, true])(
|
||||
"retains a physical failure and later workspace-cleanup cancellation (cleanup fails: %s)",
|
||||
async (cleanupFails) => {
|
||||
const root = await createTestTempDirectory("rea-recovery-late-cancel-");
|
||||
const source = join(root, "source.js");
|
||||
await writeFile(source, "module.exports = 42;");
|
||||
const controller = new AbortController();
|
||||
const physical = new Error("physical recovery failure");
|
||||
publicationCancellation.cleanupController = controller;
|
||||
publicationCancellation.physicalFailure = physical;
|
||||
publicationCancellation.failRootCleanup = cleanupFails;
|
||||
const provider = new WakaruProvider({});
|
||||
try {
|
||||
const result = await provider.recover(
|
||||
{
|
||||
path: source,
|
||||
output_directory: join(root, "output"),
|
||||
extraction_mode: "structural",
|
||||
rewrite_level: "standard",
|
||||
},
|
||||
{ signal: controller.signal },
|
||||
);
|
||||
if (result.ok) throw new Error("Expected physical failure");
|
||||
expect(controller.signal.aborted).toBe(true);
|
||||
const primary = primaryAnalysisError(result.error);
|
||||
expect(primary).toMatchObject({
|
||||
_tag: "ProviderAdapterError",
|
||||
cause: physical,
|
||||
});
|
||||
expect(
|
||||
primary.secondaryFailures.some(
|
||||
(failure) =>
|
||||
failure instanceof AnalysisCancelledError &&
|
||||
failure.cause === controller.signal.reason,
|
||||
),
|
||||
).toBe(true);
|
||||
expect(result.error.cleanupIncomplete).toBe(cleanupFails);
|
||||
if (cleanupFails) {
|
||||
await expect(provider.close()).rejects.toMatchObject({
|
||||
cleanupIncomplete: true,
|
||||
});
|
||||
publicationCancellation.failRootCleanup = false;
|
||||
await provider.close();
|
||||
expect(result.error.cleanupIncomplete).toBe(false);
|
||||
expect(resolvedAnalysisError(result.error)).toBe(primary);
|
||||
expect(
|
||||
primary.secondaryFailures.some(
|
||||
(failure) =>
|
||||
failure instanceof AnalysisCancelledError &&
|
||||
failure.cause === controller.signal.reason,
|
||||
),
|
||||
).toBe(true);
|
||||
}
|
||||
} finally {
|
||||
publicationCancellation.failRootCleanup = false;
|
||||
publicationCancellation.cleanupController = undefined;
|
||||
await provider.close();
|
||||
publicationCancellation.cleanupRoot = undefined;
|
||||
publicationCancellation.physicalFailure = undefined;
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it.skipIf(process.platform === "win32").each([false, true])(
|
||||
"preserves committed recovery when cancellation arrives during parent close (first close fails: %s)",
|
||||
async (failFirstClose) => {
|
||||
const root = await createTestTempDirectory("rea-recovery-commit-cancel-");
|
||||
const source = join(root, "source.js");
|
||||
const output = join(root, "output");
|
||||
await writeFile(source, "module.exports = 42;");
|
||||
const controller = new AbortController();
|
||||
publicationCancellation.commitController = controller;
|
||||
publicationCancellation.commitParent = root;
|
||||
publicationCancellation.failPublishedClose = failFirstClose;
|
||||
publicationCancellation.publishedCloseAttempts = 0;
|
||||
const provider = new WakaruProvider({});
|
||||
try {
|
||||
const result = await provider.recover(
|
||||
{
|
||||
path: source,
|
||||
output_directory: output,
|
||||
extraction_mode: "structural",
|
||||
rewrite_level: "standard",
|
||||
},
|
||||
{ signal: controller.signal },
|
||||
);
|
||||
if (!result.ok) throw result.error;
|
||||
expect(controller.signal.aborted).toBe(true);
|
||||
expect(publicationCancellation.publishedCloseAttempts).toBe(
|
||||
failFirstClose ? 2 : 1,
|
||||
);
|
||||
await expect(access(join(output, "module.js"))).resolves.toBeUndefined();
|
||||
} finally {
|
||||
publicationCancellation.commitController = undefined;
|
||||
publicationCancellation.commitParent = undefined;
|
||||
publicationCancellation.failPublishedClose = false;
|
||||
await provider.close();
|
||||
}
|
||||
},
|
||||
);
|
||||
Reference in new issue
Block a user