From 962dfd28110890059f06346d48e7bcc76e02bc1f Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Tue, 11 Aug 2026 22:30:30 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20Settle=20process=20join()=20only?= =?UTF-8?q?=20after=20stdio=20pumps=20complete?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Process.join() could resolve on the child-process close event while the stdout/stderr pump tasks were still forwarding the final chunks through Stdio middleware, so callers could observe an exit status before all output was delivered. - posix: wait for both pump done-resolvers after the close event before resolving processResult, matching the win32 adapter's existing drain-before-result ordering - both adapters: a failing Stdio handler now resolves processResult with Err so join()/expect() throw instead of hanging, and the pumps always resolve their done-resolvers and close their signals on the way out - add an internal CloseEvent context so tests can order assertions deterministically around the close event without scheduler sleeps Fixes #244 --- process/src/exec/internal.ts | 12 ++++ process/src/exec/posix.ts | 49 ++++++++++---- process/src/exec/win32.ts | 45 +++++++++---- process/test/exec.test.ts | 122 +++++++++++++++++++++++++++++++++++ 4 files changed, 200 insertions(+), 28 deletions(-) create mode 100644 process/src/exec/internal.ts diff --git a/process/src/exec/internal.ts b/process/src/exec/internal.ts new file mode 100644 index 00000000..35352b59 --- /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 "close" event is + * received, before waiting on the output pumps. 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 6b5bb9bc..bb93ca46 100644 --- a/process/src/exec/posix.ts +++ b/process/src/exec/posix.ts @@ -24,6 +24,7 @@ import type { } from "./types.ts"; import { Stdio } from "../api.ts"; import { ExecError } from "./error.ts"; +import { CloseEvent } from "./internal.ts"; type ProcessResultValue = [number?, string?]; @@ -69,25 +70,37 @@ export function* createPosixProcess( let stderr = createSignal(); yield* spawn(function* () { - let next = yield* io.stdout.next(); - while (!next.done) { - yield* Stdio.operations.stdout(next.value); - stdout.send(next.value); - next = yield* io.stdout.next(); + try { + let next = yield* io.stdout.next(); + while (!next.done) { + yield* Stdio.operations.stdout(next.value); + stdout.send(next.value); + next = yield* io.stdout.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 next = yield* io.stderr.next(); - while (!next.done) { - yield* Stdio.operations.stderr(next.value); - stderr.send(next.value); - next = yield* io.stderr.next(); + try { + let next = yield* io.stderr.next(); + while (!next.done) { + yield* Stdio.operations.stderr(next.value); + stderr.send(next.value); + next = yield* io.stderr.next(); + } + } catch (error) { + processResult.resolve(Err(error as Error)); + } finally { + stderr.close(); + io.stderrDone.resolve(); } - stderr.close(); - io.stderrDone.resolve(); }); let stdin: Writable = { @@ -103,6 +116,14 @@ export function* createPosixProcess( yield* spawn(function* () { let value = yield* once(childProcess, "close"); + let closeEvent = yield* CloseEvent.get(); + if (closeEvent) { + yield* closeEvent(); + } + // wait for both pumps to finish forwarding buffered output so that + // join() never settles before the final chunks have passed through + // Stdio middleware and the public signals + yield* all([io.stdoutDone.operation, io.stderrDone.operation]); processResult.resolve(Ok(value)); }); diff --git a/process/src/exec/win32.ts b/process/src/exec/win32.ts index 3a2c78b2..414eac2b 100644 --- a/process/src/exec/win32.ts +++ b/process/src/exec/win32.ts @@ -25,6 +25,7 @@ import type { } from "./types.ts"; import { Stdio } from "../api.ts"; import { ExecError } from "./error.ts"; +import { CloseEvent } from "./internal.ts"; import { unbox, useEvalScope } from "@effectionx/scope-eval"; type ProcessResultValue = [number?, string?]; @@ -87,25 +88,37 @@ export function* createWin32Process( const stderr = createSignal(); yield* spawn(function* () { - let next = yield* io.stdout.next(); - while (!next.done) { - yield* Stdio.operations.stdout(next.value); - stdout.send(next.value); - next = yield* io.stdout.next(); + try { + let next = yield* io.stdout.next(); + while (!next.done) { + yield* Stdio.operations.stdout(next.value); + stdout.send(next.value); + next = yield* io.stdout.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 next = yield* io.stderr.next(); - while (!next.done) { - yield* Stdio.operations.stderr(next.value); - stderr.send(next.value); - next = yield* io.stderr.next(); + try { + let next = yield* io.stderr.next(); + while (!next.done) { + yield* Stdio.operations.stderr(next.value); + stderr.send(next.value); + next = yield* io.stderr.next(); + } + } catch (error) { + processResult.resolve(Err(error as Error)); + } finally { + stderr.close(); + io.stderrDone.resolve(); } - stderr.close(); - io.stderrDone.resolve(); }); let stdin: Writable = { @@ -121,6 +134,10 @@ export function* createWin32Process( yield* spawn(function* () { let value = yield* once(childProcess, "close"); + 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..0ae8733b 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 + // stdout pump 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[] = [];