mirror of
https://github.com/morluto/rea.git
synced 2026-11-10 00:36:35 +00:00
fix(javascript): preserve cancellation during recovery report reads (#9605)
* fix(javascript): preserve recovery report cancellation * fix(javascript): retain recovery output on report failures
This commit is contained in:
1 parent
94728785ea
commit
8358cfcab3
6 files changed
+344
-13
No files matched your search
@@ -174,7 +174,10 @@ export abstract class AnalysisError extends Error {
|
||||
readonly invocationDiagnostics:
|
||||
| Readonly<Record<string, JsonValue>>
|
||||
| undefined;
|
||||
readonly capturedOutput: AnalysisCapturedOutput | undefined;
|
||||
#capturedOutput: AnalysisCapturedOutput | undefined;
|
||||
get capturedOutput(): AnalysisCapturedOutput | undefined {
|
||||
return this.#capturedOutput;
|
||||
}
|
||||
readonly cleanup: AnalysisCleanupObservation | undefined;
|
||||
cleanupIncomplete: boolean;
|
||||
#ownedCleanupResources: readonly string[];
|
||||
@@ -198,7 +201,7 @@ export abstract class AnalysisError extends Error {
|
||||
options?.invocationDiagnostics === undefined
|
||||
? undefined
|
||||
: structuredClone(options.invocationDiagnostics);
|
||||
this.capturedOutput =
|
||||
this.#capturedOutput =
|
||||
options?.capturedOutput === undefined
|
||||
? undefined
|
||||
: { ...options.capturedOutput };
|
||||
@@ -369,6 +372,12 @@ export abstract class AnalysisError extends Error {
|
||||
return this;
|
||||
}
|
||||
|
||||
/** Keep command output observed before this error without replacing the owner. */
|
||||
retainCapturedOutput(output: AnalysisCapturedOutput): this {
|
||||
this.#capturedOutput ??= { ...output };
|
||||
return this;
|
||||
}
|
||||
|
||||
/** Preserve the exact committed prefix when automatic question publication stops. */
|
||||
retainWorkflowUnknownPublication(
|
||||
publication: WorkflowUnknownPublication,
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import type { Dir } from "node:fs";
|
||||
import { renameSync, writeFileSync, type Dir } from "node:fs";
|
||||
import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js";
|
||||
import { createHash } from "node:crypto";
|
||||
import { writeFile, type FileHandle } from "node:fs/promises";
|
||||
@@ -6,6 +6,11 @@ import { dirname, join } from "node:path";
|
||||
import { afterEach, expect, it, onTestFinished, vi } from "vitest";
|
||||
import { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js";
|
||||
import { ArtifactFileCleanupFailure } from "../../artifacts/ArtifactFileOperation.js";
|
||||
import {
|
||||
AnalysisCancelledError,
|
||||
AnalysisArtifactChangedError,
|
||||
} from "../../domain/analysisErrorCore.js";
|
||||
import { parseWakaruReports, WakaruReportReadFailure } from "./WakaruReport.js";
|
||||
import { createTestTempDirectory } from "../../../tests/fixtures/temporaryDirectory.js";
|
||||
import {
|
||||
fingerprintRecoveryFile,
|
||||
@@ -16,6 +21,7 @@ import {
|
||||
const closeFault = vi.hoisted(() => ({
|
||||
path: undefined as string | undefined,
|
||||
denied: false,
|
||||
replaceAfterRead: false,
|
||||
cancel: undefined as AbortController | undefined,
|
||||
handles: [] as FileHandle[],
|
||||
failure: new Error("controlled recovery descriptor close failure"),
|
||||
@@ -56,6 +62,17 @@ vi.mock("node:fs/promises", async (importOriginal) => {
|
||||
open: async (...args: Parameters<typeof actual.open>) => {
|
||||
const handle = await actual.open(...args);
|
||||
if (args[0] === closeFault.path) {
|
||||
if (closeFault.replaceAfterRead) {
|
||||
const createStream = handle.createReadStream.bind(handle);
|
||||
handle.createReadStream = (...options) => {
|
||||
const stream = createStream(...options);
|
||||
stream.once("end", () => {
|
||||
renameSync(String(args[0]), `${String(args[0])}.original`);
|
||||
writeFileSync(String(args[0]), "replacement");
|
||||
});
|
||||
return stream;
|
||||
};
|
||||
}
|
||||
const close = handle.close.bind(handle);
|
||||
closeFault.handles.push(handle);
|
||||
onTestFinished(async () => {
|
||||
@@ -79,10 +96,49 @@ afterEach(() => {
|
||||
directoryFault.closeAttempts = 0;
|
||||
closeFault.path = undefined;
|
||||
closeFault.denied = false;
|
||||
closeFault.replaceAfterRead = false;
|
||||
closeFault.cancel = undefined;
|
||||
closeFault.handles = [];
|
||||
});
|
||||
|
||||
it("retains command output when the provenance file changes during its read", async () => {
|
||||
const { path } = await selectedFile();
|
||||
closeFault.replaceAfterRead = true;
|
||||
const stdout = JSON.stringify({
|
||||
detected_formats: [],
|
||||
safety: "safe",
|
||||
modules: [],
|
||||
warnings: [],
|
||||
total: 0,
|
||||
failed: 0,
|
||||
elapsed_ms: 1,
|
||||
});
|
||||
const failure: unknown = await parseWakaruReports({
|
||||
execution: {
|
||||
exit_code: 0,
|
||||
signal: null,
|
||||
stdout,
|
||||
stderr: "engine diagnostic",
|
||||
command: "/wakaru",
|
||||
arguments: [],
|
||||
resource_limiter: "/prlimit",
|
||||
},
|
||||
provenancePath: path,
|
||||
snapshots: new Map(),
|
||||
files: [],
|
||||
}).catch((cause: unknown) => cause);
|
||||
expect(failure).toBeInstanceOf(AnalysisArtifactChangedError);
|
||||
if (!(failure instanceof AnalysisArtifactChangedError))
|
||||
throw new Error("Expected changed artifact");
|
||||
expect(failure.path).toBe(path);
|
||||
expect(failure.capturedOutput).toMatchObject({
|
||||
stdout,
|
||||
stderr: "engine diagnostic",
|
||||
exit_code: 0,
|
||||
signal: null,
|
||||
});
|
||||
});
|
||||
|
||||
const selectedFile = async () => {
|
||||
const root = await createTestTempDirectory("rea-recovery-file-owner-");
|
||||
const path = join(root, "bundle.js");
|
||||
@@ -92,6 +148,102 @@ const selectedFile = async () => {
|
||||
return { path, bytes };
|
||||
};
|
||||
|
||||
it("retains report output and observed file hash when cancellation arrives during provenance close", async () => {
|
||||
const root = await createTestTempDirectory("rea-recovery-provenance-cancel-");
|
||||
const path = join(root, "provenance.json");
|
||||
const bytes = Buffer.from(
|
||||
'{"format":"unknown","strategy":"safe","modules":{}}',
|
||||
);
|
||||
await writeFile(path, bytes);
|
||||
closeFault.path = path;
|
||||
const controller = new AbortController();
|
||||
closeFault.cancel = controller;
|
||||
const stdout = JSON.stringify({
|
||||
detected_formats: [],
|
||||
safety: "safe",
|
||||
modules: [],
|
||||
warnings: [],
|
||||
total: 0,
|
||||
failed: 0,
|
||||
elapsed_ms: 1,
|
||||
});
|
||||
const failure: unknown = await parseWakaruReports({
|
||||
execution: {
|
||||
exit_code: 0,
|
||||
signal: null,
|
||||
stdout,
|
||||
stderr: "",
|
||||
command: "/wakaru",
|
||||
arguments: [],
|
||||
resource_limiter: "/prlimit",
|
||||
},
|
||||
provenancePath: path,
|
||||
snapshots: new Map(),
|
||||
files: [{ relativePath: "provenance.json" }],
|
||||
signal: controller.signal,
|
||||
}).catch((cause: unknown) => cause);
|
||||
expect(failure).toBeInstanceOf(AnalysisCancelledError);
|
||||
if (!(failure instanceof AnalysisCancelledError))
|
||||
throw new Error("Expected cancellation");
|
||||
expect(failure.cause).toBe(controller.signal.reason);
|
||||
expect(failure.partialObservation).toMatchObject({
|
||||
kind: "file-snapshot-hash",
|
||||
path,
|
||||
bytes: bytes.length,
|
||||
});
|
||||
expect(failure.capturedOutput).toMatchObject({
|
||||
stdout,
|
||||
stderr: "",
|
||||
exit_code: 0,
|
||||
});
|
||||
});
|
||||
|
||||
it("retains command output and the exact provenance cleanup owner", async () => {
|
||||
const { path } = await selectedFile();
|
||||
closeFault.denied = true;
|
||||
const resources = new ArtifactResourceScope();
|
||||
const stdout = JSON.stringify({
|
||||
detected_formats: [],
|
||||
safety: "safe",
|
||||
modules: [],
|
||||
warnings: [],
|
||||
total: 0,
|
||||
failed: 0,
|
||||
elapsed_ms: 1,
|
||||
});
|
||||
const failure: unknown = await parseWakaruReports({
|
||||
execution: {
|
||||
exit_code: 0,
|
||||
signal: null,
|
||||
stdout,
|
||||
stderr: "engine diagnostic",
|
||||
command: "/wakaru",
|
||||
arguments: [],
|
||||
resource_limiter: "/prlimit",
|
||||
},
|
||||
provenancePath: path,
|
||||
snapshots: new Map(),
|
||||
files: [],
|
||||
resources,
|
||||
}).catch((cause: unknown) => cause);
|
||||
expect(failure).toMatchObject({
|
||||
capturedOutput: {
|
||||
stdout,
|
||||
stderr: "engine diagnostic",
|
||||
exit_code: 0,
|
||||
signal: null,
|
||||
},
|
||||
});
|
||||
expect(failure).toBeInstanceOf(WakaruReportReadFailure);
|
||||
if (!(failure instanceof WakaruReportReadFailure))
|
||||
throw new Error("Expected retained report failure");
|
||||
expect(failure.failure).toBeInstanceOf(ArtifactFileCleanupFailure);
|
||||
expect(resources.hasPending).toBe(true);
|
||||
closeFault.denied = false;
|
||||
await resources.retryCleanup();
|
||||
expect(resources.hasPending).toBe(false);
|
||||
});
|
||||
|
||||
it.each(["completed", "size-limit"] as const)(
|
||||
"retains the exact descriptor and %s read outcome through a failed close",
|
||||
async (mode) => {
|
||||
|
||||
@@ -40,6 +40,7 @@ import {
|
||||
import type { ProviderProcessSupervisor } from "../../process/ProviderProcess.js";
|
||||
import { prepareWakaruExecution } from "./WakaruExecution.js";
|
||||
import { RECOVERY_LIMITS } from "./WakaruRelease.js";
|
||||
import { WakaruReportReadFailure } from "./WakaruReport.js";
|
||||
|
||||
const OPERATION = "recover_javascript_sources";
|
||||
|
||||
@@ -166,14 +167,25 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
|
||||
failed = { cause };
|
||||
}
|
||||
if (failed !== undefined) {
|
||||
const reportFailure =
|
||||
failed.cause instanceof WakaruReportReadFailure
|
||||
? failed.cause
|
||||
: undefined;
|
||||
const primary = reportFailure?.failure ?? failed.cause;
|
||||
const failure =
|
||||
failed.cause instanceof ArtifactReaderFailure &&
|
||||
failed.cause.reason === "cancelled" &&
|
||||
failed.cause.cleanup === undefined
|
||||
primary instanceof ArtifactReaderFailure &&
|
||||
primary.reason === "cancelled" &&
|
||||
primary.cleanup === undefined
|
||||
? new AnalysisCancelledError(OPERATION, {
|
||||
cause: artifactFileCancellationCause(failed.cause),
|
||||
cause: artifactFileCancellationCause(primary),
|
||||
...(primary.partialObservation === undefined
|
||||
? {}
|
||||
: { partialObservation: primary.partialObservation }),
|
||||
...(reportFailure === undefined
|
||||
? {}
|
||||
: { capturedOutput: reportFailure.capturedOutput }),
|
||||
})
|
||||
: failed.cause;
|
||||
: primary;
|
||||
const residuals: string[] = [];
|
||||
const cleanupFailures: { resource: string; reason: string }[] = [];
|
||||
const cleanupOwner =
|
||||
@@ -263,6 +275,9 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
|
||||
{
|
||||
operation: OPERATION,
|
||||
cause: failure,
|
||||
...(reportFailure === undefined
|
||||
? {}
|
||||
: { capturedOutput: reportFailure.capturedOutput }),
|
||||
...(tree?.published && execution !== undefined
|
||||
? {
|
||||
partialObservation: {
|
||||
@@ -272,7 +287,10 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
|
||||
),
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
: failure instanceof ArtifactReaderFailure &&
|
||||
failure.partialObservation !== undefined
|
||||
? { partialObservation: failure.partialObservation }
|
||||
: {}),
|
||||
},
|
||||
);
|
||||
const lease = this.#pendingCleanup?.lease;
|
||||
@@ -290,6 +308,9 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
|
||||
: new ProviderAdapterError("wakaru", OPERATION, {
|
||||
cause: failure,
|
||||
diagnostics: { reason: recoveryFailureMessage(failure) },
|
||||
...(reportFailure === undefined
|
||||
? {}
|
||||
: { capturedOutput: reportFailure.capturedOutput }),
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1,10 +1,15 @@
|
||||
import { readFile, truncate, writeFile } from "node:fs/promises";
|
||||
import { chmod, readFile, truncate, writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { expect, it } from "vitest";
|
||||
import { projectAnalysisError } from "../../domain/analysisErrorProjection.js";
|
||||
import { ProviderAdapterError } from "../../domain/providerAdapterError.js";
|
||||
import { createTestTempDirectory } from "../../../tests/fixtures/temporaryDirectory.js";
|
||||
import { AnalysisOutputError } from "../../domain/analysisErrorCore.js";
|
||||
import {
|
||||
AnalysisAccessDeniedError,
|
||||
AnalysisCancelledError,
|
||||
AnalysisOutputError,
|
||||
} from "../../domain/analysisErrorCore.js";
|
||||
import { requirePermissionDenial } from "../../../tests/fixtures/permissionDenial.js";
|
||||
import {
|
||||
parseWakaruReports,
|
||||
validateRecoverySourceMap,
|
||||
@@ -31,6 +36,60 @@ const execution = {
|
||||
resource_limiter: "/usr/bin/prlimit",
|
||||
};
|
||||
|
||||
it.skipIf(process.platform === "win32")(
|
||||
"retains command output when the provenance file cannot be read",
|
||||
async (context) => {
|
||||
await requirePermissionDenial(context, "file-read");
|
||||
const root = await createTestTempDirectory("rea-wakaru-provenance-denied-");
|
||||
const path = join(root, "provenance.json");
|
||||
await writeFile(path, "{}");
|
||||
await chmod(path, 0);
|
||||
const reported = { ...execution, stderr: "engine diagnostic" };
|
||||
try {
|
||||
const failure: unknown = await parseWakaruReports({
|
||||
execution: reported,
|
||||
provenancePath: path,
|
||||
snapshots: new Map(),
|
||||
files: [],
|
||||
}).catch((cause: unknown) => cause);
|
||||
expect(failure).toBeInstanceOf(AnalysisAccessDeniedError);
|
||||
if (!(failure instanceof AnalysisAccessDeniedError))
|
||||
throw new Error("Expected access denial");
|
||||
expect(failure.path).toBe(path);
|
||||
expect(failure.capturedOutput).toMatchObject({
|
||||
stdout: reported.stdout,
|
||||
stderr: reported.stderr,
|
||||
exit_code: 0,
|
||||
signal: null,
|
||||
});
|
||||
} finally {
|
||||
await chmod(path, 0o600);
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it("preserves cancellation and its exact reason when the provenance read is cancelled", async () => {
|
||||
const controller = new AbortController();
|
||||
const reason = { requested: "stop provenance read" };
|
||||
controller.abort(reason);
|
||||
const failure: unknown = await parseWakaruReports({
|
||||
execution,
|
||||
provenancePath: "/unread-provenance.json",
|
||||
snapshots: new Map(),
|
||||
files: [],
|
||||
signal: controller.signal,
|
||||
}).catch((cause: unknown) => cause);
|
||||
expect(failure).toBeInstanceOf(AnalysisCancelledError);
|
||||
if (!(failure instanceof AnalysisCancelledError))
|
||||
throw new Error("Expected cancellation");
|
||||
expect(failure.cause).toBe(reason);
|
||||
expect(failure.capturedOutput).toMatchObject({
|
||||
stdout: reportStdout,
|
||||
stderr: "",
|
||||
exit_code: 0,
|
||||
});
|
||||
});
|
||||
|
||||
it("names the fixed report allowance when provenance.json exceeds 8 MiB", async () => {
|
||||
const root = await createTestTempDirectory("rea-wakaru-provenance-size-");
|
||||
const provenancePath = join(root, "provenance.json");
|
||||
|
||||
@@ -1,12 +1,18 @@
|
||||
import { z } from "zod";
|
||||
import { safeParseUtf8Json } from "../../domain/safeJson.js";
|
||||
import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js";
|
||||
import type { AnalysisCapturedOutput } from "../../domain/analysisErrorBase.js";
|
||||
import type { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js";
|
||||
import {
|
||||
normalizeArtifactPath,
|
||||
ArtifactPathRegistry,
|
||||
} from "../../artifacts/ArtifactPaths.js";
|
||||
import { AnalysisOutputError } from "../../domain/analysisErrorCore.js";
|
||||
import {
|
||||
AnalysisAccessDeniedError,
|
||||
AnalysisArtifactChangedError,
|
||||
AnalysisCancelledError,
|
||||
AnalysisOutputError,
|
||||
} from "../../domain/analysisErrorCore.js";
|
||||
import { ProviderAdapterError } from "../../domain/providerAdapterError.js";
|
||||
import type { JavaScriptRecoveryResult } from "../../domain/javascript/javascriptRecovery.js";
|
||||
import {
|
||||
@@ -55,6 +61,16 @@ const provenanceSchema = z.looseObject({
|
||||
|
||||
const OPERATION = "recover_javascript_sources";
|
||||
|
||||
/** Keep the reader's exact cleanup owner alongside the finished command output. */
|
||||
export class WakaruReportReadFailure extends Error {
|
||||
constructor(
|
||||
readonly failure: ArtifactReaderFailure,
|
||||
readonly capturedOutput: AnalysisCapturedOutput,
|
||||
) {
|
||||
super(failure.message, { cause: failure });
|
||||
}
|
||||
}
|
||||
|
||||
type ReportContext = {
|
||||
execution: Awaited<ReturnType<typeof runWakaruCommand>>;
|
||||
provenancePath: string;
|
||||
@@ -106,11 +122,24 @@ export const parseWakaruReports = async (context: ReportContext) => {
|
||||
: ("complete" as const),
|
||||
};
|
||||
} catch (cause: unknown) {
|
||||
const capturedOutput = {
|
||||
stdout: execution.stdout,
|
||||
stderr: execution.stderr,
|
||||
exit_code: execution.exit_code,
|
||||
signal: execution.signal,
|
||||
truncated: false,
|
||||
};
|
||||
if (
|
||||
cause instanceof AnalysisCancelledError ||
|
||||
cause instanceof AnalysisAccessDeniedError ||
|
||||
cause instanceof AnalysisArtifactChangedError
|
||||
)
|
||||
throw cause.retainCapturedOutput(capturedOutput);
|
||||
if (
|
||||
cause instanceof ArtifactReaderFailure &&
|
||||
(cause.cleanup !== undefined || cause.reason === "cancelled")
|
||||
)
|
||||
throw cause;
|
||||
throw new WakaruReportReadFailure(cause, capturedOutput);
|
||||
throw new ProviderAdapterError("wakaru", OPERATION, {
|
||||
cause,
|
||||
...wakaruReportGuidance(cause),
|
||||
|
||||
@@ -8,6 +8,8 @@ import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
|
||||
const publicationCancellation = vi.hoisted(() => ({
|
||||
controller: undefined as AbortController | undefined,
|
||||
directorySyncs: 0,
|
||||
reportCleanup: false,
|
||||
allowCleanup: false,
|
||||
}));
|
||||
|
||||
vi.mock("node:fs/promises", async (importOriginal) => {
|
||||
@@ -72,6 +74,32 @@ vi.mock(
|
||||
prepareWakaruExecution: async (
|
||||
context: Parameters<typeof actual.prepareWakaruExecution>[0],
|
||||
) => {
|
||||
if (publicationCancellation.reportCleanup) {
|
||||
const cleanup = await context.resources.release({
|
||||
kind: "lease",
|
||||
resource: "provenance descriptor",
|
||||
lease: {
|
||||
release: async () => {
|
||||
if (!publicationCancellation.allowCleanup)
|
||||
throw new Error("close denied");
|
||||
},
|
||||
},
|
||||
});
|
||||
if (cleanup.kind !== "failed")
|
||||
throw new Error("Expected cleanup failure");
|
||||
const { WakaruReportReadFailure } =
|
||||
await import("../../../src/javascript/recovery/WakaruReport.js");
|
||||
throw new WakaruReportReadFailure(
|
||||
context.resources.cleanupFailure({ kind: "completed" }, [cleanup]),
|
||||
{
|
||||
stdout: "report output",
|
||||
stderr: "engine diagnostic",
|
||||
exit_code: 1,
|
||||
signal: null,
|
||||
truncated: false,
|
||||
},
|
||||
);
|
||||
}
|
||||
const source = context.sources[0];
|
||||
if (source === undefined) throw new Error("Missing recovery source");
|
||||
const bytes = Buffer.from("module.exports = 42;");
|
||||
@@ -99,6 +127,39 @@ import { AnalysisCancelledError } from "../../../src/domain/analysisErrorCore.js
|
||||
import { projectAnalysisError } from "../../../src/domain/analysisErrorProjection.js";
|
||||
import { WakaruProvider } from "../../../src/javascript/recovery/WakaruProvider.js";
|
||||
|
||||
it("retains report command output and cleanup ownership through provider retry", async () => {
|
||||
const root = await createTestTempDirectory("rea-wakaru-report-cleanup-");
|
||||
const sourcePath = join(root, "source.js");
|
||||
await writeFile(sourcePath, "module.exports = 42;");
|
||||
publicationCancellation.reportCleanup = true;
|
||||
publicationCancellation.allowCleanup = false;
|
||||
const provider = new WakaruProvider({});
|
||||
try {
|
||||
const result = await provider.recover({
|
||||
path: sourcePath,
|
||||
output_directory: join(root, "output"),
|
||||
extraction_mode: "structural",
|
||||
rewrite_level: "standard",
|
||||
});
|
||||
if (result.ok) throw new Error("Expected cleanup failure");
|
||||
expect(result.error.cleanupIncomplete).toBe(true);
|
||||
expect(result.error.capturedOutput).toMatchObject({
|
||||
stdout: "report output",
|
||||
stderr: "engine diagnostic",
|
||||
exit_code: 1,
|
||||
signal: null,
|
||||
});
|
||||
publicationCancellation.allowCleanup = true;
|
||||
await provider.close();
|
||||
expect(result.error.cleanupIncomplete).toBe(false);
|
||||
expect(result.error.capturedOutput?.stdout).toBe("report output");
|
||||
} finally {
|
||||
publicationCancellation.allowCleanup = true;
|
||||
publicationCancellation.reportCleanup = false;
|
||||
await provider.close();
|
||||
}
|
||||
});
|
||||
|
||||
it.skipIf(process.platform === "win32")(
|
||||
"keeps mid-commit output-tree cancellation typed through the provider",
|
||||
async () => {
|
||||
|
||||
Reference in new issue
Block a user