diff --git a/docs/firmware-analysis.md b/docs/firmware-analysis.md index 12886bd0d2..1674d8fd11 100644 --- a/docs/firmware-analysis.md +++ b/docs/firmware-analysis.md @@ -25,6 +25,12 @@ extraction have independent prerequisites; one does not require the other. Both require util-linux `prlimit`, normally `/usr/bin/prlimit`; override its absolute path with `REA_FIRMWARE_PRLIMIT_COMMAND` if needed. +REA retains at most 8 MiB of combined stdout and stderr per command. Exceeding +that diagnostic budget stops analysis with an explicit truncated-output error; +child process limits do not bound parent-side pipe retention. Completed commands +drain their output streams before recording diagnostics. Failed worker cleanup +retains the worker and workspace for retry before another request or at shutdown. + Unblob needs the external extractor for the selected format on its inherited `PATH`: for example, `debugfs` for ext2/ext3/ext4. Its default Landlock sandbox and skip rules stay enabled. Sandbox startup failures remain actionable errors; diff --git a/src/application/WebSourceLocationPorts.ts b/src/application/WebSourceLocationPorts.ts index d6ea281c1e..04fe1742c6 100644 --- a/src/application/WebSourceLocationPorts.ts +++ b/src/application/WebSourceLocationPorts.ts @@ -28,6 +28,7 @@ export interface WebSourceLocationArtifactPort { /** Replaceable codec receives map data and one explicit generated point. */ export interface WebSourceMapPort { + close?(): Promise; trace( input: { readonly text: string; diff --git a/src/application/WebSourceLocationService.ts b/src/application/WebSourceLocationService.ts index 2b52556104..c7c98d48c9 100644 --- a/src/application/WebSourceLocationService.ts +++ b/src/application/WebSourceLocationService.ts @@ -59,6 +59,10 @@ export class WebSourceLocationService { readonly decoder: WebSourceMapPort, ) {} + close(): Promise { + return this.decoder.close?.() ?? Promise.resolve(); + } + /** Shared CLI/MCP workflow preserves raw declarations and every equal-position match. */ async trace( rawInput: unknown, diff --git a/src/application/evm/EvmInterfacePort.ts b/src/application/evm/EvmInterfacePort.ts index 1d5c39ef45..e9e29b756f 100644 --- a/src/application/evm/EvmInterfacePort.ts +++ b/src/application/evm/EvmInterfacePort.ts @@ -8,6 +8,7 @@ import type { Result } from "../../domain/result.js"; /** Replaceable offline interface-recovery engine, independent of wallets and chain clients. */ export interface EvmInterfacePort { + close?(): Promise; inspect( input: InspectEvmInterfaceInput, options?: ExecutionOptions, diff --git a/src/application/evm/EvmInterfaceService.ts b/src/application/evm/EvmInterfaceService.ts index 9295e86454..eac1986d37 100644 --- a/src/application/evm/EvmInterfaceService.ts +++ b/src/application/evm/EvmInterfaceService.ts @@ -25,6 +25,10 @@ const diagnosticsSchema = z.object({ export class EvmInterfaceService { constructor(readonly provider: EvmInterfacePort) {} + close(): Promise { + return this.provider.close?.() ?? Promise.resolve(); + } + /** Preserve selected byte identity while clearly labeling recovered ABI candidates as inferences. */ async inspect( rawInput: unknown, diff --git a/src/application/firmware/FirmwareAnalysisPort.ts b/src/application/firmware/FirmwareAnalysisPort.ts index a82d6c994e..9db644ebd8 100644 --- a/src/application/firmware/FirmwareAnalysisPort.ts +++ b/src/application/firmware/FirmwareAnalysisPort.ts @@ -8,6 +8,7 @@ import type { Result } from "../../domain/result.js"; /** File-oriented firmware engines; executable admission remains a separate operation. */ export interface FirmwareAnalysisPort { + close?(): Promise; execute( request: FirmwareRequest, options?: ExecutionOptions, diff --git a/src/application/firmware/FirmwareAnalysisService.ts b/src/application/firmware/FirmwareAnalysisService.ts index eac38240bb..50e7ee7bb2 100644 --- a/src/application/firmware/FirmwareAnalysisService.ts +++ b/src/application/firmware/FirmwareAnalysisService.ts @@ -18,6 +18,10 @@ import { err, ok, type Result } from "../../domain/result.js"; export class FirmwareAnalysisService { constructor(readonly provider: FirmwareAnalysisPort) {} + close(): Promise { + return this.provider.close?.() ?? Promise.resolve(); + } + /** Execute an explicit inspection or extraction and preserve its producing identity. */ async execute( operation: FirmwareOperation, diff --git a/src/application/javascript/JavaScriptRecoveryPort.ts b/src/application/javascript/JavaScriptRecoveryPort.ts index a1975e1677..fb1bab025f 100644 --- a/src/application/javascript/JavaScriptRecoveryPort.ts +++ b/src/application/javascript/JavaScriptRecoveryPort.ts @@ -8,6 +8,7 @@ import type { /** Typed boundary for upstream source recovery without exposing its protocol. */ export interface JavaScriptRecoveryPort { + close?(): Promise; recover( input: JavaScriptRecoveryInput, options?: ExecutionOptions, diff --git a/src/application/javascript/JavaScriptRecoveryService.ts b/src/application/javascript/JavaScriptRecoveryService.ts index c21e9e1e3f..3572403015 100644 --- a/src/application/javascript/JavaScriptRecoveryService.ts +++ b/src/application/javascript/JavaScriptRecoveryService.ts @@ -23,6 +23,10 @@ const OPERATION = "recover_javascript_sources"; export class JavaScriptRecoveryService { constructor(readonly provider: JavaScriptRecoveryPort) {} + close(): Promise { + return this.provider.close?.() ?? Promise.resolve(); + } + /** Recover sources and bind the result to its original immutable input. */ async recover( input: unknown, diff --git a/src/cli/evmCommands.ts b/src/cli/evmCommands.ts index 6a9dd9504c..735cfee1ec 100644 --- a/src/cli/evmCommands.ts +++ b/src/cli/evmCommands.ts @@ -24,14 +24,24 @@ export const registerEvmCommands = ( .describe("Selected carrier representation"), }), run: ({ args }) => - withCommandCancellation((signal) => - logCliCommand(logger, CLI_COMMANDS.inspectEvmInterface, async () => { - const result = await service.inspect( - { path: resolve(args.path), encoding: args.encoding }, - { signal }, + withCommandCancellation(async (signal) => { + try { + return await logCliCommand( + logger, + CLI_COMMANDS.inspectEvmInterface, + async () => { + const result = await service.inspect( + { path: resolve(args.path), encoding: args.encoding }, + { signal }, + ); + return result.ok + ? result.value + : projectAnalysisError(result.error); + }, ); - return result.ok ? result.value : projectAnalysisError(result.error); - }), - ), + } finally { + await service.close(); + } + }), }); }; diff --git a/src/cli/firmwareCommands.ts b/src/cli/firmwareCommands.ts index 8822c2354f..9c59711e21 100644 --- a/src/cli/firmwareCommands.ts +++ b/src/cli/firmwareCommands.ts @@ -26,8 +26,12 @@ export const registerFirmwareCommands = ( ) => withCommandCancellation((signal) => logCliCommand(logger, name, async () => { - const result = await service.execute(operation, input, { signal }); - return result.ok ? result.value : projectAnalysisError(result.error); + try { + const result = await service.execute(operation, input, { signal }); + return result.ok ? result.value : projectAnalysisError(result.error); + } finally { + await service.close(); + } }), ); const path = z.string().describe("Local firmware path; never executed"); diff --git a/src/cli/javascriptRecoveryCommands.ts b/src/cli/javascriptRecoveryCommands.ts index d51f2e471d..705851f977 100644 --- a/src/cli/javascriptRecoveryCommands.ts +++ b/src/cli/javascriptRecoveryCommands.ts @@ -41,29 +41,33 @@ export const registerJavaScriptRecoveryCommands = ( ), }), run: ({ args, options }) => - withCommandCancellation((signal) => - logCliCommand( - logger, - CLI_COMMANDS.recoverJavaScriptSources, - async () => { - const result = await service.recover( - { - path: args.path, - output_directory: args.outputDirectory, - ...(options.extractionMode === undefined - ? {} - : { extraction_mode: options.extractionMode }), - ...(options.rewriteLevel === undefined - ? {} - : { rewrite_level: options.rewriteLevel }), - }, - { signal }, - ); - return result.ok - ? result.value - : projectAnalysisError(result.error); - }, - ), - ), + withCommandCancellation(async (signal) => { + try { + return await logCliCommand( + logger, + CLI_COMMANDS.recoverJavaScriptSources, + async () => { + const result = await service.recover( + { + path: args.path, + output_directory: args.outputDirectory, + ...(options.extractionMode === undefined + ? {} + : { extraction_mode: options.extractionMode }), + ...(options.rewriteLevel === undefined + ? {} + : { rewrite_level: options.rewriteLevel }), + }, + { signal }, + ); + return result.ok + ? result.value + : projectAnalysisError(result.error); + }, + ); + } finally { + await service.close(); + } + }), }); }; diff --git a/src/cli/webSourceLocationCommands.ts b/src/cli/webSourceLocationCommands.ts index 42175e95d8..1025733abb 100644 --- a/src/cli/webSourceLocationCommands.ts +++ b/src/cli/webSourceLocationCommands.ts @@ -38,19 +38,32 @@ export const registerWebSourceLocationCommands = ( .describe("Zero-based UTF-16 generated column"), }), run: ({ args }) => - withCommandCancellation((signal) => - logCliCommand(logger, CLI_COMMANDS.traceWebSourceLocation, async () => { - const result = await service.trace( - { - manifest_path: args.manifestPath, - script_index: args.scriptIndex, - source_map: { path: args.sourceMapPath, url: args.sourceMapUrl }, - generated_position: { line: args.line, column: args.column }, + withCommandCancellation(async (signal) => { + try { + return await logCliCommand( + logger, + CLI_COMMANDS.traceWebSourceLocation, + async () => { + const result = await service.trace( + { + manifest_path: args.manifestPath, + script_index: args.scriptIndex, + source_map: { + path: args.sourceMapPath, + url: args.sourceMapUrl, + }, + generated_position: { line: args.line, column: args.column }, + }, + { signal }, + ); + return result.ok + ? result.value + : projectAnalysisError(result.error); }, - { signal }, ); - return result.ok ? result.value : projectAnalysisError(result.error); - }), - ), + } finally { + await service.close(); + } + }), }); }; diff --git a/src/evm/EvmoleInterfaceProvider.ts b/src/evm/EvmoleInterfaceProvider.ts index c38aefa091..42a4e08480 100644 --- a/src/evm/EvmoleInterfaceProvider.ts +++ b/src/evm/EvmoleInterfaceProvider.ts @@ -31,6 +31,7 @@ import { ProviderCleanupError } from "../domain/providerCleanupError.js"; import { err, ok, type Result } from "../domain/result.js"; import { runOwnedCommand } from "../process/OwnedCommand.js"; import { PrivateRuntimeRoot } from "../process/PrivateRuntimeRoot.js"; +import type { ProviderProcessSupervisor } from "../process/ProviderProcess.js"; import { evmInterfaceFailure, capturedEvmOutput, @@ -75,6 +76,16 @@ type Launcher = NonNullable[2]>["launcher"]; /** Run unchanged EVMole/WASM in an owned worker; no target, chain or network execution. */ export class EvmoleInterfaceProvider implements EvmInterfacePort { + #tail: Promise = Promise.resolve(); + #closing = false; + #closePromise: Promise | undefined; + #pendingCleanup: + | { + root: Pick | undefined; + process: ProviderProcessSupervisor | undefined; + } + | undefined; + constructor( readonly environment: Readonly = process.env, readonly launcher?: Launcher, @@ -84,11 +95,29 @@ export class EvmoleInterfaceProvider implements EvmInterfacePort { ) {} /** Bind recovered interface candidates to both original carrier and decoded byte identity. */ - async inspect( + inspect( input: InspectEvmInterfaceInput, options?: ExecutionOptions, ): Promise> { + if (this.#closing) return Promise.resolve(err(this.#closedError())); + const operation = this.#tail.then(() => + this.#closing ? err(this.#closedError()) : this.#inspect(input, options), + ); + this.#tail = operation.then( + () => undefined, + () => undefined, + ); + return operation; + } + + async #inspect( + input: InspectEvmInterfaceInput, + options?: ExecutionOptions, + ): Promise> { + const pending = await this.#retire(); + if (pending !== undefined) return err(pending); let root: Pick | undefined; + let processOwner: ProviderProcessSupervisor | undefined; let retainedOutput: AnalysisCapturedOutput | undefined; let phase: "configuration" | "artifact-read" | "worker" = "configuration"; let workerLimits: EvmWorkerLimits | undefined; @@ -321,6 +350,8 @@ export class EvmoleInterfaceProvider implements EvmInterfacePort { }), ); } catch (cause: unknown) { + if (cause instanceof OwnedCommandFailure) + processOwner = cause.cleanupOwner; let fileFailure: EvmFileSizeFailureEvidence | undefined; if ( cause instanceof OwnedCommandFailure && @@ -368,14 +399,42 @@ export class EvmoleInterfaceProvider implements EvmInterfacePort { ), ); } + if (processOwner !== undefined) { + this.#pendingCleanup = { root, process: processOwner }; + root = undefined; + return err( + new ProviderCleanupError( + EVMOLE_PROVIDER_IDENTITY.id, + [ + processOwner.launch.ownership?.runId ?? "owned-worker", + ...(this.#pendingCleanup.root === undefined + ? [] + : [this.#pendingCleanup.root.path]), + ], + { + reason: "Owned EVM worker cleanup could not be confirmed", + previous_error: result.ok + ? null + : projectAnalysisError(result.error), + ...(retainedOutput === undefined + ? {} + : { captured_output: { ...retainedOutput } }), + }, + { operation: OPERATION }, + ), + ); + } if (root !== undefined) { try { await root.close(); } catch (cause: unknown) { + const retainedRoot = root; + this.#pendingCleanup = { root: retainedRoot, process: undefined }; + root = undefined; return err( new ProviderCleanupError( EVMOLE_PROVIDER_IDENTITY.id, - [root.path], + [retainedRoot.path], { reason: cause instanceof Error ? cause.message : String(cause), previous_error: result.ok @@ -401,4 +460,59 @@ export class EvmoleInterfaceProvider implements EvmInterfacePort { ) : result; } + + async close(): Promise { + this.#closing = true; + this.#closePromise ??= this.#tail + .then(async () => { + const failure = await this.#retire(); + if (failure !== undefined) throw failure; + }) + .catch((cause: unknown) => { + this.#closePromise = undefined; + throw cause; + }); + return this.#closePromise; + } + + #closedError(): ProviderCleanupError { + return new ProviderCleanupError( + EVMOLE_PROVIDER_IDENTITY.id, + [], + { reason: "Provider is closing" }, + { operation: OPERATION }, + ); + } + + async #retire(): Promise { + const pending = this.#pendingCleanup; + if (pending === undefined) return undefined; + if (pending.process !== undefined) { + const stopped = await pending.process.stop(); + if (stopped.status === "incomplete") + return new ProviderCleanupError( + EVMOLE_PROVIDER_IDENTITY.id, + [pending.process.launch.ownership?.runId ?? "owned-worker"], + { reason: stopped.reason }, + { operation: OPERATION }, + ); + pending.process = undefined; + } + if (pending.root !== undefined) { + const root = pending.root; + try { + await root.close(); + pending.root = undefined; + } catch (cause: unknown) { + return new ProviderCleanupError( + EVMOLE_PROVIDER_IDENTITY.id, + [root.path], + { reason: cause instanceof Error ? cause.message : String(cause) }, + { operation: OPERATION, cause }, + ); + } + } + this.#pendingCleanup = undefined; + return undefined; + } } diff --git a/src/firmware/FirmwareCommand.ts b/src/firmware/FirmwareCommand.ts index 27bd57aa39..b5db919f59 100644 --- a/src/firmware/FirmwareCommand.ts +++ b/src/firmware/FirmwareCommand.ts @@ -113,6 +113,7 @@ export const runFirmwareCommand = async (context: { launcher?: FirmwareLauncher; outputBudget?: { root: string; bytes: number; entries: number }; acceptReportedExtractionFailure?: boolean; + retainCleanup(supervisor: ProviderProcessSupervisor): void; }) => { const { engine, operation, signal } = context; signal.throwIfAborted(); @@ -141,16 +142,43 @@ export const runFirmwareCommand = async (context: { NO_COLOR: "1", }, }); - const supervisor = new ProviderProcessSupervisor({ - ...spawned, - ownsProcessLifetime: true, - cleanup: - spawned.cleanup ?? (() => cleanupOwnedProcessGroup(spawned.ownership)), - }); + let diagnosticTruncated = false; + const supervisor = new ProviderProcessSupervisor( + { + ...spawned, + ownsProcessLifetime: true, + cleanup: + spawned.cleanup ?? (() => cleanupOwnedProcessGroup(spawned.ownership)), + }, + { + maxDiagnosticBytes: FIRMWARE_LIMITS.diagnosticBytes, + onDiagnostic: (event) => { + if (event.type === "output" && event.truncated === true) + diagnosticTruncated = true; + }, + }, + ); + const checkDiagnostics = () => { + if (diagnosticTruncated) { + const snapshot = supervisor.snapshot(); + throw new AnalysisOutputError( + operation, + `Tool diagnostics exceeded the ${FIRMWARE_LIMITS.diagnosticBytes}-byte capture budget`, + { + capturedOutput: { + stdout: snapshot.stdout.text, + stderr: snapshot.stderr.text, + truncated: true, + }, + }, + ); + } + }; let failure: unknown; try { const deadline = Date.now() + FIRMWARE_LIMITS.timeoutMs; - while (!(await supervisor.waitForExit(100))) { + while (!(await supervisor.waitForOutputClose(100))) { + checkDiagnostics(); if (signal.aborted) throw new AnalysisCancelledError(operation); if (Date.now() >= deadline) throw new AnalysisTimeoutError(operation, FIRMWARE_LIMITS.timeoutMs); @@ -163,14 +191,15 @@ export const runFirmwareCommand = async (context: { signal, ); } + checkDiagnostics(); if (signal.aborted) throw new AnalysisCancelledError(operation); } catch (cause: unknown) { failure = cause; } const stopped = await supervisor.stop(); const snapshot = supervisor.snapshot(); - supervisor.dispose(); - if (stopped.status === "incomplete") + if (stopped.status === "incomplete") { + context.retainCleanup(supervisor); throw new ProviderCleanupError(engine, [runId, context.cwd], { reason: stopped.reason, leader_pid: spawned.ownership.leaderPid, @@ -183,6 +212,7 @@ export const runFirmwareCommand = async (context: { stderr_bytes: snapshot.stderr.bytes, previous_error: failure instanceof Error ? failure.message : null, }); + } if (failure !== undefined) throw failure; if ( snapshot.exitCode !== 0 && diff --git a/src/firmware/FirmwareFiles.ts b/src/firmware/FirmwareFiles.ts index a118984b81..4a5aa3d842 100644 --- a/src/firmware/FirmwareFiles.ts +++ b/src/firmware/FirmwareFiles.ts @@ -3,8 +3,8 @@ import { constants, createReadStream, createWriteStream } from "node:fs"; import { chmod, lstat, + opendir, open, - readdir, readlink, realpath, } from "node:fs/promises"; @@ -29,6 +29,13 @@ export interface FirmwareEntry { readonly linkTarget: string | null; } +const throwIfFirmwareInventoryCancelled = ( + signal: AbortSignal | undefined, + operation: string, +): void => { + if (signal?.aborted === true) throw new AnalysisCancelledError(operation); +}; + /** Hash a regular tool or extracted file without retaining its bytes in memory. */ export const hashFirmwareFile = async ( path: string, @@ -184,7 +191,7 @@ export const inventoryFirmwareOutput = async ( let bytes = 0; let count = 0; while (pending.length > 0) { - if (signal?.aborted === true) throw new AnalysisCancelledError(operation); + throwIfFirmwareInventoryCancelled(signal, operation); const directory = pending.pop(); if (directory === undefined) break; const directoryStat = await lstat(directory).catch((cause: unknown) => { @@ -198,18 +205,28 @@ export const inventoryFirmwareOutput = async ( directoryStat.isSymbolicLink() ) continue; - const names = await readdir(directory).catch((cause: unknown) => { + const handle = await opendir(directory).catch((cause: unknown) => { if (cause instanceof Error && "code" in cause && cause.code === "ENOENT") - return []; + return undefined; throw cause; }); + if (handle === undefined) continue; + const names: string[] = []; + try { + for await (const entry of handle) { + throwIfFirmwareInventoryCancelled(signal, operation); + count++; + if (count > maxEntries) + throw new AnalysisOutputError( + operation, + `Staging entry budget exceeded (${maxEntries})`, + ); + names.push(entry.name); + } + } finally { + await handle.close().catch(() => undefined); + } for (const name of names.sort()) { - count++; - if (count > maxEntries) - throw new AnalysisOutputError( - operation, - `Staging entry budget exceeded (${maxEntries})`, - ); const path = join(directory, name); const stat = await lstat(path).catch((cause: unknown) => { if ( diff --git a/src/firmware/FirmwareProvider.ts b/src/firmware/FirmwareProvider.ts index a8b3bf560f..8be7901404 100644 --- a/src/firmware/FirmwareProvider.ts +++ b/src/firmware/FirmwareProvider.ts @@ -20,6 +20,7 @@ import { ProviderAdapterError } from "../domain/providerAdapterError.js"; import { ProviderCleanupError } from "../domain/providerCleanupError.js"; import { err, ok } from "../domain/result.js"; import { PrivateRuntimeRoot } from "../process/PrivateRuntimeRoot.js"; +import type { ProviderProcessSupervisor } from "../process/ProviderProcess.js"; import { publishFirmwareExtraction } from "./FirmwarePublication.js"; import { FIRMWARE_LIMITS, @@ -65,6 +66,15 @@ const limitations = [ /** Queued bring-your-own firmware adapters with private snapshots and owned cleanup. */ export class FirmwareProvider implements FirmwareAnalysisPort { #tail: Promise = Promise.resolve(); + #closed = false; + #pendingCleanup: + | { + supervisor?: ProviderProcessSupervisor; + root: PrivateRuntimeRoot; + engine: string; + operation: string; + } + | undefined; constructor( readonly environment: Readonly< Record @@ -77,6 +87,7 @@ export class FirmwareProvider implements FirmwareAnalysisPort { request: FirmwareRequest, options?: ExecutionOptions, ): Promise { + if (this.#closed) return err(new AnalysisCancelledError(request.operation)); const predecessor = this.#tail; let release = () => {}; this.#tail = new Promise((resolve) => { @@ -94,12 +105,57 @@ export class FirmwareProvider implements FirmwareAnalysisPort { ); } try { + if (this.#closed) + return err(new AnalysisCancelledError(request.operation)); + try { + await this.#retryCleanup(); + } catch (cause: unknown) { + return err( + cause instanceof AnalysisError + ? cause + : new ProviderCleanupError("firmware", [], { + reason: cause instanceof Error ? cause.message : String(cause), + }), + ); + } return await this.#execute(request, options); } finally { release(); } } + /** Retry identity-bound cleanup before relinquishing private workspaces. */ + async close(): Promise { + this.#closed = true; + await this.#tail; + await this.#retryCleanup(); + } + + async #retryCleanup(): Promise { + const pending = this.#pendingCleanup; + if (pending === undefined) return; + const stopped = await pending.supervisor?.stop(); + if (stopped?.status === "incomplete") + throw new ProviderCleanupError( + pending.engine, + [pending.root.path], + { reason: stopped.reason }, + { operation: pending.operation }, + ); + delete pending.supervisor; + try { + await pending.root.close(); + } catch (cause: unknown) { + throw new ProviderCleanupError( + pending.engine, + [pending.root.path], + { reason: cause instanceof Error ? cause.message : String(cause) }, + { operation: pending.operation, cause }, + ); + } + this.#pendingCleanup = undefined; + } + async #execute( request: FirmwareRequest, options?: ExecutionOptions, @@ -121,6 +177,7 @@ export class FirmwareProvider implements FirmwareAnalysisPort { request.operation, ); root = await PrivateRuntimeRoot.create({ prefix: "rea-firmware-" }); + const ownedRoot = root; const target = await snapshotFirmware(request, root.path, signal); const run = ( args: readonly string[], @@ -134,6 +191,14 @@ export class FirmwareProvider implements FirmwareAnalysisPort { cwd: root?.path ?? "", environment: this.environment, signal, + retainCleanup: (supervisor) => { + this.#pendingCleanup = { + supervisor, + root: ownedRoot, + engine: engineName, + operation: request.operation, + }; + }, ...(this.launcher === undefined ? {} : { launcher: this.launcher }), ...(outputBudget === undefined ? {} @@ -315,10 +380,19 @@ export class FirmwareProvider implements FirmwareAnalysisPort { }), ); } - // Never remove a workspace while process ownership remains uncertain. - // A leaked root stays recorded on this result and is not deleted. The next - // operation may create a new root; this failure is not replayed for it. - return finishFirmwareWorkspace(root, outcome, engineName, request); + const finished = await finishFirmwareWorkspace( + root, + outcome, + engineName, + request, + ); + if (!finished.ok && finished.error.cleanupIncomplete && root !== undefined) + this.#pendingCleanup ??= { + root, + engine: engineName, + operation: request.operation, + }; + return finished; } } diff --git a/src/firmware/FirmwareRelease.ts b/src/firmware/FirmwareRelease.ts index b68a41511e..8699013c33 100644 --- a/src/firmware/FirmwareRelease.ts +++ b/src/firmware/FirmwareRelease.ts @@ -29,6 +29,7 @@ export const UNBLOB_PROVIDER_IDENTITY = { export const FIRMWARE_LIMITS = { inputBytes: 128 * 1024 * 1024, reportBytes: 8 * 1024 * 1024, + diagnosticBytes: 8 * 1024 * 1024, addressSpaceBytes: 1024 * 1024 * 1024, timeoutMs: 120_000, } as const; diff --git a/src/javascript/recovery/RecoveryFiles.ts b/src/javascript/recovery/RecoveryFiles.ts index 3e284179fc..636232c074 100644 --- a/src/javascript/recovery/RecoveryFiles.ts +++ b/src/javascript/recovery/RecoveryFiles.ts @@ -1,6 +1,6 @@ import { createHash } from "node:crypto"; import { constants } from "node:fs"; -import { lstat, open, readdir, writeFile } from "node:fs/promises"; +import { lstat, open, opendir, writeFile } from "node:fs/promises"; import { isAbsolute, join } from "node:path"; import { TextDecoder } from "node:util"; import { @@ -138,18 +138,27 @@ export const inventoryRecoveryFiles = async ( OPERATION, `Unsafe staging directory: ${directory}`, ); - for (const entry of await readdir(directory, { withFileTypes: true })) { + const handle = await opendir(directory); + const children: string[] = []; + try { + for await (const entry of handle) { + checkRecoveryCancellation(signal); + entries++; + if (entries > RECOVERY_LIMITS.outputEntries) + throw new AnalysisOutputError( + OPERATION, + "Recovery staging exceeds the 10000-entry resource limit", + ); + children.push(entry.name); + } + } finally { + await handle.close().catch(() => undefined); + } + for (const name of children) { checkRecoveryCancellation(signal); - entries++; - const relativePath = - prefix === "" ? entry.name : `${prefix}/${entry.name}`; - const path = join(directory, entry.name); + const relativePath = prefix === "" ? name : `${prefix}/${name}`; + const path = join(directory, name); const info = await lstat(path); - if (entries > RECOVERY_LIMITS.outputEntries) - throw new AnalysisOutputError( - OPERATION, - "Recovery staging exceeds the 10000-entry resource limit", - ); if (info.isSymbolicLink() || (!info.isDirectory() && !info.isFile())) throw new AnalysisOutputError( OPERATION, diff --git a/src/javascript/recovery/WakaruCommand.ts b/src/javascript/recovery/WakaruCommand.ts index 464cd17459..e024474818 100644 --- a/src/javascript/recovery/WakaruCommand.ts +++ b/src/javascript/recovery/WakaruCommand.ts @@ -33,6 +33,31 @@ export type WakaruLauncher = ( options: OwnedProviderProcessSpawnOptions, ) => Promise; +/** Cleanup failure carrying the live supervisor so the provider can retry it. */ +export class WakaruCleanupFailure extends ProviderCleanupError { + constructor( + readonly cleanupOwner: ProviderProcessSupervisor, + readonly runId: string, + readonly cwd: string, + reason: string, + previousError: unknown, + snapshot: ReturnType, + ) { + super( + "wakaru", + [runId, cwd], + { + reason, + previous_error: + previousError instanceof Error ? previousError.message : null, + exit_code: snapshot.exitCode ?? null, + signal: snapshot.signal ?? null, + }, + { operation: OPERATION, cause: previousError }, + ); + } +} + /** Resolve and fingerprint explicitly configured tools; acquire nothing at startup. */ export const resolveWakaruCommand = async ( environment: Readonly>, @@ -158,14 +183,15 @@ export const runWakaruCommand = async (context: WakaruCommandContext) => { } const stopped = await supervisor.stop(); const snapshot = supervisor.snapshot(); - supervisor.dispose(); if (stopped.status === "incomplete") - throw new ProviderCleanupError("wakaru", [runId, context.cwd], { - reason: stopped.reason, - previous_error: failure instanceof Error ? failure.message : null, - exit_code: snapshot.exitCode ?? null, - signal: snapshot.signal ?? null, - }); + throw new WakaruCleanupFailure( + supervisor, + runId, + context.cwd, + stopped.reason, + failure, + snapshot, + ); if (failure !== undefined) throw failure; if (context.signal?.aborted === true) throw new AnalysisCancelledError(OPERATION); diff --git a/src/javascript/recovery/WakaruProvider.ts b/src/javascript/recovery/WakaruProvider.ts index 1273b26eda..8f3181f2cb 100644 --- a/src/javascript/recovery/WakaruProvider.ts +++ b/src/javascript/recovery/WakaruProvider.ts @@ -20,14 +20,30 @@ import { recoveryInputFailure, snapshotRecoveryInput, } from "./RecoveryFiles.js"; -import { resolveWakaruCommand, type WakaruLauncher } from "./WakaruCommand.js"; +import { + resolveWakaruCommand, + type WakaruLauncher, + WakaruCleanupFailure, +} from "./WakaruCommand.js"; +import type { ProviderProcessSupervisor } from "../../process/ProviderProcess.js"; import { prepareWakaruExecution } from "./WakaruExecution.js"; import { RECOVERY_LIMITS } from "./WakaruRelease.js"; const OPERATION = "recover_javascript_sources"; -/** Stateless adapter; each recovery owns its snapshot, process and publication. */ +/** Serialized recovery adapter retaining failed process, workspace and publication cleanup. */ export class WakaruProvider implements JavaScriptRecoveryPort { + #tail: Promise = Promise.resolve(); + #closing = false; + #closePromise: Promise | undefined; + #pendingCleanup: + | { + root: string | undefined; + process: ProviderProcessSupervisor | undefined; + tree: SafeOutputTree | undefined; + } + | undefined; + constructor( readonly environment: Readonly< Record @@ -36,10 +52,46 @@ export class WakaruProvider implements JavaScriptRecoveryPort { ) {} /** Recover one script using validated upstream reports and verified output bytes. */ - async recover( + recover( input: JavaScriptRecoveryInput, options?: ExecutionOptions, ): Promise> { + if (this.#closing) + return Promise.resolve( + err( + new ProviderCleanupError( + "wakaru", + [], + { reason: "Provider is closing" }, + { operation: OPERATION }, + ), + ), + ); + const operation = this.#tail.then(() => + this.#closing + ? err( + new ProviderCleanupError( + "wakaru", + [], + { reason: "Provider is closing" }, + { operation: OPERATION }, + ), + ) + : this.#recover(input, options), + ); + this.#tail = operation.then( + () => undefined, + () => undefined, + ); + return operation; + } + + async #recover( + input: JavaScriptRecoveryInput, + options?: ExecutionOptions, + ): Promise> { + const pending = await this.#retire(); + if (pending !== undefined) return err(pending); let root: string | undefined; let tree: SafeOutputTree | undefined; let failure: unknown; @@ -88,20 +140,42 @@ export class WakaruProvider implements JavaScriptRecoveryPort { } if (failure !== undefined) { const residuals: string[] = []; + const cleanupOwner = + failure instanceof WakaruCleanupFailure + ? failure.cleanupOwner + : undefined; + let failedTree: SafeOutputTree | undefined; try { await tree?.rollback(); } catch { - if (tree !== undefined) residuals.push(tree.outputRoot); + if (tree !== undefined) { + failedTree = tree; + residuals.push(tree.outputRoot); + } + } + if (cleanupOwner !== undefined) + residuals.push(cleanupOwner.launch.ownership?.runId ?? "wakaru-worker"); + if (root !== undefined && cleanupOwner === undefined) { + const failedRoot = root; + try { + await rm(failedRoot, { recursive: true, force: true }); + root = undefined; + } catch { + residuals.push(failedRoot); + } } if ( - root !== undefined && - !(failure instanceof AnalysisError && failure.cleanupIncomplete) + cleanupOwner !== undefined || + failedTree !== undefined || + root !== undefined ) { - try { - await rm(root, { recursive: true, force: true }); - } catch { - residuals.push(root); - } + this.#pendingCleanup = { + root, + process: cleanupOwner, + tree: failedTree, + }; + if (root !== undefined) residuals.push(root); + root = undefined; } if (residuals.length > 0) return err( @@ -124,4 +198,67 @@ export class WakaruProvider implements JavaScriptRecoveryPort { ) : ok(execution); } + + async close(): Promise { + this.#closing = true; + this.#closePromise ??= this.#tail + .then(async () => { + const failure = await this.#retire(); + if (failure !== undefined) throw failure; + }) + .catch((cause: unknown) => { + this.#closePromise = undefined; + throw cause; + }); + return this.#closePromise; + } + + async #retire(): Promise { + const pending = this.#pendingCleanup; + if (pending === undefined) return undefined; + 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 }, + ); + pending.process = undefined; + } + 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 }, + ); + } + } + 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 }, + ); + } + } + this.#pendingCleanup = undefined; + return undefined; + } } diff --git a/src/javascript/sourceMaps/SourceMapCommand.ts b/src/javascript/sourceMaps/SourceMapCommand.ts index 2caebcdb2f..ea1a9fca2f 100644 --- a/src/javascript/sourceMaps/SourceMapCommand.ts +++ b/src/javascript/sourceMaps/SourceMapCommand.ts @@ -12,6 +12,7 @@ import { } from "../../domain/analysisErrorCore.js"; import { ProviderAdapterError } from "../../domain/providerAdapterError.js"; import { ProviderCleanupError } from "../../domain/providerCleanupError.js"; +import type { JsonValue } from "../../domain/jsonValue.js"; import { err, ok, type Result } from "../../domain/result.js"; import { WEB_SOURCE_MAP_LIMITS } from "../../domain/webSourceLocation.js"; import { @@ -48,6 +49,22 @@ export interface SourceMapDecoderDependencies { readonly prepareOwnership?: (signal?: AbortSignal) => Promise; } +export interface SourceMapCleanupOwner { + close( + previousError: string | null, + ): Promise; +} + +export class SourceMapCleanupFailure extends ProviderCleanupError { + constructor( + readonly cleanupOwner: SourceMapCleanupOwner, + resources: readonly string[], + diagnostics: Readonly>, + ) { + super(PROVIDER, resources, diagnostics, { operation: OPERATION }); + } +} + /** Run a trusted codec command and release its process/root before returning output. */ export const runSourceMapCommand = async ( request: string, @@ -94,7 +111,7 @@ export const runSourceMapCommand = async ( : result; }; -class OwnedCodec { +class OwnedCodec implements SourceMapCleanupOwner { readonly #runId = `rea-source-map-${randomUUID()}`; #runtime: SourceMapCodecRuntime | undefined; #supervisor: ProviderProcessSupervisor | undefined; @@ -231,21 +248,27 @@ class OwnedCodec { previousError: string | null, ): Promise { const failures: string[] = []; + let processReleased = this.#supervisor === undefined; if (this.#supervisor !== undefined) { const stopped = await this.#supervisor.stop(); if (stopped.status === "incomplete") failures.push(stopped.reason); + else { + processReleased = true; + this.#supervisor = undefined; + } } - if (this.#runtime !== undefined) { + if (processReleased && this.#runtime !== undefined) { try { await this.#runtime.close(); + this.#runtime = undefined; } catch (cause: unknown) { failures.push(message(cause)); } } return failures.length === 0 ? undefined - : new ProviderCleanupError( - PROVIDER, + : new SourceMapCleanupFailure( + this, [ this.#runId, ...(this.#runtime === undefined ? [] : [this.#runtime.path]), @@ -256,7 +279,6 @@ class OwnedCodec { previous_error: previousError, codec_pid: this.#supervisor?.launch.process.pid ?? null, }, - { operation: OPERATION }, ); } } diff --git a/src/javascript/sourceMaps/SourceMapDecoder.ts b/src/javascript/sourceMaps/SourceMapDecoder.ts index 7c49d5fa97..8db70ee472 100644 --- a/src/javascript/sourceMaps/SourceMapDecoder.ts +++ b/src/javascript/sourceMaps/SourceMapDecoder.ts @@ -18,19 +18,64 @@ import { sourceMapCodecReplySchema, } from "./SourceMapCodecProtocol.js"; import { + SourceMapCleanupFailure, runSourceMapCommand, + type SourceMapCleanupOwner, type SourceMapDecoderDependencies, } from "./SourceMapCommand.js"; +import { ProviderCleanupError } from "../../domain/providerCleanupError.js"; +import { jsonParts } from "../../domain/jsonSerialization.js"; const OPERATION = "trace_web_source_location"; /** Integrate the pinned upstream codec through an independently bounded owned process. */ export class SourceMapDecoder implements WebSourceMapPort { + #tail: Promise = Promise.resolve(); + #closing = false; + #closePromise: Promise | undefined; + #pendingCleanup: SourceMapCleanupOwner | undefined; + constructor(readonly dependencies: SourceMapDecoderDependencies = {}) {} /** Return complete point evidence after process and private-root cleanup. */ - async trace( + trace( input: Parameters[0], options?: ExecutionOptions, ): Promise> { + if (this.#closing) + return Promise.resolve( + err( + new ProviderCleanupError( + "source-map-decoder", + [], + { reason: "Provider is closing" }, + { operation: OPERATION }, + ), + ), + ); + const operation = this.#tail.then(() => + this.#closing + ? err( + new ProviderCleanupError( + "source-map-decoder", + [], + { reason: "Provider is closing" }, + { operation: OPERATION }, + ), + ) + : this.#trace(input, options), + ); + this.#tail = operation.then( + () => undefined, + () => undefined, + ); + return operation; + } + + async #trace( + input: Parameters[0], + options?: ExecutionOptions, + ): Promise> { + const cleanupFailure = await this.#retryCleanup(); + if (cleanupFailure !== undefined) return err(cleanupFailure); if (options?.signal?.aborted) return err(new AnalysisCancelledError(OPERATION)); const request = sourceMapCodecInputSchema.safeParse({ @@ -40,26 +85,38 @@ export class SourceMapDecoder implements WebSourceMapPort { }); if (!request.success) return err(new AnalysisInputError(OPERATION, { cause: request.error })); - const serialized = JSON.stringify(request.data); - if ( - Buffer.byteLength(input.text) > WEB_SOURCE_MAP_LIMITS.mapBytes || - Buffer.byteLength(serialized) > WEB_SOURCE_MAP_LIMITS.outputBytes - ) - return err( - new ArtifactOperationError( - OPERATION, - "limit", - undefined, - `${input.path}: Source-map input exceeds the 4 MiB map or 32 MiB encoded request budget.`, - ), + const capacityFailure = () => + new ArtifactOperationError( + OPERATION, + "limit", + undefined, + `${input.path}: Source-map input exceeds the 4 MiB map or 32 MiB encoded request budget.`, ); + if (Buffer.byteLength(input.text) > WEB_SOURCE_MAP_LIMITS.mapBytes) + return err(capacityFailure()); + const encoded: string[] = []; + let encodedBytes = 0; + for (const part of jsonParts(request.data)) { + const bytes = Buffer.byteLength(part); + if (bytes > WEB_SOURCE_MAP_LIMITS.outputBytes - encodedBytes) + return err(capacityFailure()); + encoded.push(part); + encodedBytes += bytes; + } + if (options?.signal?.aborted) + return err(new AnalysisCancelledError(OPERATION)); + const serialized = encoded.join(""); const response = await runSourceMapCommand( serialized, input.path, options, this.dependencies, ); - if (!response.ok) return response; + if (!response.ok) { + if (response.error instanceof SourceMapCleanupFailure) + this.#pendingCleanup = response.error.cleanupOwner; + return response; + } let value: unknown; try { value = JSON.parse(response.value); @@ -92,6 +149,29 @@ export class SourceMapDecoder implements WebSourceMapPort { ); return ok(reply.data.report); } + + async close(): Promise { + this.#closing = true; + this.#closePromise ??= this.#tail + .then(async () => { + const failure = await this.#retryCleanup(); + if (failure !== undefined) throw failure; + }) + .catch((cause: unknown) => { + this.#closePromise = undefined; + throw cause; + }); + return this.#closePromise; + } + + async #retryCleanup(): Promise { + const pending = this.#pendingCleanup; + if (pending === undefined) return undefined; + const failure = await pending.close(null); + if (failure !== undefined) return failure; + this.#pendingCleanup = undefined; + return undefined; + } } const codecFailure = ( path: string, diff --git a/src/process/OwnedCommand.ts b/src/process/OwnedCommand.ts index b5de06b644..33d06c5eab 100644 --- a/src/process/OwnedCommand.ts +++ b/src/process/OwnedCommand.ts @@ -22,6 +22,7 @@ export class OwnedCommandFailure extends Error { readonly cleanupFailure: string | null = null, options?: ErrorOptions, readonly resources: readonly string[] = [], + readonly cleanupOwner?: ProviderProcessSupervisor, ) { super(message, options); } @@ -183,7 +184,6 @@ const collectOwnedCommand = async ( } const stopped = await supervisor.stop(); const snapshot = supervisor.snapshot(); - supervisor.dispose(); if (stopped.status === "incomplete") throw new OwnedCommandFailure( failure?.reason ?? "process", @@ -197,6 +197,7 @@ const collectOwnedCommand = async ( `pid:${launched.ownership.leaderPid}`, `process-group:${launched.ownership.processGroupId}`, ], + supervisor, ); if (failure !== undefined) throw new OwnedCommandFailure(