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 new file mode 100644 index 00000000..f7884d7d --- /dev/null +++ b/process/src/exec/native.ts @@ -0,0 +1,116 @@ +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; + + /** + * 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; + + /** 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; + + 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. + yield* ensure(function* () { + if (child) { + 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); + } + } + }); + + const spawned = create(); + child = spawned; + + 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"); + } + + // 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, + stdout, + stderr, + result: result.operation, + }); + }); +} diff --git a/process/src/exec/posix.ts b/process/src/exec/posix.ts index 6b5bb9bc..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, @@ -13,8 +10,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, @@ -24,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 @@ -44,23 +37,20 @@ 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"); - } - let io = { - stdout: yield* fromReadable(childProcess.stdout), - stderr: yield* fromReadable(childProcess.stderr), stdoutDone: withResolvers(), stderrDone: withResolvers(), }; @@ -69,22 +59,24 @@ export function* createPosixProcess( let stderr = createSignal(); yield* spawn(function* () { - let next = yield* io.stdout.next(); + 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* io.stdout.next(); + next = yield* subscription.next(); } stdout.close(); io.stdoutDone.resolve(); }); yield* spawn(function* () { - let next = yield* io.stderr.next(); + 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* io.stderr.next(); + next = yield* subscription.next(); } stderr.close(); io.stderrDone.resolve(); @@ -92,22 +84,12 @@ export function* createPosixProcess( let stdin: Writable = { send(data: string) { - childProcess.stdin.write(data); + childProcess.stdin?.write(data); }, }; - 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; + 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 3a2c78b2..1a0bb24d 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"; @@ -8,8 +7,6 @@ import { type Operation, type Result, type Yielded, - Err, - Ok, all, createSignal, ensure, @@ -25,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( @@ -46,39 +42,37 @@ 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"); - } + let processResult = withResolvers>(); let io = { - stdout: yield* fromReadable(childProcess.stdout), - stderr: yield* fromReadable(childProcess.stderr), stdoutDone: withResolvers(), stderrDone: withResolvers(), }; @@ -87,22 +81,24 @@ export function* createWin32Process( const stderr = createSignal(); yield* spawn(function* () { - let next = yield* io.stdout.next(); + 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* io.stdout.next(); + next = yield* subscription.next(); } stdout.close(); io.stdoutDone.resolve(); }); yield* spawn(function* () { - let next = yield* io.stderr.next(); + 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* io.stderr.next(); + next = yield* subscription.next(); } stderr.close(); io.stderrDone.resolve(); @@ -110,22 +106,17 @@ export function* createWin32Process( let stdin: Writable = { send(data: string) { - childProcess.stdin.write(data); + childProcess.stdin?.write(data); }, }; - 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* 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() { @@ -148,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; } @@ -166,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