diff --git a/src/process/DarwinProcessRunTokenReader.ts b/src/process/DarwinProcessRunTokenReader.ts index d8dce5bb37..a3769866bd 100644 --- a/src/process/DarwinProcessRunTokenReader.ts +++ b/src/process/DarwinProcessRunTokenReader.ts @@ -111,11 +111,16 @@ const STDERR_EXCERPT_LIMIT = 512; /** The native Darwin process-ownership reader could not be prepared. */ export class DarwinProcessOwnershipInspectionError extends Error { + readonly #cleanupOwner: { close(): Promise } | undefined; + get cleanupOwner(): { close(): Promise } | undefined { + return this.#cleanupOwner; + } constructor( message: string, options?: ErrorOptions & { readonly compilerOutcome?: DarwinCompilerOutcome; readonly capturedOutput?: AnalysisCapturedOutput; + readonly cleanupOwner?: { close(): Promise } | undefined; readonly cleanupFailure?: AnalysisError; readonly cancellation?: { readonly reason: unknown }; }, @@ -124,6 +129,7 @@ export class DarwinProcessOwnershipInspectionError extends Error { compilerOutcome, capturedOutput, cleanupFailure, + cleanupOwner, cancellation, ...errorOptions } = options ?? {}; @@ -131,6 +137,7 @@ export class DarwinProcessOwnershipInspectionError extends Error { this.name = "DarwinProcessOwnershipInspectionError"; this.compilerOutcome = compilerOutcome; this.capturedOutput = capturedOutput; + this.#cleanupOwner = cleanupOwner; this.cleanupFailure = cleanupFailure; this.cancellation = cancellation; } @@ -143,6 +150,7 @@ export class DarwinProcessOwnershipInspectionError extends Error { /** Preserve the physical preparation failure while following the exact cleanup owner. */ withCleanupFailure( cleanup: AnalysisError, + owner?: { close(): Promise }, ): DarwinProcessOwnershipInspectionError { const combined = this.cleanupFailure === undefined || this.cleanupFailure === cleanup @@ -160,6 +168,7 @@ export class DarwinProcessOwnershipInspectionError extends Error { ? {} : { capturedOutput: this.capturedOutput }), cleanupFailure: combined, + cleanupOwner: owner ?? this.cleanupOwner, ...(this.cancellation === undefined ? {} : { cancellation: this.cancellation }), @@ -504,7 +513,9 @@ export const createDarwinProcessRunTokenReader = ( "Darwin process ownership helper preparation failed", { cause }, ); - throw primary.withCleanupFailure(cleanup); + throw primary.withCleanupFailure(cleanup, { + close: () => cleanupRoot(operationRoot), + }); } } throw cause; @@ -578,6 +589,7 @@ export const createDarwinProcessRunTokenReader = ( cause: waited.reason, cancellation: { reason: waited.reason }, cleanupFailure: finalized.error.cleanupFailure, + cleanupOwner: finalized.error.cleanupOwner, }, ); } diff --git a/src/process/OwnedCommand.ts b/src/process/OwnedCommand.ts index 06f59a3030..6247ce26ff 100644 --- a/src/process/OwnedCommand.ts +++ b/src/process/OwnedCommand.ts @@ -31,6 +31,7 @@ export class OwnedCommandFailure extends Error { get cancellation(): { readonly reason: unknown } | undefined { return this.#cancellation; } + readonly processError: Error | undefined; readonly cleanupResult: ProcessCleanupFailure | undefined; readonly collectionFailure: boolean; readonly preparationFailure: @@ -44,6 +45,7 @@ export class OwnedCommandFailure extends Error { readonly cleanupFailure: string | null = null, options?: ErrorOptions & { readonly cancellation?: { readonly reason: unknown } | undefined; + readonly processError?: Error | undefined; readonly cleanupResult?: ProcessCleanupFailure; readonly collectionFailure?: boolean; readonly preparationFailure?: DarwinProcessOwnershipInspectionError; @@ -56,6 +58,7 @@ export class OwnedCommandFailure extends Error { options?.cancellation === undefined ? undefined : { reason: options.cancellation.reason }; + this.processError = options?.processError; this.cleanupResult = options?.cleanupResult; this.collectionFailure = options?.collectionFailure ?? false; this.preparationFailure = options?.preparationFailure; @@ -79,6 +82,8 @@ interface OwnedCommandLimits { readonly diagnosticBytes: number; /** Accept an ordinary nonzero exit after output closes without other process errors. */ readonly acceptNonZeroExit?: boolean; + /** Interpret complete physical output synchronously before awaited final cleanup. */ + readonly observeCompleted?: (snapshot: ProviderProcessSnapshot) => void; /** * Receives stdout instead of the retained snapshot, for output parsed as it * arrives. The command completes only after the sink finishes. The sink @@ -184,6 +189,7 @@ const collectOwnedCommand = async ( } let exceeded = false; let processFailure: string | undefined; + let processError: Error | undefined; const supervisor = new ProviderProcessSupervisor( { ...launched, @@ -198,7 +204,10 @@ const collectOwnedCommand = async ( onDiagnostic: (event) => { if (event.type === "output" && event.truncated === true) exceeded = true; - if (event.type === "error") processFailure = event.message; + if (event.type === "error") { + processFailure ??= event.message; + processError ??= event.cause; + } }, }, ); @@ -235,6 +244,7 @@ const collectOwnedCommand = async ( ); if (processFailure !== undefined) throw new OwnedCommandFailure("process", processFailure, null, null, { + processError, collectionFailure: true, }); await waitForAbortableDelay(25, signal); @@ -279,8 +289,9 @@ const collectOwnedCommand = async ( "Command failed; inspect its exit status and diagnostics.", snapshot, null, - { collectionFailure: processFailure !== undefined }, + { processError, collectionFailure: processFailure !== undefined }, ); + limits.observeCompleted?.(snapshot); } catch (cause: unknown) { failure = cause instanceof OwnedCommandFailure @@ -297,6 +308,9 @@ const collectOwnedCommand = async ( abandoned.abort(); const stopped = await supervisor.stop(); const snapshot = supervisor.snapshot(); + const cancellation = + failure?.cancellation ?? + (callerSignal?.aborted ? { reason: callerSignal.reason } : undefined); if (stopped.status === "incomplete" && retainCleanup !== undefined) retainCleanup(supervisor, stopped); else if (stopped.status === "incomplete") @@ -306,9 +320,10 @@ const collectOwnedCommand = async ( snapshot, stopped.reason, { + cancellation, ...(failure === undefined ? {} - : { cause: failure, cancellation: failure.cancellation }), + : { cause: failure, processError: failure.processError }), ...(stopped.cleanupResult === undefined ? {} : { cleanupResult: stopped.cleanupResult }), @@ -327,7 +342,8 @@ const collectOwnedCommand = async ( failure.cleanupFailure, { cause: failure, - cancellation: failure.cancellation, + cancellation, + processError: failure.processError, ...(failure.cleanupResult === undefined ? {} : { cleanupResult: failure.cleanupResult }), diff --git a/src/process/ProviderProcess.ts b/src/process/ProviderProcess.ts index 6f42dab6fc..c74b604b5c 100644 --- a/src/process/ProviderProcess.ts +++ b/src/process/ProviderProcess.ts @@ -163,7 +163,11 @@ export type ProviderProcessDiagnostic = readonly signal: NodeJS.Signals | null; readonly snapshot: ProviderProcessSnapshot; } - | { readonly type: "error"; readonly message: string }; + | { + readonly type: "error"; + readonly message: string; + readonly cause?: Error; + }; /** Options for process lifecycle diagnostics. */ export interface ProviderProcessSupervisorOptions { @@ -329,7 +333,11 @@ export class ProviderProcessSupervisor { this.#outputClose.resolve(undefined); }; readonly #onError = (cause: Error): void => { - this.#options.onDiagnostic?.({ type: "error", message: cause.message }); + this.#options.onDiagnostic?.({ + type: "error", + message: cause.message, + cause, + }); }; #exitObservation: ProviderProcessExitObservation | undefined; #stopPromise: Promise | undefined; @@ -656,6 +664,7 @@ export class ProviderProcessSupervisor { this.#options.onDiagnostic?.({ type: "error", message: `${stream} stream failed: ${cause.message}`, + cause, }); } diff --git a/tests/boundary/process/darwinProcessRunTokenReader.test.ts b/tests/boundary/process/darwinProcessRunTokenReader.test.ts index cd8df24369..ebae535d21 100644 --- a/tests/boundary/process/darwinProcessRunTokenReader.test.ts +++ b/tests/boundary/process/darwinProcessRunTokenReader.test.ts @@ -829,9 +829,11 @@ it.skipIf(process.platform === "win32").each(["first", "second"] as const)( }, ); -it.skipIf(process.platform === "win32" || process.getuid?.() === 0)( - "retains a failed compilation cleanup for a later close retry", - async () => { +it + .skipIf(process.platform === "win32" || process.getuid?.() === 0) + .each(["reader-close", "exact-root"] as const)( + "retains a failed compilation cleanup for a later %s retry", + async (retry) => { const directory = await mkdtemp( join(tmpdir(), "rea-process-token-cleanup-retry-test-"), ); @@ -854,10 +856,31 @@ it.skipIf(process.platform === "win32" || process.getuid?.() === 0)( ); await chmod(state.root, 0o000); controller.abort(); - expect((await preparation).state).toBe("rejected"); - await expect(reader.close()).rejects.toThrow(); - await chmod(state.root, 0o700); - await reader.close(); + const outcome = await preparation; + expect(outcome.state).toBe("rejected"); + if (retry === "reader-close") { + await expect(reader.close()).rejects.toThrow(); + await chmod(state.root, 0o700); + await reader.close(); + } else { + if ( + outcome.state !== "rejected" || + !(outcome.cause instanceof DarwinProcessOwnershipInspectionError) + ) + throw new Error("Expected owned compilation cleanup failure"); + const failure = outcome.cause; + expect(failure.cleanupOwner).toBeDefined(); + expect(failure.cleanupFailure?.cleanupIncomplete).toBe(true); + await expect(failure.cleanupOwner?.close()).rejects.toThrow(); + await chmod(state.root, 0o700); + await failure.cleanupOwner?.close(); + expect(failure.cleanupFailure?.cleanupIncomplete).toBe(false); + // Releasing the failed root does not close the shared reader itself. + const stopped = new AbortController(); + const reason = { request: "probe reader remains open" }; + stopped.abort(reason); + await expect(reader.prepare(stopped.signal)).rejects.toBe(reason); + } await expect(access(state.root)).rejects.toThrow(); } finally { if (!controller.signal.aborted) controller.abort(); diff --git a/tests/boundary/process/ownedCommand.test.ts b/tests/boundary/process/ownedCommand.test.ts index cb4a326f81..1b1846a3b5 100644 --- a/tests/boundary/process/ownedCommand.test.ts +++ b/tests/boundary/process/ownedCommand.test.ts @@ -103,6 +103,99 @@ it("collects a real owned command and verifies its exit before returning diagnos }); }); +it.each([0, 2])( + "observes complete accepted output once before final cleanup (exit %s)", + async (exitCode) => { + const stages: string[] = []; + const snapshot = await runOwnedCommand( + command( + `process.stdout.write("complete"); process.exitCode = ${exitCode}`, + ), + { + timeoutMs: 2000, + diagnosticBytes: 1024, + acceptNonZeroExit: true, + observeCompleted: (observed) => { + stages.push("observed"); + expect(observed.exitCode).toBe(exitCode); + expect(observed.stdout.text).toBe("complete"); + }, + }, + { + launcher: async (input) => { + const acquired = await spawnOwnedProviderProcess(input); + return { + ...acquired, + cleanup: async () => { + stages.push("cleanup"); + return cleanupOwnedProcessGroup(acquired.ownership); + }, + }; + }, + }, + ); + expect(snapshot.exitCode).toBe(exitCode); + expect(stages).toEqual(["observed", "cleanup"]); + }, +); + +it("does not interpret output rejected by its complete-output budget", async () => { + const observeCompleted = vi.fn(); + await expect( + runOwnedCommand(command('process.stdout.write("x".repeat(2048))'), { + timeoutMs: 2000, + diagnosticBytes: 1024, + observeCompleted, + }), + ).rejects.toMatchObject({ reason: "output-limit" }); + expect(observeCompleted).not.toHaveBeenCalled(); +}); + +it("retains late cancellation separately from a physical failure with refused cleanup", async () => { + const controller = new AbortController(); + const callerReason = { request: "cancel during refused finalization" }; + let denied = true; + let failure: OwnedCommandFailure | undefined; + try { + await runOwnedCommand( + command("process.exit(2)"), + { timeoutMs: 2000, diagnosticBytes: 1024 }, + { + signal: controller.signal, + launcher: async (input) => { + const acquired = await spawnOwnedProviderProcess(input); + return { + ...acquired, + cleanup: async () => { + if (denied) { + controller.abort(callerReason); + return { cleaned: false, reason: "controlled cleanup refusal" }; + } + return cleanupOwnedProcessGroup(acquired.ownership); + }, + }; + }, + }, + ).catch((cause: unknown) => { + if (!(cause instanceof OwnedCommandFailure)) throw cause; + failure = cause; + }); + expect(failure).toMatchObject({ + reason: "process", + snapshot: { exitCode: 2 }, + cleanupFailure: "controlled cleanup refusal", + }); + expect(failure?.cancellation?.reason).toBe(callerReason); + denied = false; + expect(await failure?.cleanupOwner?.stop()).toEqual({ + status: "verified-cleanup", + }); + } finally { + denied = false; + await failure?.cleanupOwner?.stop(); + } +}); + it("streams stdout beyond the diagnostic budget into a sink and returns after it finishes", async () => { let received = 0; let finished = false; @@ -175,6 +268,9 @@ it("reports a failed stdout sink as a collection failure", async () => { it("fails an accepted nonzero command when an output stream errors and cleans it up", async () => { const stdout = new PassThrough(); const stderr = new PassThrough(); + const readFailure = Object.assign(new Error("fixture read failure"), { + code: "EIO", + }); class FakeProcess extends EventEmitter { readonly stdout = stdout; readonly stderr = stderr; @@ -199,7 +295,7 @@ it("fails an accepted nonzero command when an output stream errors and cleans it launcher: async () => { setImmediate(() => { stdout.write("partial output"); - stdout.destroy(new Error("fixture read failure")); + stdout.destroy(readFailure); stderr.end(); child.exitCode = 7; child.emit("exit", 7, null); @@ -223,6 +319,7 @@ it("fails an accepted nonzero command when an output stream errors and cleans it message: "stdout stream failed: fixture read failure", snapshot: { stdout: { text: "partial output" }, exitCode: 7 }, }); + await expect(failure).rejects.toHaveProperty("processError", readFailure); expect(cleanup).toHaveBeenCalledOnce(); }); @@ -286,6 +383,9 @@ it.skipIf(process.platform === "win32")( hostEnvironment: process.env, }); let descendantPid: number | undefined; + let descendantOwner: + | Awaited> + | undefined; try { const result = await runOwnedCommand( { @@ -302,6 +402,12 @@ it.skipIf(process.platform === "win32")( sweepTokenOwnedProcesses: true, }, { timeoutMs: 5_000, diagnosticBytes: 1024 }, + { + launcher: async (input) => { + descendantOwner = await spawnOwnedProviderProcess(input); + return descendantOwner; + }, + }, ); descendantPid = Number(result.stdout.text); expect(Number.isSafeInteger(descendantPid)).toBe(true); @@ -316,13 +422,13 @@ it.skipIf(process.platform === "win32")( expect(remainsLive).toBe(false); expect(() => process.kill(unrelated.process.pid!, 0)).not.toThrow(); } finally { - if (descendantPid !== undefined) { - try { - process.kill(descendantPid, "SIGKILL"); - } catch { - // The run-token sweep may already have removed it. - } - } + if (descendantOwner !== undefined) + expect( + ( + await (descendantOwner.cleanup?.() ?? + cleanupOwnedProcessGroup(descendantOwner.ownership)) + ).cleaned, + ).toBe(true); await cleanupOwnedProcessGroup(unrelated.ownership); } },