✨ useNativeProcess: a native-listener child-process resource

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",

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<Uint8Array, void>;
+  stderr: Stream<Uint8Array, void>;
+
+  /** The `close` event value, or the error that prevented the spawn. */
+  result: Operation<Result<ProcessResultValue>>;
+}
+
+export function useNativeProcess(
+  create: () => ChildProcess,
+): Operation<NativeProcess> {
+  return resource(function* (provide) {
+    let result = withResolvers<Result<ProcessResultValue>>();
+    let stdout = createSignal<Uint8Array, void>();
+    let stderr = createSignal<Uint8Array, void>();
+
+    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,
+    });
+  });
+}

process/src/exec/posix.ts

@@ -1,20 +1,15 @@
 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,
   spawn,
   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<Process> {
-  let processResult = withResolvers<Result<ProcessResultValue>>();
   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<void>(),
       stderrDone: withResolvers<void>(),
     };
@@ -69,45 +59,37 @@ export function* createPosixProcess(
     let stderr = createSignal<Uint8Array, void>();
 
     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();
     });
 
     let stdin: Writable<string> = {
       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<ProcessResultValue>(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;