From f2964df03155fa4570f6e8ac5e80a6100948d9a0 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Tue, 11 Aug 2026 23:44:17 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20Settle=20join()=20only=20after?= =?UTF-8?q?=20output=20clears=20Stdio=20middleware?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On POSIX, join()/expect() resolved directly from the useNativeProcess result, which settles on the child-process close event while the Stdio middleware tasks may still be forwarding the final chunks. Callers could observe an exit status before all output had passed through their middleware and the public signals. - posix gains the same drain-then-resolve task win32 already had: processResult settles only after the close result plus both middleware tasks completing, so a blocked handler keeps join() pending - both platforms: a failing Stdio handler resolves processResult with Err, so join()/expect() throw the handler's error instead of crashing the scope or hanging the drain, and the middleware tasks always close their signals and resolve their done-resolvers on the way out - an internal CloseEvent context (not exported from mod.ts) lets the regression tests order assertions deterministically around the close event without scheduler sleeps Fixes #244 --- process/src/exec/internal.ts | 12 ++++ process/src/exec/posix.ts | 61 +++++++++++++----- process/src/exec/win32.ts | 46 +++++++++---- process/test/exec.test.ts | 122 +++++++++++++++++++++++++++++++++++ 4 files changed, 211 insertions(+), 30 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..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[] = [];