diff --git a/process/src/exec/internal.ts b/process/src/exec/internal.ts new file mode 100644 index 00000000..9cda66a7 --- /dev/null +++ b/process/src/exec/internal.ts @@ -0,0 +1,12 @@ +import { type Operation, createContext } from "effection"; + +/** + * Internal test seam, deliberately not exported from mod.ts. When set, the + * adapters run this operation as soon as the child process result resolves, + * before waiting on the Stdio middleware tasks. This lets tests order their + * assertions deterministically around the close event instead of relying on + * scheduler timing. + */ +export const CloseEvent = createContext<() => Operation>( + "@effectionx/process:close-event", +); diff --git a/process/src/exec/posix.ts b/process/src/exec/posix.ts index 13897c2f..d728a04b 100644 --- a/process/src/exec/posix.ts +++ b/process/src/exec/posix.ts @@ -2,7 +2,9 @@ import { spawn as spawnProcess } from "node:child_process"; import process from "node:process"; import { type Operation, + type Result, type Yielded, + Err, all, createSignal, ensure, @@ -19,7 +21,8 @@ import type { } from "./types.ts"; import { Stdio } from "../api.ts"; import { ExecError } from "./error.ts"; -import { useNativeProcess } from "./native.ts"; +import { type ProcessResultValue, useNativeProcess } from "./native.ts"; +import { CloseEvent } from "./internal.ts"; export function* createPosixProcess( command: string, @@ -50,6 +53,8 @@ export function* createPosixProcess( let { childProcess } = os; let { pid } = childProcess; + let processResult = withResolvers>(); + let io = { stdoutDone: withResolvers(), stderrDone: withResolvers(), @@ -60,26 +65,38 @@ export function* createPosixProcess( yield* spawn(function* () { let subscription = yield* os.stdout; - let next = yield* subscription.next(); - while (!next.done) { - yield* Stdio.operations.stdout(next.value); - stdout.send(next.value); - next = yield* subscription.next(); + try { + let next = yield* subscription.next(); + while (!next.done) { + yield* Stdio.operations.stdout(next.value); + stdout.send(next.value); + next = yield* subscription.next(); + } + } catch (error) { + // deliver Stdio middleware failures through join()/expect() so a + // broken handler cannot leave callers waiting on processResult + processResult.resolve(Err(error as Error)); + } finally { + stdout.close(); + io.stdoutDone.resolve(); } - stdout.close(); - io.stdoutDone.resolve(); }); yield* spawn(function* () { let subscription = yield* os.stderr; - let next = yield* subscription.next(); - while (!next.done) { - yield* Stdio.operations.stderr(next.value); - stderr.send(next.value); - next = yield* subscription.next(); + try { + let next = yield* subscription.next(); + while (!next.done) { + yield* Stdio.operations.stderr(next.value); + stderr.send(next.value); + next = yield* subscription.next(); + } + } catch (error) { + processResult.resolve(Err(error as Error)); + } finally { + stderr.close(); + io.stderrDone.resolve(); } - stderr.close(); - io.stderrDone.resolve(); }); let stdin: Writable = { @@ -88,8 +105,20 @@ export function* createPosixProcess( }, }; + yield* spawn(function* () { + let value = yield* os.result; + let closeEvent = yield* CloseEvent.get(); + if (closeEvent) { + yield* closeEvent(); + } + // join() settles only after every chunk has also cleared Stdio + // middleware and the public signals + yield* all([io.stdoutDone.operation, io.stderrDone.operation]); + processResult.resolve(value); + }); + function* join() { - let result = yield* os.result; + let result = yield* processResult.operation; if (result.ok) { let [code, signal] = result.value; return { command, options, code, signal } as ExitStatus; diff --git a/process/src/exec/win32.ts b/process/src/exec/win32.ts index 1a0bb24d..ccf3d7c3 100644 --- a/process/src/exec/win32.ts +++ b/process/src/exec/win32.ts @@ -7,6 +7,7 @@ import { type Operation, type Result, type Yielded, + Err, all, createSignal, ensure, @@ -23,6 +24,7 @@ import type { import { Stdio } from "../api.ts"; import { ExecError } from "./error.ts"; import { type ProcessResultValue, useNativeProcess } from "./native.ts"; +import { CloseEvent } from "./internal.ts"; import { unbox, useEvalScope } from "@effectionx/scope-eval"; function* killTree(pid: number) { @@ -82,26 +84,38 @@ export function* createWin32Process( yield* spawn(function* () { let subscription = yield* os.stdout; - let next = yield* subscription.next(); - while (!next.done) { - yield* Stdio.operations.stdout(next.value); - stdout.send(next.value); - next = yield* subscription.next(); + try { + let next = yield* subscription.next(); + while (!next.done) { + yield* Stdio.operations.stdout(next.value); + stdout.send(next.value); + next = yield* subscription.next(); + } + } catch (error) { + // deliver Stdio middleware failures through join()/expect() so a + // broken handler cannot leave callers waiting on processResult + processResult.resolve(Err(error as Error)); + } finally { + stdout.close(); + io.stdoutDone.resolve(); } - stdout.close(); - io.stdoutDone.resolve(); }); yield* spawn(function* () { let subscription = yield* os.stderr; - let next = yield* subscription.next(); - while (!next.done) { - yield* Stdio.operations.stderr(next.value); - stderr.send(next.value); - next = yield* subscription.next(); + try { + let next = yield* subscription.next(); + while (!next.done) { + yield* Stdio.operations.stderr(next.value); + stderr.send(next.value); + next = yield* subscription.next(); + } + } catch (error) { + processResult.resolve(Err(error as Error)); + } finally { + stderr.close(); + io.stderrDone.resolve(); } - stderr.close(); - io.stderrDone.resolve(); }); let stdin: Writable = { @@ -112,6 +126,10 @@ export function* createWin32Process( yield* spawn(function* () { let value = yield* os.result; + let closeEvent = yield* CloseEvent.get(); + if (closeEvent) { + yield* closeEvent(); + } // out of band with the finally block below compared to posix as // win32 is more sensitive to graceful shutdown timing that it is // worth waiting for stdout and stderr to close before resolving the process result diff --git a/process/test/exec.test.ts b/process/test/exec.test.ts index 3b6908fa..3ded925e 100644 --- a/process/test/exec.test.ts +++ b/process/test/exec.test.ts @@ -8,6 +8,7 @@ import { captureError, expectMatch, fetchText } from "./helpers.ts"; import { lines } from "@effectionx/stream-helpers"; import { type Process, type ProcessResult, exec } from "../mod.ts"; import { Stdio } from "../src/api.ts"; +import { CloseEvent } from "../src/exec/internal.ts"; const SystemRoot = process.env.SystemRoot; @@ -299,6 +300,127 @@ describe("exec", () => { }); }); + describe("output completeness", () => { + // Gate every stdout chunk behind `release`, and only resolve `release` + // once the adapter has received the child-process close event (observed + // through the internal CloseEvent seam). join() is therefore settling + // strictly after the close event, and must still have the full gated + // output forwarded through Stdio middleware by the time it settles. + it("delivers all stdout through Stdio middleware before join() settles", function* () { + const closed = withResolvers(); + const release = withResolvers(); + const procReady = withResolvers(); + const observed: Uint8Array[] = []; + let stdoutAtSettle: string | undefined; + + yield* CloseEvent.set(function* () { + closed.resolve(); + }); + + yield* Stdio.around({ + *stdout([bytes]) { + yield* release.operation; + observed.push(bytes); + }, + }); + + // spawned before exec so that whenever this task and the adapter's + // middleware task are both runnable, this one resumes first and + // captures what had been forwarded at the moment join() settled + const joined = yield* spawn(function* () { + const proc = yield* procReady.operation; + const status = yield* proc.join(); + stdoutAtSettle = Buffer.concat(observed) + .toString("utf8") + .replace(/\r\n/g, "\n"); + return status; + }); + + const proc = yield* exec("node './fixtures/hello-world.js'", { + cwd: import.meta.dirname, + }); + procReady.resolve(proc); + + // the child process has closed, but the gate is still held + yield* closed.operation; + release.resolve(); + + const status = yield* joined; + expect(status.code).toEqual(0); + expect(stdoutAtSettle).toEqual("hello\nworld\n"); + }); + + it("delivers all stderr through Stdio middleware before join() settles", function* () { + const closed = withResolvers(); + const release = withResolvers(); + const procReady = withResolvers(); + const observed: Uint8Array[] = []; + let stderrAtSettle: string | undefined; + + yield* CloseEvent.set(function* () { + closed.resolve(); + }); + + yield* Stdio.around({ + *stderr([bytes]) { + yield* release.operation; + observed.push(bytes); + }, + }); + + const joined = yield* spawn(function* () { + const proc = yield* procReady.operation; + const status = yield* proc.join(); + stderrAtSettle = Buffer.concat(observed) + .toString("utf8") + .replace(/\r\n/g, "\n"); + return status; + }); + + const proc = yield* exec("node './fixtures/hello-world.js'", { + cwd: import.meta.dirname, + }); + procReady.resolve(proc); + + yield* closed.operation; + release.resolve(); + + const status = yield* joined; + expect(status.code).toEqual(0); + expect(stderrAtSettle).toContain("boom\n"); + }); + + it("fails join() instead of hanging when a Stdio handler fails", function* () { + yield* Stdio.around({ + *stdout() { + throw new Error("stdout handler exploded"); + }, + }); + + const proc = yield* exec("node './fixtures/hello-world.js'", { + cwd: import.meta.dirname, + }); + + const error = yield* captureError(proc.join()); + expect(error.message).toEqual("stdout handler exploded"); + }); + + it("fails expect() instead of hanging when a Stdio handler fails", function* () { + yield* Stdio.around({ + *stderr() { + throw new Error("stderr handler exploded"); + }, + }); + + const proc = yield* exec("node './fixtures/hello-world.js'", { + cwd: import.meta.dirname, + }); + + const error = yield* captureError(proc.expect()); + expect(error.message).toEqual("stderr handler exploded"); + }); + }); + describe("io api", () => { it("allows redirecting stdio to array", function* () { let outputStdout: Uint8Array[] = [];