From a52a9a075137ff16a8b030476ecf941cdea0c42d Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Tue, 11 Aug 2026 23:20:02 -0400 Subject: [PATCH 1/4] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Feed=20process=20stdio?= =?UTF-8?q?=20from=20native=20listeners=20instead=20of=20readable=20pumps?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Attach data/close listeners to the child's stdio streams and error/close listeners to the child process itself, all in the same synchronous continuation as the spawn, and push chunks straight into signals. The Stdio middleware tasks now consume those signals, which removes the fromReadable subscriptions and the separate error/close watcher tasks. No behavior change: join()/expect() settle exactly as before on each platform, teardown ordering is untouched, and the public API is unchanged. --- process/src/exec/posix.ts | 47 ++++++++++++++++++++++++--------------- process/src/exec/win32.ts | 44 +++++++++++++++++++++++++----------- 2 files changed, 60 insertions(+), 31 deletions(-) diff --git a/process/src/exec/posix.ts b/process/src/exec/posix.ts index 6b5bb9bc..e3aa82f2 100644 --- a/process/src/exec/posix.ts +++ b/process/src/exec/posix.ts @@ -13,8 +13,6 @@ import { withResolvers, } from "effection"; import { unbox, useEvalScope } from "@effectionx/scope-eval"; -import { once } from "@effectionx/node/events"; -import { fromReadable } from "@effectionx/node/stream"; import type { CreateOSProcess, ExecOptions, @@ -58,9 +56,30 @@ export function* createPosixProcess( throw new Error("stdout and stderr must be available with stdio: pipe"); } + // Native listeners attached in the same synchronous continuation as the + // spawn: process events and stdio chunks cannot be missed, and no + // readable pumps are needed. + childProcess.once("error", (error) => { + processResult.resolve(Err(error)); + }); + childProcess.once("close", (code, signal) => { + processResult.resolve(Ok([code ?? undefined, signal ?? undefined])); + }); + + let raw = { + stdout: createSignal(), + stderr: createSignal(), + }; + childProcess.stdout.on("data", (chunk: Uint8Array) => + raw.stdout.send(chunk), + ); + childProcess.stdout.once("close", () => raw.stdout.close()); + childProcess.stderr.on("data", (chunk: Uint8Array) => + raw.stderr.send(chunk), + ); + childProcess.stderr.once("close", () => raw.stderr.close()); + let io = { - stdout: yield* fromReadable(childProcess.stdout), - stderr: yield* fromReadable(childProcess.stderr), stdoutDone: withResolvers(), stderrDone: withResolvers(), }; @@ -69,22 +88,24 @@ export function* createPosixProcess( let stderr = createSignal(); yield* spawn(function* () { - let next = yield* io.stdout.next(); + let subscription = yield* raw.stdout; + let next = yield* subscription.next(); while (!next.done) { yield* Stdio.operations.stdout(next.value); stdout.send(next.value); - next = yield* io.stdout.next(); + next = yield* subscription.next(); } stdout.close(); io.stdoutDone.resolve(); }); yield* spawn(function* () { - let next = yield* io.stderr.next(); + let subscription = yield* raw.stderr; + let next = yield* subscription.next(); while (!next.done) { yield* Stdio.operations.stderr(next.value); stderr.send(next.value); - next = yield* io.stderr.next(); + next = yield* subscription.next(); } stderr.close(); io.stderrDone.resolve(); @@ -96,16 +117,6 @@ export function* createPosixProcess( }, }; - yield* spawn(function* trapError() { - let [error] = yield* once<[Error]>(childProcess, "error"); - processResult.resolve(Err(error)); - }); - - yield* spawn(function* () { - let value = yield* once(childProcess, "close"); - processResult.resolve(Ok(value)); - }); - function* join() { let result = yield* processResult.operation; if (result.ok) { diff --git a/process/src/exec/win32.ts b/process/src/exec/win32.ts index 3a2c78b2..07968bc9 100644 --- a/process/src/exec/win32.ts +++ b/process/src/exec/win32.ts @@ -1,6 +1,5 @@ import { platform } from "node:os"; import { once } from "@effectionx/node/events"; -import { fromReadable } from "@effectionx/node/stream"; // @ts-types="npm:@types/cross-spawn@6.0.6" import { spawn as spawnProcess } from "cross-spawn"; import { ctrlc } from "ctrlc-windows"; @@ -76,9 +75,31 @@ export function* createWin32Process( throw new Error("stdout and stderr must be available with stdio: pipe"); } + // Native listeners attached in the same synchronous continuation as the + // spawn: process events and stdio chunks cannot be missed, and no + // readable pumps are needed. + let rawClose = withResolvers(); + childProcess.once("error", (error) => { + processResult.resolve(Err(error)); + }); + childProcess.once("close", (code, signal) => { + rawClose.resolve([code ?? undefined, signal ?? undefined]); + }); + + let raw = { + stdout: createSignal(), + stderr: createSignal(), + }; + childProcess.stdout.on("data", (chunk: Uint8Array) => + raw.stdout.send(chunk), + ); + childProcess.stdout.once("close", () => raw.stdout.close()); + childProcess.stderr.on("data", (chunk: Uint8Array) => + raw.stderr.send(chunk), + ); + childProcess.stderr.once("close", () => raw.stderr.close()); + let io = { - stdout: yield* fromReadable(childProcess.stdout), - stderr: yield* fromReadable(childProcess.stderr), stdoutDone: withResolvers(), stderrDone: withResolvers(), }; @@ -87,22 +108,24 @@ export function* createWin32Process( const stderr = createSignal(); yield* spawn(function* () { - let next = yield* io.stdout.next(); + let subscription = yield* raw.stdout; + let next = yield* subscription.next(); while (!next.done) { yield* Stdio.operations.stdout(next.value); stdout.send(next.value); - next = yield* io.stdout.next(); + next = yield* subscription.next(); } stdout.close(); io.stdoutDone.resolve(); }); yield* spawn(function* () { - let next = yield* io.stderr.next(); + let subscription = yield* raw.stderr; + let next = yield* subscription.next(); while (!next.done) { yield* Stdio.operations.stderr(next.value); stderr.send(next.value); - next = yield* io.stderr.next(); + next = yield* subscription.next(); } stderr.close(); io.stderrDone.resolve(); @@ -114,13 +137,8 @@ export function* createWin32Process( }, }; - yield* spawn(function* trapError() { - const [error] = yield* once(childProcess, "error"); - processResult.resolve(Err(error)); - }); - yield* spawn(function* () { - let value = yield* once(childProcess, "close"); + let value = yield* rawClose.operation; // 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 From 26eb47e28422846940548c9c2d9b773d7355d4a0 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Tue, 11 Aug 2026 23:27:18 -0400 Subject: [PATCH 2/4] =?UTF-8?q?=E2=9C=A8=20Extract=20useNativeProcess,=20a?= =?UTF-8?q?=20child-process=20resource=20with=20standard=20cleanup?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both adapters now acquire their child through useNativeProcess: a resource that spawns via a caller-provided factory, wires error/close and stdio entirely through native listeners attached in the same synchronous continuation as the spawn, and performs standard cleanup on scope exit — SIGTERM via child.kill() if the process is still alive, then hold the scope until the OS has reaped the process and closed its pipes. The teardown is registered before the child exists, so a halt can never find a live process without it armed. Platform-specific teardown stays in each adapter's own ensure, which runs before the resource backstop: posix keeps the process-group SIGTERM and middleware drain; win32 keeps ctrlc, the Y-newline stdin answer, the drain, and the killTree escalation. The adapters lose their inline listener wiring and, on posix, the processResult plumbing: join() reads the resource result directly. --- process/src/exec/native.ts | 86 +++++++++++++++++++++++++++++++ process/src/exec/posix.ts | 59 ++++++--------------- process/src/exec/win32.ts | 103 ++++++++++++++----------------------- 3 files changed, 140 insertions(+), 108 deletions(-) create mode 100644 process/src/exec/native.ts diff --git a/process/src/exec/native.ts b/process/src/exec/native.ts new file mode 100644 index 00000000..133d4eb5 --- /dev/null +++ b/process/src/exec/native.ts @@ -0,0 +1,86 @@ +import type { ChildProcess } from "node:child_process"; +import { + type Operation, + type Result, + type Stream, + Err, + Ok, + createSignal, + ensure, + resource, + withResolvers, +} from "effection"; + +export type ProcessResultValue = [number?, string?]; + +/** + * A child process as a resource, wired entirely through native Node event + * listeners: exit through the `close` event, output pushed straight into + * signals. All listeners are attached in the same synchronous continuation + * as the spawn, so no event can be missed. + * + * Standard cleanup: if the process is still alive when the scope exits it is + * sent SIGTERM with `child.kill()`, and the scope is held open until the OS + * has reaped the process and closed its stdio pipes. Platform-specific + * termination (process groups, ctrlc, force kill) belongs to the caller's + * own teardown, which runs before this backstop. + */ +export interface NativeProcess { + childProcess: ChildProcess; + stdout: Stream; + stderr: Stream; + + /** The `close` event value, or the error that prevented the spawn. */ + result: Operation>; +} + +export function useNativeProcess( + create: () => ChildProcess, +): Operation { + return resource(function* (provide) { + let result = withResolvers>(); + let stdout = createSignal(); + let stderr = createSignal(); + + let child: ChildProcess | undefined; + + // Registered while `child` is still undefined; the spawn below follows + // in the same synchronous continuation, so the process never exists + // without this teardown armed. + yield* ensure(function* () { + if (child) { + if (child.exitCode === null && child.signalCode === null) { + child.kill("SIGTERM"); + } + // hold the scope until the OS has reaped the process and closed its + // pipes; `result` is a value, so an error outcome cannot throw here + yield* result.operation; + } + }); + + const spawned = create(); + child = spawned; + + spawned.once("error", (error) => { + result.resolve(Err(error)); + }); + spawned.once("close", (code, signal) => { + result.resolve(Ok([code ?? undefined, signal ?? undefined])); + }); + + if (!spawned.stdout || !spawned.stderr) { + throw new Error("stdout and stderr must be available with stdio: pipe"); + } + spawned.stdout.on("data", (chunk: Uint8Array) => stdout.send(chunk)); + spawned.stdout.once("close", () => stdout.close()); + spawned.stderr.on("data", (chunk: Uint8Array) => stderr.send(chunk)); + spawned.stderr.once("close", () => stderr.close()); + + yield* provide({ + childProcess: spawned, + stdout, + stderr, + result: result.operation, + }); + }); +} diff --git a/process/src/exec/posix.ts b/process/src/exec/posix.ts index e3aa82f2..13897c2f 100644 --- a/process/src/exec/posix.ts +++ b/process/src/exec/posix.ts @@ -1,11 +1,8 @@ import { spawn as spawnProcess } from "node:child_process"; import process from "node:process"; import { - type Result, type Operation, type Yielded, - Err, - Ok, all, createSignal, ensure, @@ -22,14 +19,12 @@ import type { } from "./types.ts"; import { Stdio } from "../api.ts"; import { ExecError } from "./error.ts"; - -type ProcessResultValue = [number?, string?]; +import { useNativeProcess } from "./native.ts"; export function* createPosixProcess( command: string, options: ExecOptions, ): Operation { - let processResult = withResolvers>(); const evalScope = yield* useEvalScope(); const result = yield* evalScope.eval(function* () { // Killing all child processes started by this command is surprisingly @@ -42,43 +37,19 @@ export function* createPosixProcess( // process. // // More information here: https://unix.stackexchange.com/questions/14815/process-descendants - let childProcess = spawnProcess(command, options.arguments || [], { - detached: true, - shell: options.shell, - env: options.env, - cwd: options.cwd, - stdio: "pipe", - }); + const os = yield* useNativeProcess(() => + spawnProcess(command, options.arguments || [], { + detached: true, + shell: options.shell, + env: options.env, + cwd: options.cwd, + stdio: "pipe", + }), + ); + let { childProcess } = os; let { pid } = childProcess; - if (!childProcess.stdout || !childProcess.stderr) { - throw new Error("stdout and stderr must be available with stdio: pipe"); - } - - // Native listeners attached in the same synchronous continuation as the - // spawn: process events and stdio chunks cannot be missed, and no - // readable pumps are needed. - childProcess.once("error", (error) => { - processResult.resolve(Err(error)); - }); - childProcess.once("close", (code, signal) => { - processResult.resolve(Ok([code ?? undefined, signal ?? undefined])); - }); - - let raw = { - stdout: createSignal(), - stderr: createSignal(), - }; - childProcess.stdout.on("data", (chunk: Uint8Array) => - raw.stdout.send(chunk), - ); - childProcess.stdout.once("close", () => raw.stdout.close()); - childProcess.stderr.on("data", (chunk: Uint8Array) => - raw.stderr.send(chunk), - ); - childProcess.stderr.once("close", () => raw.stderr.close()); - let io = { stdoutDone: withResolvers(), stderrDone: withResolvers(), @@ -88,7 +59,7 @@ export function* createPosixProcess( let stderr = createSignal(); yield* spawn(function* () { - let subscription = yield* raw.stdout; + let subscription = yield* os.stdout; let next = yield* subscription.next(); while (!next.done) { yield* Stdio.operations.stdout(next.value); @@ -100,7 +71,7 @@ export function* createPosixProcess( }); yield* spawn(function* () { - let subscription = yield* raw.stderr; + let subscription = yield* os.stderr; let next = yield* subscription.next(); while (!next.done) { yield* Stdio.operations.stderr(next.value); @@ -113,12 +84,12 @@ export function* createPosixProcess( let stdin: Writable = { send(data: string) { - childProcess.stdin.write(data); + childProcess.stdin?.write(data); }, }; function* join() { - let result = yield* processResult.operation; + let result = yield* os.result; 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 07968bc9..1a0bb24d 100644 --- a/process/src/exec/win32.ts +++ b/process/src/exec/win32.ts @@ -7,8 +7,6 @@ import { type Operation, type Result, type Yielded, - Err, - Ok, all, createSignal, ensure, @@ -24,10 +22,9 @@ import type { } from "./types.ts"; import { Stdio } from "../api.ts"; import { ExecError } from "./error.ts"; +import { type ProcessResultValue, useNativeProcess } from "./native.ts"; import { unbox, useEvalScope } from "@effectionx/scope-eval"; -type ProcessResultValue = [number?, string?]; - function* killTree(pid: number) { try { const killer = spawnProcess( @@ -45,59 +42,35 @@ export function* createWin32Process( command: string, options: ExecOptions, ): Operation { - let processResult = withResolvers>(); const evalScope = yield* useEvalScope(); const result = yield* evalScope.eval(function* () { // Windows-specific process spawning with different options than POSIX - let childProcess = spawnProcess(command, options.arguments || [], { - // We lose exit information and events if this is detached in windows - // and it opens a window in windows+powershell. - detached: false, - // The `shell` option is passed to `cross-spawn` to control whether a shell is used. - // On Windows, `shell: true` is necessary to run command strings, as it uses - // `cmd.exe` to parse the command and find executables in the PATH. - // Using a boolean `true` was previously disabled, causing ENOENT errors for - // commands that were not a direct path to an executable. - shell: options.shell || false, - // With stdio as pipe, windows gets stuck where neither the child nor the - // parent wants to close the stream, so we call it ourselves in the exit event. - stdio: "pipe", - // Hide the child window so that killing it will not block the parent - // with a Terminate Batch Process (Y/n) - windowsHide: true, - env: options.env, - cwd: options.cwd, - }); + const os = yield* useNativeProcess(() => + spawnProcess(command, options.arguments || [], { + // We lose exit information and events if this is detached in windows + // and it opens a window in windows+powershell. + detached: false, + // The `shell` option is passed to `cross-spawn` to control whether a shell is used. + // On Windows, `shell: true` is necessary to run command strings, as it uses + // `cmd.exe` to parse the command and find executables in the PATH. + // Using a boolean `true` was previously disabled, causing ENOENT errors for + // commands that were not a direct path to an executable. + shell: options.shell || false, + // With stdio as pipe, windows gets stuck where neither the child nor the + // parent wants to close the stream, so we call it ourselves in the exit event. + stdio: "pipe", + // Hide the child window so that killing it will not block the parent + // with a Terminate Batch Process (Y/n) + windowsHide: true, + env: options.env, + cwd: options.cwd, + }), + ); + let { childProcess } = os; let { pid } = childProcess; - if (!childProcess.stdout || !childProcess.stderr) { - throw new Error("stdout and stderr must be available with stdio: pipe"); - } - - // Native listeners attached in the same synchronous continuation as the - // spawn: process events and stdio chunks cannot be missed, and no - // readable pumps are needed. - let rawClose = withResolvers(); - childProcess.once("error", (error) => { - processResult.resolve(Err(error)); - }); - childProcess.once("close", (code, signal) => { - rawClose.resolve([code ?? undefined, signal ?? undefined]); - }); - - let raw = { - stdout: createSignal(), - stderr: createSignal(), - }; - childProcess.stdout.on("data", (chunk: Uint8Array) => - raw.stdout.send(chunk), - ); - childProcess.stdout.once("close", () => raw.stdout.close()); - childProcess.stderr.on("data", (chunk: Uint8Array) => - raw.stderr.send(chunk), - ); - childProcess.stderr.once("close", () => raw.stderr.close()); + let processResult = withResolvers>(); let io = { stdoutDone: withResolvers(), @@ -108,7 +81,7 @@ export function* createWin32Process( const stderr = createSignal(); yield* spawn(function* () { - let subscription = yield* raw.stdout; + let subscription = yield* os.stdout; let next = yield* subscription.next(); while (!next.done) { yield* Stdio.operations.stdout(next.value); @@ -120,7 +93,7 @@ export function* createWin32Process( }); yield* spawn(function* () { - let subscription = yield* raw.stderr; + let subscription = yield* os.stderr; let next = yield* subscription.next(); while (!next.done) { yield* Stdio.operations.stderr(next.value); @@ -133,17 +106,17 @@ export function* createWin32Process( let stdin: Writable = { send(data: string) { - childProcess.stdin.write(data); + childProcess.stdin?.write(data); }, }; yield* spawn(function* () { - let value = yield* rawClose.operation; + let value = yield* os.result; // 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 yield* all([io.stdoutDone.operation, io.stderrDone.operation]); - processResult.resolve(Ok(value)); + processResult.resolve(value); }); function* join() { @@ -166,7 +139,7 @@ export function* createWin32Process( // Suppress EPIPE errors on stdin - these occur on Windows when the child // process exits before we finish writing to it. This is expected during // cleanup when we're killing the process. - childProcess.stdin.on("error", (err: Error & { code?: string }) => { + childProcess.stdin?.on("error", (err: Error & { code?: string }) => { if (err.code !== "EPIPE") { throw err; } @@ -184,15 +157,17 @@ export function* createWin32Process( } let stdin = childProcess.stdin; - if (stdin.writable) { - try { - //Terminate batch process (Y/N) - stdin.write("Y\n"); - } catch (_err) { - /* not much we can do here */ + if (stdin) { + if (stdin.writable) { + try { + //Terminate batch process (Y/N) + stdin.write("Y\n"); + } catch (_err) { + /* not much we can do here */ + } } + stdin.end(); } - stdin.end(); } // depending on how we shutdown, this may already be closed and From 226cfa9293c6754de57fc30d677494f8367d42fe Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Wed, 12 Aug 2026 21:55:26 -0400 Subject: [PATCH 3/4] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Bind=20native=20listen?= =?UTF-8?q?er=20deregistration=20to=20the=20resource=20scope?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace every once() registration with on(), name the handlers, and remove them in the standard-cleanup ensure. Un-registering a once() listener depends on its event firing; listener lifetime must depend only on the scope, especially for stdio pipes that may be shared beyond a single process. Removal comes after the ensure's wait for the process result, which needs the close listener still attached. --- process/src/exec/native.ts | 40 ++++++++++++++++++++++++++++---------- 1 file changed, 30 insertions(+), 10 deletions(-) diff --git a/process/src/exec/native.ts b/process/src/exec/native.ts index 133d4eb5..ea425e42 100644 --- a/process/src/exec/native.ts +++ b/process/src/exec/native.ts @@ -44,6 +44,17 @@ export function useNativeProcess( let child: ChildProcess | undefined; + let onError = (error: Error) => { + result.resolve(Err(error)); + }; + let onClose = (code: number | null, signal: NodeJS.Signals | null) => { + result.resolve(Ok([code ?? undefined, signal ?? undefined])); + }; + let onStdout = (chunk: Uint8Array) => stdout.send(chunk); + let onStdoutClose = () => stdout.close(); + let onStderr = (chunk: Uint8Array) => stderr.send(chunk); + let onStderrClose = () => stderr.close(); + // Registered while `child` is still undefined; the spawn below follows // in the same synchronous continuation, so the process never exists // without this teardown armed. @@ -55,26 +66,35 @@ export function useNativeProcess( // hold the scope until the OS has reaped the process and closed its // pipes; `result` is a value, so an error outcome cannot throw here yield* result.operation; + // deregistration is bound to this scope, never to an event firing, + // because the pipes may be shared beyond this process. It must + // follow the wait above, which needs `onClose` still attached. + child.off("error", onError); + child.off("close", onClose); + child.stdout?.off("data", onStdout); + child.stdout?.off("close", onStdoutClose); + child.stderr?.off("data", onStderr); + child.stderr?.off("close", onStderrClose); } }); const spawned = create(); child = spawned; - spawned.once("error", (error) => { - result.resolve(Err(error)); - }); - spawned.once("close", (code, signal) => { - result.resolve(Ok([code ?? undefined, signal ?? undefined])); - }); + spawned.on("error", onError); + spawned.on("close", onClose); if (!spawned.stdout || !spawned.stderr) { throw new Error("stdout and stderr must be available with stdio: pipe"); } - spawned.stdout.on("data", (chunk: Uint8Array) => stdout.send(chunk)); - spawned.stdout.once("close", () => stdout.close()); - spawned.stderr.on("data", (chunk: Uint8Array) => stderr.send(chunk)); - spawned.stderr.once("close", () => stderr.close()); + + // Node emits the child "close" event only after both stdio streams have + // closed, and these listeners run synchronously with stream emission, so + // by the time `result` resolves every chunk is already in the signals. + spawned.stdout.on("data", onStdout); + spawned.stdout.on("close", onStdoutClose); + spawned.stderr.on("data", onStderr); + spawned.stderr.on("close", onStderrClose); yield* provide({ childProcess: spawned, From 126a52a0d365c281d86c07561f4757c1f84b46cd Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Wed, 12 Aug 2026 22:13:32 -0400 Subject: [PATCH 4/4] =?UTF-8?q?=F0=9F=90=9B=20Guarantee=20listener=20dereg?= =?UTF-8?q?istration=20and=20document=20hot=20streams?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wrap the standard-cleanup wait in a synchronous finally so the off() calls run even if the wait never completes; deregistration must not depend on the close event arriving. Document on NativeProcess that the output streams are hot: subscribe in the acquiring continuation, before awaiting result. Bump @effectionx/process to 0.8.3. --- process/package.json | 2 +- process/src/exec/native.ts | 38 ++++++++++++++++++++++++-------------- 2 files changed, 25 insertions(+), 15 deletions(-) diff --git a/process/package.json b/process/package.json index 6215178a..9a8e72cf 100644 --- a/process/package.json +++ b/process/package.json @@ -1,7 +1,7 @@ { "name": "@effectionx/process", "description": "Spawn and manage child processes with structured concurrency", - "version": "0.8.2", + "version": "0.8.3", "keywords": ["process"], "type": "module", "main": "./dist/mod.js", diff --git a/process/src/exec/native.ts b/process/src/exec/native.ts index ea425e42..f7884d7d 100644 --- a/process/src/exec/native.ts +++ b/process/src/exec/native.ts @@ -27,6 +27,12 @@ export type ProcessResultValue = [number?, string?]; */ export interface NativeProcess { childProcess: ChildProcess; + + /** + * Hot streams: chunks that arrive before a subscription exists are + * dropped. Subscribe in the same continuation that acquires the resource, + * before awaiting `result`. + */ stdout: Stream; stderr: Stream; @@ -60,21 +66,25 @@ export function useNativeProcess( // without this teardown armed. yield* ensure(function* () { if (child) { - if (child.exitCode === null && child.signalCode === null) { - child.kill("SIGTERM"); + try { + if (child.exitCode === null && child.signalCode === null) { + child.kill("SIGTERM"); + } + // hold the scope until the OS has reaped the process and closed its + // pipes; `result` is a value, so an error outcome cannot throw here + yield* result.operation; + } finally { + // deregistration is bound to this scope, never to an event firing, + // because the pipes may be shared beyond this process. It follows + // the wait above, which needs `onClose` still attached, but must + // not depend on that wait completing — hence the sync finally. + child.off("error", onError); + child.off("close", onClose); + child.stdout?.off("data", onStdout); + child.stdout?.off("close", onStdoutClose); + child.stderr?.off("data", onStderr); + child.stderr?.off("close", onStderrClose); } - // hold the scope until the OS has reaped the process and closed its - // pipes; `result` is a value, so an error outcome cannot throw here - yield* result.operation; - // deregistration is bound to this scope, never to an event firing, - // because the pipes may be shared beyond this process. It must - // follow the wait above, which needs `onClose` still attached. - child.off("error", onError); - child.off("close", onClose); - child.stdout?.off("data", onStdout); - child.stdout?.off("close", onStdoutClose); - child.stderr?.off("data", onStderr); - child.stderr?.off("close", onStderrClose); } });