🐛 Settle join() only after output clears Stdio middleware

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<void>>(
+  "@effectionx/process:close-event",
+);

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<Result<ProcessResultValue>>();
+
     let io = {
       stdoutDone: withResolvers<void>(),
       stderrDone: withResolvers<void>(),
@@ -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<string> = {
@@ -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;

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<string> = {
@@ -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

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<void>();
+      const release = withResolvers<void>();
+      const procReady = withResolvers<Process>();
+      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<void>();
+      const release = withResolvers<void>();
+      const procReady = withResolvers<Process>();
+      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[] = [];