diff --git a/src/javascript/recovery/RecoveryFiles.ownership.test.ts b/src/javascript/recovery/RecoveryFiles.ownership.test.ts new file mode 100644 index 0000000000..80b578ce8c --- /dev/null +++ b/src/javascript/recovery/RecoveryFiles.ownership.test.ts @@ -0,0 +1,167 @@ +import type { Dir } from "node:fs"; +import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js"; +import { createHash } from "node:crypto"; +import { writeFile, type FileHandle } from "node:fs/promises"; +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 { createTestTempDirectory } from "../../../tests/fixtures/temporaryDirectory.js"; +import { inventoryRecoveryFiles, readRecoveryFile } from "./RecoveryFiles.js"; + +const closeFault = vi.hoisted(() => ({ + path: undefined as string | undefined, + denied: false, + handles: [] as FileHandle[], + failure: new Error("controlled recovery descriptor close failure"), +})); + +const directoryFault = vi.hoisted(() => ({ + path: undefined as string | undefined, + readFailure: undefined as Error | undefined, + cancel: undefined as AbortController | undefined, + closeAttempts: 0, +})); + +vi.mock("node:fs/promises", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + opendir: async (...args: Parameters) => { + const handle: Dir = await actual.opendir(...args); + if (args[0] === directoryFault.path) { + const close = handle.close.bind(handle); + onTestFinished(async () => { + await close(); + }); + const read = handle.read.bind(handle); + handle.read = async () => { + if (directoryFault.readFailure !== undefined) + throw directoryFault.readFailure; + return read(); + }; + handle.close = async () => { + directoryFault.closeAttempts++; + directoryFault.cancel?.abort(); + throw new Error("controlled recovery directory close failure"); + }; + } + return handle; + }, + open: async (...args: Parameters) => { + const handle = await actual.open(...args); + if (args[0] === closeFault.path) { + const close = handle.close.bind(handle); + closeFault.handles.push(handle); + onTestFinished(async () => { + if (handle.fd >= 0) await close(); + }); + handle.close = async () => { + if (closeFault.denied) throw closeFault.failure; + await close(); + }; + } + return handle; + }, + }; +}); + +afterEach(() => { + directoryFault.path = undefined; + directoryFault.readFailure = undefined; + directoryFault.cancel = undefined; + directoryFault.closeAttempts = 0; + closeFault.path = undefined; + closeFault.denied = false; + closeFault.handles = []; +}); + +const selectedFile = async () => { + const root = await createTestTempDirectory("rea-recovery-file-owner-"); + const path = join(root, "bundle.js"); + const bytes = Buffer.from("module.exports = 42;\n"); + await writeFile(path, bytes); + closeFault.path = path; + return { path, bytes }; +}; + +it.each(["completed", "size-limit"] as const)( + "retains the exact descriptor and %s read outcome through a failed close", + async (mode) => { + const { path, bytes } = await selectedFile(); + const resources = new ArtifactResourceScope(); + closeFault.denied = true; + const failure = await readRecoveryFile( + path, + mode === "completed" ? 1024 : 0, + undefined, + resources, + ).catch((cause: unknown) => cause); + expect(failure).toBeInstanceOf(ArtifactFileCleanupFailure); + if (!(failure instanceof ArtifactFileCleanupFailure)) return; + const physical = failure.fileCleanupFailure; + expect(physical.owner.handle).toBe(closeFault.handles[0]); + expect(physical.cleanupCause).toBe(closeFault.failure); + if (mode === "completed") + expect(physical.outcome).toEqual({ + kind: "completed", + value: { + sha256: createHash("sha256").update(bytes).digest("hex"), + bytes: bytes.length, + }, + }); + else + expect(physical.outcome).toMatchObject({ + kind: "failed", + cause: { maximum: 0 }, + }); + expect(resources.hasPending).toBe(true); + await expect(resources.retryCleanup()).rejects.toBeDefined(); + expect(closeFault.handles[0]?.fd).toBeGreaterThanOrEqual(0); + closeFault.denied = false; + await resources.retryCleanup(); + expect(resources.hasPending).toBe(false); + expect(closeFault.handles[0]?.fd).toBe(-1); + }, +); + +it.each([ + [false, false], + [true, false], + [false, true], + [true, true], +] as const)( + "retains an uncertain directory close and its primary read outcome (read denied: %s, cancel on close: %s)", + async (readDenied, cancelOnClose) => { + const { path } = await selectedFile(); + const root = dirname(path); + directoryFault.path = root; + const readFailure = new Error("controlled directory read failure"); + if (readDenied) directoryFault.readFailure = readFailure; + const resources = new ArtifactResourceScope(); + const controller = new AbortController(); + if (cancelOnClose) directoryFault.cancel = controller; + const failure = await inventoryRecoveryFiles( + root, + controller.signal, + resources, + ).catch((cause: unknown) => cause); + expect(failure).toBeInstanceOf(ArtifactReaderFailure); + if (!(failure instanceof ArtifactReaderFailure)) return; + expect(failure.cleanup?.resources).toContain(root); + if (readDenied) { + expect(failure.cause).toBeInstanceOf(ArtifactReaderFailure); + expect((failure.cause as ArtifactReaderFailure).cause).toBe(readFailure); + } + if (cancelOnClose && !readDenied) { + expect(failure.reason).toBe("cancelled"); + expect(failure.cancellation?.reason).toBe(controller.signal.reason); + } + expect(resources.hasPending).toBe(true); + expect(directoryFault.closeAttempts).toBe(1); + await expect(resources.retryCleanup()).rejects.toBeDefined(); + // Node has no public directory validity signal, so an unconfirmed close cannot be retried as if it were open. + expect(directoryFault.closeAttempts).toBe(1); + expect(resources.hasPending).toBe(true); + }, +); diff --git a/src/javascript/recovery/RecoveryFiles.test.ts b/src/javascript/recovery/RecoveryFiles.test.ts index 10c2c5d689..b878dd192d 100644 --- a/src/javascript/recovery/RecoveryFiles.test.ts +++ b/src/javascript/recovery/RecoveryFiles.test.ts @@ -7,6 +7,7 @@ import { AnalysisInputError, AnalysisOutputError, } from "../../domain/analysisErrorCore.js"; +import { ArtifactFileCleanupFailure } from "../../artifacts/ArtifactFileOperation.js"; import { OwnedFileHandle } from "../../filesystem/OwnedFileHandle.js"; import { openRegularFile, @@ -92,12 +93,24 @@ it("keeps a failed admission close owned when cancellation is also requested", a new Error("close failed"), owner, ); - vi.mocked(openRegularFile).mockRejectedValueOnce(failure); const controller = new AbortController(); - controller.abort(); - await expect( - readRecoveryFile("/selected.js", 1024, controller.signal), - ).rejects.toBe(failure); + vi.mocked(openRegularFile).mockImplementationOnce(async () => { + controller.abort(); + throw failure; + }); + const observed = await readRecoveryFile( + "/selected.js", + 1024, + controller.signal, + ).catch((cause: unknown) => cause); + expect(observed).toBeInstanceOf(ArtifactFileCleanupFailure); + if (!(observed instanceof ArtifactFileCleanupFailure)) return; + expect(observed.fileCleanupFailure.owner).toBe(owner); + expect(observed.fileCleanupFailure.outcome).toEqual({ + kind: "failed", + cause: failure.cause, + }); + expect(observed.fileCleanupFailure.cleanupCause).toBe(failure.cleanupCause); }); it("reports one file selected under two letter cases as an input error", async (context) => { diff --git a/src/javascript/recovery/RecoveryFiles.ts b/src/javascript/recovery/RecoveryFiles.ts index 28571a7345..1da3e3e7de 100644 --- a/src/javascript/recovery/RecoveryFiles.ts +++ b/src/javascript/recovery/RecoveryFiles.ts @@ -1,4 +1,12 @@ -import { regularFileCancellation } from "../../filesystem/OwnedRegularFile.js"; +import { + artifactFileCancellationCause, + ArtifactFileCancelled, + ArtifactFileCleanupFailure, + withArtifactRegularFile, +} from "../../artifacts/ArtifactFileOperation.js"; +import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js"; +import { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js"; +import { OwnedDirectoryHandle } from "../../filesystem/OwnedDirectoryHandle.js"; import { createHash } from "node:crypto"; import { lstat, mkdir, opendir, writeFile } from "node:fs/promises"; import { dirname, isAbsolute, join, relative, resolve, sep } from "node:path"; @@ -13,8 +21,7 @@ import { } from "../../domain/analysisErrorCore.js"; import { NonRegularFileReadError, - openRegularFile, - RegularFileAdmissionFailure, + sameRegularFileState, } from "../../filesystem/RegularFile.js"; import { RECOVERY_LIMITS, wakaruFixedLimitRecovery } from "./WakaruRelease.js"; @@ -48,19 +55,56 @@ const scanRecoveryFile = async ( maximum: number, consume: (chunk: Buffer) => void, signal?: AbortSignal, + resources?: ArtifactResourceScope, ) => { - // A FIFO without a writer must be rejected before it can block the open. - const handle = await openRegularFile(path, { - symlinks: "reject", - signal, - }).catch((cause: unknown) => { - // A descriptor whose close failed stays owned by its error, even when - // cancellation was also requested. - if (cause instanceof RegularFileAdmissionFailure) throw cause; - const cancellation = regularFileCancellation(cause); - if (cancellation !== undefined) + try { + return await withArtifactRegularFile( + path, + resources ?? new ArtifactResourceScope(), + signal, + async (handle) => { + const hash = createHash("sha256"); + const before = await handle.stat(); + if (before.size > maximum) + throw new RecoveryFileSizeError( + `Expected a regular file no larger than ${String(maximum)} bytes: ${path}`, + maximum, + ); + let bytes = 0; + for await (const raw of handle.createReadStream({ autoClose: false })) { + checkRecoveryCancellation(signal); + const chunk = Buffer.isBuffer(raw) ? raw : Buffer.from(raw); + bytes += chunk.length; + if (bytes > maximum) + throw new RecoveryFileSizeError( + `File exceeds ${String(maximum)} bytes: ${path}`, + maximum, + ); + hash.update(chunk); + consume(chunk); + } + const after = await handle.stat(); + const current = await lstat(path); + if ( + before.size !== bytes || + [after, current].some((state) => !sameRegularFileState(before, state)) + ) + throw new AnalysisOutputError( + OPERATION, + `File identity changed while reading: ${path}`, + ); + return { sha256: hash.digest("hex"), bytes }; + }, + ); + } catch (cause: unknown) { + if ( + cause instanceof ArtifactFileCleanupFailure || + cause instanceof ArtifactFileCancelled + ) + throw cause; + if (cause instanceof ArtifactReaderFailure && cause.reason === "cancelled") throw new AnalysisCancelledError(OPERATION, { - cause: cancellation.reason, + cause: artifactFileCancellationCause(cause), }); if (cause instanceof NonRegularFileReadError) throw new AnalysisOutputError( @@ -68,45 +112,6 @@ const scanRecoveryFile = async ( `Expected a regular file no larger than ${String(maximum)} bytes: ${path}`, ); throw cause; - }); - const hash = createHash("sha256"); - try { - const before = await handle.stat(); - if (before.size > maximum) - throw new RecoveryFileSizeError( - `Expected a regular file no larger than ${String(maximum)} bytes: ${path}`, - maximum, - ); - let bytes = 0; - for await (const raw of handle.createReadStream({ autoClose: false })) { - checkRecoveryCancellation(signal); - const chunk = Buffer.isBuffer(raw) ? raw : Buffer.from(raw); - bytes += chunk.length; - if (bytes > maximum) - throw new RecoveryFileSizeError( - `File exceeds ${String(maximum)} bytes: ${path}`, - maximum, - ); - hash.update(chunk); - consume(chunk); - } - const after = await handle.stat(); - const current = await lstat(path); - if ( - before.dev !== current.dev || - before.ino !== current.ino || - before.size !== bytes || - before.mtimeMs !== after.mtimeMs || - before.ctimeMs !== after.ctimeMs || - before.size !== after.size - ) - throw new AnalysisOutputError( - OPERATION, - `File identity changed while reading: ${path}`, - ); - return { sha256: hash.digest("hex"), bytes }; - } finally { - await handle.close(); } }; @@ -115,13 +120,15 @@ export const fingerprintRecoveryFile = async ( path: string, maximum: number, signal?: AbortSignal, -) => scanRecoveryFile(path, maximum, () => {}, signal); + resources?: ArtifactResourceScope, +) => scanRecoveryFile(path, maximum, () => {}, signal, resources); /** Read only a regular file through one handle with a bounded, stable identity. */ export const readRecoveryFile = async ( path: string, maximum: number, signal?: AbortSignal, + resources?: ArtifactResourceScope, ): Promise => { const chunks: Buffer[] = []; const result = await scanRecoveryFile( @@ -129,6 +136,7 @@ export const readRecoveryFile = async ( maximum, (chunk) => chunks.push(chunk), signal, + resources, ); return Buffer.concat(chunks, result.bytes); }; @@ -148,6 +156,7 @@ const snapshotRecoveryInput = async ( relativePath: string, maximum: number, signal?: AbortSignal, + resources?: ArtifactResourceScope, ): Promise => { if (!isAbsolute(path)) throw recoveryInputFailure( @@ -156,9 +165,14 @@ const snapshotRecoveryInput = async ( ); let bytes: Buffer; try { - bytes = await readRecoveryFile(path, maximum, signal); + bytes = await readRecoveryFile(path, maximum, signal, resources); } catch (cause: unknown) { - if (cause instanceof AnalysisCancelledError) throw cause; + if ( + cause instanceof AnalysisCancelledError || + cause instanceof ArtifactFileCleanupFailure || + cause instanceof ArtifactFileCancelled + ) + throw cause; // The operation-wide input allowance is a deterministic byte budget, not // a malformed argument; the same selection fails again unchanged. if (cause instanceof RecoveryFileSizeError) @@ -227,6 +241,7 @@ export const snapshotRecoveryInputs = async ( paths: readonly [string, ...string[]], directory: string, signal?: AbortSignal, + resources?: ArtifactResourceScope, ): Promise => { let commonDirectory = dirname(resolve(paths[0])); for (const path of paths) { @@ -277,6 +292,7 @@ export const snapshotRecoveryInputs = async ( relativePath, remaining, signal, + resources, ); snapshots.push(snapshot); remaining -= snapshot.bytes; @@ -292,6 +308,7 @@ export const snapshotRecoveryInputs = async ( export const inventoryRecoveryFiles = async ( root: string, signal?: AbortSignal, + resources: ArtifactResourceScope = new ArtifactResourceScope(), ): Promise< readonly { relativePath: string; path: string; bytes: number }[] > => { @@ -306,9 +323,15 @@ export const inventoryRecoveryFiles = async ( `Unsafe staging directory: ${directory}`, ); const handle = await opendir(directory); + const owner = new OwnedDirectoryHandle(handle); const children: string[] = []; + let outcome: + | { readonly kind: "completed" } + | { readonly kind: "failed"; readonly cause: unknown }; try { - for await (const entry of handle) { + for (;;) { + const entry = await handle.read(); + if (entry === null) break; checkRecoveryCancellation(signal); entries++; if (entries > RECOVERY_LIMITS.outputEntries) @@ -321,9 +344,32 @@ export const inventoryRecoveryFiles = async ( ); children.push(entry.name); } - } finally { - await handle.close().catch(() => undefined); + outcome = { kind: "completed" }; + } catch (cause: unknown) { + outcome = { kind: "failed", cause }; } + const cleanup = await resources.release({ + kind: "directory-handle", + handle: owner, + resource: directory, + }); + if (cleanup.kind === "failed") { + const failure = resources.cleanupFailure(outcome, [cleanup]); + if (outcome.kind === "completed" && signal?.aborted === true) + throw new ArtifactReaderFailure( + "cancelled", + "Recovery inventory cancelled during directory close", + { + cause: failure, + cancellation: { reason: signal.reason }, + cleanup: failure.cleanup, + cleanupLeases: failure.cleanupLeases, + }, + ); + throw failure; + } + if (outcome.kind === "failed") throw outcome.cause; + checkRecoveryCancellation(signal); for (const name of children) { checkRecoveryCancellation(signal); const relativePath = prefix === "" ? name : `${prefix}/${name}`; @@ -369,7 +415,8 @@ export const recoveryFailureMessage = (cause: unknown): string => /** Cancellation remains distinct from malformed source or provider output. */ export const checkRecoveryCancellation = (signal?: AbortSignal): void => { - if (signal?.aborted === true) throw new AnalysisCancelledError(OPERATION); + if (signal?.aborted === true) + throw new AnalysisCancelledError(OPERATION, { cause: signal.reason }); }; /** Keep both tool invocations within one operation's execution budget. */ diff --git a/src/javascript/recovery/WakaruCommand.ts b/src/javascript/recovery/WakaruCommand.ts index 69a050daa5..3227b9fb45 100644 --- a/src/javascript/recovery/WakaruCommand.ts +++ b/src/javascript/recovery/WakaruCommand.ts @@ -7,6 +7,8 @@ import { AnalysisCapabilityUnavailableError, AnalysisOutputError, } from "../../domain/analysisErrorCore.js"; +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 { prlimitArgumentVector } from "../../process/LinuxResourceLimits.js"; @@ -68,6 +70,8 @@ export class WakaruCleanupFailure extends ProviderCleanupError { /** Resolve and fingerprint explicitly configured tools; acquire nothing at startup. */ export const resolveWakaruCommand = async ( environment: Readonly>, + resources?: ArtifactResourceScope, + signal?: AbortSignal, ) => { const configured = environment.REA_WAKARU_COMMAND; if ( @@ -87,6 +91,8 @@ export const resolveWakaruCommand = async ( const fingerprint = await fingerprintRecoveryFile( command, RECOVERY_LIMITS.outputBytes, + signal, + resources, ); const limiter = environment.REA_JAVASCRIPT_PRLIMIT_COMMAND ?? "/usr/bin/prlimit"; @@ -97,6 +103,11 @@ export const resolveWakaruCommand = async ( throw new TypeError(`Resource limiter is not a regular file: ${limiter}`); return { command, limiter, ...fingerprint }; } catch (cause: unknown) { + if ( + cause instanceof ArtifactReaderFailure && + (cause.cleanup !== undefined || cause.reason === "cancelled") + ) + throw cause; const reason = `Configured Wakaru or util-linux prlimit is unavailable: ${recoveryFailureMessage(cause)}`; throw new AnalysisCapabilityUnavailableError("wakaru", OPERATION, reason, { cause, @@ -115,6 +126,7 @@ type WakaruCommandContext = { signal?: AbortSignal; launcher?: WakaruLauncher; staging?: string; + resources?: ArtifactResourceScope; }; /** Run one bounded command and verify cleanup before consuming its output. */ @@ -183,7 +195,11 @@ export const runWakaruCommand = async (context: WakaruCommandContext) => { ), ); if (context.staging !== undefined) - await inventoryRecoveryFiles(context.staging, context.signal); + await inventoryRecoveryFiles( + context.staging, + context.signal, + context.resources, + ); } checkRecoveryDeadline(context.deadline, context.signal); if (oversized) diff --git a/src/javascript/recovery/WakaruExecution.ts b/src/javascript/recovery/WakaruExecution.ts index 7ec93ded50..5da9f86359 100644 --- a/src/javascript/recovery/WakaruExecution.ts +++ b/src/javascript/recovery/WakaruExecution.ts @@ -4,6 +4,7 @@ import { type ExecutionOptions, } from "../../application/AnalysisProvider.js"; import type { SafeOutputTree } from "../../artifacts/SafeOutputTree.js"; +import type { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js"; import { AnalysisCapabilityUnavailableError, AnalysisOutputError, @@ -43,12 +44,21 @@ type RecoveryWorkspace = { launcher: WakaruLauncher | undefined; options: ExecutionOptions | undefined; deadline: number; + resources: ArtifactResourceScope; }; /** Consume the pinned producer representation in a private workspace. */ const executeWakaruRecovery = async (context: RecoveryWorkspace) => { - const { root, input, sources, engine, environment, launcher, options } = - context; + const { + root, + input, + sources, + engine, + environment, + launcher, + options, + resources, + } = context; const staging = join(root, "modules"); const commandContext = { ...engine, @@ -57,6 +67,7 @@ const executeWakaruRecovery = async (context: RecoveryWorkspace) => { environment: environment, ...(options?.signal === undefined ? {} : { signal: options.signal }), ...(launcher === undefined ? {} : { launcher: launcher }), + resources, }; const version = await runWakaruCommand({ ...commandContext, @@ -90,7 +101,11 @@ const executeWakaruRecovery = async (context: RecoveryWorkspace) => { staging, ], }); - const files = await inventoryRecoveryFiles(staging, options?.signal); + const files = await inventoryRecoveryFiles( + staging, + options?.signal, + resources, + ); const snapshots = new Map(); for (const source of sources) snapshots.set( @@ -99,6 +114,7 @@ const executeWakaruRecovery = async (context: RecoveryWorkspace) => { source.snapshot_path, source.bytes, options?.signal, + resources, ), ); const parsed = await parseWakaruReports({ @@ -106,13 +122,15 @@ const executeWakaruRecovery = async (context: RecoveryWorkspace) => { provenancePath: join(staging, "provenance.json"), snapshots, files, + ...(options?.signal === undefined ? {} : { signal: options.signal }), + resources, }); return { version, admitted, run, parsed }; }; /** Publish validated recovery and retain exact source and engine bindings. */ export const prepareWakaruExecution = async (context: RecoveryWorkspace) => { - const { root, tree, input, sources, engine, options } = context; + const { root, tree, input, sources, engine, options, resources } = context; const [source] = sources; const staging = join(root, "modules"); const { version, admitted, run, parsed } = @@ -127,12 +145,14 @@ export const prepareWakaruExecution = async (context: RecoveryWorkspace) => { parsed, execution: run, ...(options?.signal === undefined ? {} : { signal: options.signal }), + resources, }); for (const selected of sources) { const current = await fingerprintRecoveryFile( selected.path, RECOVERY_LIMITS.inputBytes, options?.signal, + resources, ); if (current.sha256 !== selected.sha256 || current.bytes !== selected.bytes) throw new AnalysisOutputError( @@ -146,6 +166,7 @@ export const prepareWakaruExecution = async (context: RecoveryWorkspace) => { engine.command, RECOVERY_LIMITS.outputBytes, options?.signal, + resources, ) ).sha256 !== engine.sha256 ) diff --git a/src/javascript/recovery/WakaruProvider.ts b/src/javascript/recovery/WakaruProvider.ts index e258126630..626dd2b1cb 100644 --- a/src/javascript/recovery/WakaruProvider.ts +++ b/src/javascript/recovery/WakaruProvider.ts @@ -6,6 +6,8 @@ import type { AnalysisExecution, ExecutionOptions, } from "../../application/AnalysisProvider.js"; +import { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js"; +import { AnalysisCleanupLease } from "../../artifacts/AnalysisCleanupLease.js"; import { SafeOutputTree } from "../../artifacts/SafeOutputTree.js"; import { SafeOutputTreeCreationFailure } from "../../artifacts/SafeOutputTreeCreationFailure.js"; import { AnalysisError } from "../../domain/analysisErrorBase.js"; @@ -39,16 +41,22 @@ import { RECOVERY_LIMITS } from "./WakaruRelease.js"; const OPERATION = "recover_javascript_sources"; +interface PendingRecoveryCleanup { + root: string | undefined; + process: ProviderProcessSupervisor | undefined; + tree: SafeOutputTree | undefined; +} + /** Serialized recovery adapter retaining failed process, workspace and publication cleanup. */ export class WakaruProvider implements JavaScriptRecoveryPort { + readonly #resources = new ArtifactResourceScope(); #tail: Promise = Promise.resolve(); #closing = false; #closePromise: Promise | undefined; #pendingCleanup: | { - root: string | undefined; - process: ProviderProcessSupervisor | undefined; - tree: SafeOutputTree | undefined; + readonly resources: PendingRecoveryCleanup; + readonly lease: AnalysisCleanupLease; } | undefined; @@ -107,7 +115,11 @@ export class WakaruProvider implements JavaScriptRecoveryPort { const deadline = performance.now() + RECOVERY_LIMITS.timeoutMs; try { checkRecoveryCancellation(options?.signal); - const engine = await resolveWakaruCommand(this.environment); + const engine = await resolveWakaruCommand( + this.environment, + this.#resources, + options?.signal, + ); root = await mkdtemp(join(tmpdir(), "rea-wakaru-")); const inputs = join(root, "inputs"); const staging = join(root, "modules"); @@ -117,6 +129,7 @@ export class WakaruProvider implements JavaScriptRecoveryPort { [input.path, ...(input.additional_paths ?? [])], inputs, options?.signal, + this.#resources, ); try { tree = await SafeOutputTree.create(input.output_directory); @@ -131,6 +144,7 @@ export class WakaruProvider implements JavaScriptRecoveryPort { execution = await prepareWakaruExecution({ root, tree, + resources: this.#resources, input, sources, engine, @@ -150,7 +164,8 @@ export class WakaruProvider implements JavaScriptRecoveryPort { if (failed !== undefined) { const failure = failed.cause instanceof ArtifactReaderFailure && - failed.cause.reason === "cancelled" + failed.cause.reason === "cancelled" && + failed.cause.cleanup === undefined ? new AnalysisCancelledError(OPERATION, { cause: failed.cause }) : failed.cause; const residuals: string[] = []; @@ -174,7 +189,9 @@ export class WakaruProvider implements JavaScriptRecoveryPort { } if (cleanupOwner !== undefined) residuals.push(cleanupOwner.launch.ownership?.runId ?? "wakaru-worker"); - if (root !== undefined && cleanupOwner === undefined) { + const filesPending = this.#resources.hasPending; + if (filesPending) residuals.push(...this.#resources.cleanupResources); + if (root !== undefined && cleanupOwner === undefined && !filesPending) { const failedRoot = root; try { await rm(failedRoot, { recursive: true, force: true }); @@ -190,47 +207,75 @@ export class WakaruProvider implements JavaScriptRecoveryPort { if ( cleanupOwner !== undefined || failedTree !== undefined || - root !== undefined + root !== undefined || + filesPending ) { - this.#pendingCleanup = { + const resources: PendingRecoveryCleanup = { root, process: cleanupOwner, tree: failedTree, }; + const files = this.#resources; + const lease = new AnalysisCleanupLease( + () => this.#releasePending(resources), + { + get cleanupResources() { + return [ + ...files.cleanupResources, + ...(resources.process === undefined + ? [] + : [ + resources.process.launch.ownership?.runId ?? + "wakaru-worker", + ]), + ...(resources.tree === undefined + ? [] + : [resources.tree.outputRoot]), + ...(resources.root === undefined ? [] : [resources.root]), + ]; + }, + }, + ); + this.#pendingCleanup = { resources, lease }; if (root !== undefined) residuals.push(root); root = undefined; } - if (residuals.length > 0) - return err( - new ProviderCleanupError( - "wakaru", - residuals, - { - ...(failure instanceof WakaruCleanupFailure - ? failure.diagnostics - : {}), - previous_error: - failure instanceof WakaruCleanupFailure - ? (failure.diagnostics?.previous_error ?? null) - : recoveryFailureMessage(failure), - cleanup_failures: cleanupFailures, - }, - { - operation: OPERATION, - cause: failure, - ...(tree?.published && execution !== undefined - ? { - partialObservation: { - kind: "javascript-recovery" as const, - result: javascriptRecoveryResultSchema.parse( - execution.result, - ), - }, - } - : {}), - }, - ), + if (residuals.length > 0) { + const error = new ProviderCleanupError( + "wakaru", + residuals, + { + ...(failure instanceof WakaruCleanupFailure + ? failure.diagnostics + : {}), + previous_error: + failure instanceof WakaruCleanupFailure + ? (failure.diagnostics?.previous_error ?? null) + : recoveryFailureMessage(failure), + cleanup_failures: cleanupFailures, + }, + { + operation: OPERATION, + cause: failure, + ...(tree?.published && execution !== undefined + ? { + partialObservation: { + kind: "javascript-recovery" as const, + result: javascriptRecoveryResultSchema.parse( + execution.result, + ), + }, + } + : {}), + }, ); + const lease = this.#pendingCleanup?.lease; + if (lease !== undefined) { + lease.bindCleanupError(error); + error.bindCleanupResourceOwner(lease); + } + return err(error); + } if (tree?.published === true && execution !== undefined) return ok(execution); return err( @@ -266,49 +311,36 @@ export class WakaruProvider implements JavaScriptRecoveryPort { async #retire(): Promise { const pending = this.#pendingCleanup; if (pending === undefined) return undefined; + try { + await pending.lease.release(); + this.#pendingCleanup = undefined; + return undefined; + } catch (cause: unknown) { + const error = new ProviderCleanupError( + "wakaru", + pending.lease.cleanupResources ?? [], + { reason: recoveryFailureMessage(cause) }, + { operation: OPERATION, cause }, + ).bindCleanupResourceOwner(pending.lease); + pending.lease.bindCleanupError(error); + return error; + } + } + + async #releasePending(pending: PendingRecoveryCleanup): Promise { if (pending.process !== undefined) { const stopped = await pending.process.stop(); - if (stopped.status === "incomplete") - return new ProviderCleanupError( - "wakaru", - [ - pending.process.launch.ownership?.runId ?? "wakaru-worker", - ...(pending.root === undefined ? [] : [pending.root]), - ], - { reason: stopped.reason }, - { operation: OPERATION }, - ); + if (stopped.status === "incomplete") throw new Error(stopped.reason); pending.process = undefined; } + await this.#resources.retryCleanup(); if (pending.tree !== undefined) { - const tree = pending.tree; - try { - await tree.rollback(); - pending.tree = undefined; - } catch (cause: unknown) { - return new ProviderCleanupError( - "wakaru", - [tree.outputRoot], - { reason: recoveryFailureMessage(cause) }, - { operation: OPERATION, cause }, - ); - } + await pending.tree.rollback(); + pending.tree = undefined; } if (pending.root !== undefined) { - const root = pending.root; - try { - await rm(root, { recursive: true, force: true }); - pending.root = undefined; - } catch (cause: unknown) { - return new ProviderCleanupError( - "wakaru", - [root], - { reason: recoveryFailureMessage(cause) }, - { operation: OPERATION, cause }, - ); - } + await rm(pending.root, { recursive: true, force: true }); + pending.root = undefined; } - this.#pendingCleanup = undefined; - return undefined; } } diff --git a/src/javascript/recovery/WakaruPublication.ts b/src/javascript/recovery/WakaruPublication.ts index fc17486813..a54a0be3dc 100644 --- a/src/javascript/recovery/WakaruPublication.ts +++ b/src/javascript/recovery/WakaruPublication.ts @@ -1,6 +1,7 @@ import { join } from "node:path"; import { Readable } from "node:stream"; import type { SafeOutputTree } from "../../artifacts/SafeOutputTree.js"; +import type { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js"; import { AnalysisOutputError } from "../../domain/analysisErrorCore.js"; import { javascriptRecoveryResultSchema, @@ -37,6 +38,7 @@ type PublicationContext = { admitted: Exclude; parsed: Awaited>; execution: Awaited>; + resources?: ArtifactResourceScope; signal?: AbortSignal; }; type PublishedFile = JavaScriptRecoveryResult["manifest"]; @@ -46,7 +48,7 @@ type RecoveryWriter = (path: string, bytes: Buffer) => Promise; export const publishWakaruArtifacts = async ( context: PublicationContext, ): Promise => { - const { tree, sources, parsed, staging, signal } = context; + const { tree, sources, parsed, staging, signal, resources } = context; const write = async (path: string, bytes: Buffer) => { const sha256 = recoveryDigest(bytes); const file = await tree.write( @@ -68,6 +70,7 @@ export const publishWakaruArtifacts = async ( source.snapshot_path, source.bytes, signal, + resources, ); if (recoveryDigest(snapshot) !== source.sha256) throw new AnalysisOutputError( @@ -94,6 +97,7 @@ export const publishWakaruArtifacts = async ( join(staging, "provenance.json"), RECOVERY_LIMITS.reportBytes, signal, + resources, ), ); const details = { @@ -161,13 +165,14 @@ const publishModule = async ( module: PublicationContext["parsed"]["report"]["modules"][number], write: RecoveryWriter, ): Promise => { - const { staging, sources, parsed, signal } = context; + const { staging, sources, parsed, signal, resources } = context; const artifact = await write( `modules/${module.filename}`, await readRecoveryFile( join(staging, module.filename), RECOVERY_LIMITS.outputBytes, signal, + resources, ), ); const mapPath = join(staging, `${module.filename}.map`); @@ -177,6 +182,7 @@ const publishModule = async ( mapPath, RECOVERY_LIMITS.outputBytes, signal, + resources, ).catch((cause: unknown) => { if (cause instanceof Error && "code" in cause && cause.code === "ENOENT") return null; diff --git a/src/javascript/recovery/WakaruReport.ts b/src/javascript/recovery/WakaruReport.ts index b9a828e37a..99fc44ff2b 100644 --- a/src/javascript/recovery/WakaruReport.ts +++ b/src/javascript/recovery/WakaruReport.ts @@ -1,4 +1,6 @@ import { z } from "zod"; +import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js"; +import type { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js"; import { normalizeArtifactPath, ArtifactPathRegistry, @@ -57,6 +59,8 @@ type ReportContext = { provenancePath: string; snapshots: ReadonlyMap; files: readonly { relativePath: string }[]; + resources?: ArtifactResourceScope; + signal?: AbortSignal; }; /** @@ -66,7 +70,7 @@ type ReportContext = { * checks below remain exact. */ export const parseWakaruReports = async (context: ReportContext) => { - const { execution, provenancePath } = context; + const { execution, provenancePath, resources, signal } = context; try { if (execution.exit_code !== 0 && execution.exit_code !== 1) @@ -76,7 +80,9 @@ export const parseWakaruReports = async (context: ReportContext) => { const report = reportSchema.parse(JSON.parse(execution.stdout)); const provenance = provenanceSchema.parse( JSON.parse( - (await readWakaruProvenanceReport(provenancePath)).toString("utf8"), + ( + await readWakaruProvenanceReport(provenancePath, signal, resources) + ).toString("utf8"), ), ); if (report.total !== report.modules.length || report.failed > report.total) @@ -101,6 +107,11 @@ export const parseWakaruReports = async (context: ReportContext) => { : ("complete" as const), }; } catch (cause: unknown) { + if ( + cause instanceof ArtifactReaderFailure && + (cause.cleanup !== undefined || cause.reason === "cancelled") + ) + throw cause; throw new ProviderAdapterError("wakaru", OPERATION, { cause, ...wakaruReportGuidance(cause), @@ -118,9 +129,18 @@ export const parseWakaruReports = async (context: ReportContext) => { const PROVENANCE_REPORT_ALLOWANCE = `Wakaru provenance report exceeds the ${String(RECOVERY_LIMITS.reportBytes)}-byte report allowance`; /** Report-size refusals name the fixed allowance; other read failures stay unchanged. */ -const readWakaruProvenanceReport = async (path: string): Promise => { +const readWakaruProvenanceReport = async ( + path: string, + signal: AbortSignal | undefined, + resources: ArtifactResourceScope | undefined, +): Promise => { try { - return await readRecoveryFile(path, RECOVERY_LIMITS.reportBytes); + return await readRecoveryFile( + path, + RECOVERY_LIMITS.reportBytes, + signal, + resources, + ); } catch (cause: unknown) { if (!isRecoveryFileSizeError(cause, RECOVERY_LIMITS.reportBytes)) throw cause; diff --git a/tests/boundary/javascript-recovery/wakaruFileOwnership.test.ts b/tests/boundary/javascript-recovery/wakaruFileOwnership.test.ts new file mode 100644 index 0000000000..7933622f69 --- /dev/null +++ b/tests/boundary/javascript-recovery/wakaruFileOwnership.test.ts @@ -0,0 +1,206 @@ +import { createHash } from "node:crypto"; +import { access, readFile, type FileHandle } from "node:fs/promises"; +import { basename, join } from "node:path"; +import { afterEach, expect, it, onTestFinished, vi } from "vitest"; +import { + ArtifactFileCancelled, + ArtifactFileCleanupFailure, +} from "../../../src/artifacts/ArtifactFileOperation.js"; +import { ProviderCleanupError } from "../../../src/domain/providerCleanupError.js"; +import { AnalysisCancelledError } from "../../../src/domain/analysisErrorCore.js"; +import { + assertRecoveryCleanup, + recoveryFixture, +} from "../../fixtures/javascript-recovery/provider.js"; + +const fileClose = vi.hoisted(() => ({ + selected: (_path: string): boolean => false, + denied: false, + deniedRemoval: undefined as string | undefined, + cancel: undefined as AbortController | undefined, + handles: [] as FileHandle[], +})); + +vi.mock("node:fs/promises", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + rm: async (...args: Parameters) => { + if (args[0] === fileClose.deniedRemoval) + throw Object.assign( + new Error("controlled recovery workspace removal failure"), + { code: "EACCES" }, + ); + return actual.rm(...args); + }, + open: async (...args: Parameters) => { + const handle = await actual.open(...args); + if (typeof args[0] === "string" && fileClose.selected(args[0])) { + const close = handle.close.bind(handle); + fileClose.handles.push(handle); + onTestFinished(async () => { + if (handle.fd >= 0) await close(); + }); + handle.close = async () => { + fileClose.cancel?.abort(); + if (fileClose.denied) + throw new Error("controlled Wakaru file close failure"); + await close(); + }; + } + return handle; + }, + }; +}); + +afterEach(() => { + fileClose.selected = () => false; + fileClose.denied = false; + fileClose.deniedRemoval = undefined; + fileClose.cancel = undefined; + fileClose.handles = []; +}); + +const linuxX64 = it.skipIf( + process.platform !== "linux" || process.arch !== "x64", +); + +linuxX64.each([ + "engine", + "input", + "provenance", + "module", + "source-map", +] as const)( + "retains %s file cleanup, blocks new recovery and resolves old errors after exact-owner retry", + async (stage) => { + const fixture = await recoveryFixture(); + onTestFinished(async () => { + fileClose.denied = false; + fileClose.deniedRemoval = undefined; + await fixture.service.close(); + }); + fileClose.selected = (path) => + stage === "engine" + ? path === join(fixture.root, "fixture-wakaru") + : stage === "input" + ? path === fixture.path + : stage === "provenance" + ? path.endsWith("/modules/provenance.json") + : stage === "module" + ? path.endsWith("/modules/module.js") + : path.endsWith("/modules/module.js.map"); + fileClose.denied = true; + const input = { path: fixture.path, output_directory: fixture.output }; + const outcome = await fixture.service.recover(input); + expect(outcome.ok, JSON.stringify(outcome)).toBe(false); + if (outcome.ok) return; + expect(outcome.error).toBeInstanceOf(ProviderCleanupError); + expect(outcome.error.cleanupIncomplete).toBe(true); + expect(outcome.error.cause).toBeInstanceOf(ArtifactFileCleanupFailure); + expect(fileClose.handles).toHaveLength(1); + const handle = fileClose.handles[0]!; + expect(handle.fd).toBeGreaterThanOrEqual(0); + const resources = outcome.error.cleanupResources; + const workspace = resources.find((path) => + basename(path).startsWith("rea-wakaru-"), + ); + if (stage !== "engine") { + expect(workspace).toBeDefined(); + await expect(access(workspace!)).resolves.toBeUndefined(); + } + const launches = fixture.launches.length; + const blocked = await fixture.service.recover(input); + expect(blocked.ok).toBe(false); + if (blocked.ok) return; + expect(blocked.error).toBeInstanceOf(ProviderCleanupError); + expect(blocked.error.cleanupIncomplete).toBe(true); + expect(fileClose.handles).toHaveLength(1); + expect(fixture.launches).toHaveLength(launches); + await expect(fixture.service.close()).rejects.toBeInstanceOf( + ProviderCleanupError, + ); + expect(handle.fd).toBeGreaterThanOrEqual(0); + fileClose.denied = false; + if (stage === "input") { + fileClose.deniedRemoval = workspace; + await expect(fixture.service.close()).rejects.toBeInstanceOf( + ProviderCleanupError, + ); + expect(handle.fd).toBe(-1); + expect(outcome.error.cleanupIncomplete).toBe(true); + expect(outcome.error.cleanupResources).toEqual([workspace]); + expect(blocked.error.cleanupIncomplete).toBe(true); + fileClose.deniedRemoval = undefined; + } + await fixture.service.close(); + expect(handle.fd).toBe(-1); + expect(outcome.error.cleanupIncomplete).toBe(false); + expect(outcome.error.cleanupResources).toEqual([]); + expect(blocked.error.cleanupIncomplete).toBe(false); + if (workspace !== undefined) + await expect(access(workspace)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(access(fixture.output)).rejects.toMatchObject({ + code: "ENOENT", + }); + await assertRecoveryCleanup(fixture.launches); + }, +); + +linuxX64.each([ + ["engine", false], + ["engine", true], + ["input", false], + ["input", true], +] as const)( + "preserves %s observation and cancellation arriving during final close (close denied: %s)", + async (stage, denied) => { + const fixture = await recoveryFixture(); + onTestFinished(async () => { + fileClose.denied = false; + fileClose.deniedRemoval = undefined; + await fixture.service.close(); + }); + const selected = + stage === "engine" ? join(fixture.root, "fixture-wakaru") : fixture.path; + const bytes = await readFile(selected); + fileClose.selected = (path) => path === selected; + fileClose.denied = denied; + const controller = new AbortController(); + fileClose.cancel = controller; + const result = await fixture.service.recover( + { path: fixture.path, output_directory: fixture.output }, + { signal: controller.signal }, + ); + expect(result.ok).toBe(false); + if (result.ok) return; + expect(result.error).toBeInstanceOf( + denied ? ProviderCleanupError : AnalysisCancelledError, + ); + const failure = result.error.cause; + const physical = + failure instanceof ArtifactFileCleanupFailure + ? failure.fileCleanupFailure + : failure instanceof ArtifactFileCancelled + ? failure.fileCancellation + : undefined; + expect(physical?.outcome).toMatchObject({ + kind: "completed", + value: { + bytes: bytes.length, + sha256: createHash("sha256").update(bytes).digest("hex"), + }, + }); + expect(fixture.launches).toHaveLength(0); + if (denied) { + expect(result.error.cleanupIncomplete).toBe(true); + fileClose.denied = false; + fileClose.deniedRemoval = undefined; + await fixture.service.close(); + expect(result.error.cleanupIncomplete).toBe(false); + } + await expect(access(fixture.output)).rejects.toMatchObject({ + code: "ENOENT", + }); + }, +);