🐛 Settle process join() only after stdout/stderr pumps complete

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 "close" event is
+ * received, before waiting on the output pumps. 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

@@ -24,6 +24,7 @@ import type {
 } from "./types.ts";
 import { Stdio } from "../api.ts";
 import { ExecError } from "./error.ts";
+import { CloseEvent } from "./internal.ts";
 
 type ProcessResultValue = [number?, string?];
 
@@ -69,25 +70,37 @@ export function* createPosixProcess(
     let stderr = createSignal<Uint8Array, void>();
 
     yield* spawn(function* () {
-      let next = yield* io.stdout.next();
-      while (!next.done) {
-        yield* Stdio.operations.stdout(next.value);
-        stdout.send(next.value);
-        next = yield* io.stdout.next();
+      try {
+        let next = yield* io.stdout.next();
+        while (!next.done) {
+          yield* Stdio.operations.stdout(next.value);
+          stdout.send(next.value);
+          next = yield* io.stdout.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 next = yield* io.stderr.next();
-      while (!next.done) {
-        yield* Stdio.operations.stderr(next.value);
-        stderr.send(next.value);
-        next = yield* io.stderr.next();
+      try {
+        let next = yield* io.stderr.next();
+        while (!next.done) {
+          yield* Stdio.operations.stderr(next.value);
+          stderr.send(next.value);
+          next = yield* io.stderr.next();
+        }
+      } catch (error) {
+        processResult.resolve(Err(error as Error));
+      } finally {
+        stderr.close();
+        io.stderrDone.resolve();
       }
-      stderr.close();
-      io.stderrDone.resolve();
     });
 
     let stdin: Writable<string> = {
@@ -103,6 +116,14 @@ export function* createPosixProcess(
 
     yield* spawn(function* () {
       let value = yield* once<ProcessResultValue>(childProcess, "close");
+      let closeEvent = yield* CloseEvent.get();
+      if (closeEvent) {
+        yield* closeEvent();
+      }
+      // wait for both pumps to finish forwarding buffered output so that
+      // join() never settles before the final chunks have passed through
+      // Stdio middleware and the public signals
+      yield* all([io.stdoutDone.operation, io.stderrDone.operation]);
       processResult.resolve(Ok(value));
     });
 

process/src/exec/win32.ts

@@ -25,6 +25,7 @@ import type {
 } from "./types.ts";
 import { Stdio } from "../api.ts";
 import { ExecError } from "./error.ts";
+import { CloseEvent } from "./internal.ts";
 import { unbox, useEvalScope } from "@effectionx/scope-eval";
 
 type ProcessResultValue = [number?, string?];
@@ -87,25 +88,37 @@ export function* createWin32Process(
     const stderr = createSignal<Uint8Array, void>();
 
     yield* spawn(function* () {
-      let next = yield* io.stdout.next();
-      while (!next.done) {
-        yield* Stdio.operations.stdout(next.value);
-        stdout.send(next.value);
-        next = yield* io.stdout.next();
+      try {
+        let next = yield* io.stdout.next();
+        while (!next.done) {
+          yield* Stdio.operations.stdout(next.value);
+          stdout.send(next.value);
+          next = yield* io.stdout.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 next = yield* io.stderr.next();
-      while (!next.done) {
-        yield* Stdio.operations.stderr(next.value);
-        stderr.send(next.value);
-        next = yield* io.stderr.next();
+      try {
+        let next = yield* io.stderr.next();
+        while (!next.done) {
+          yield* Stdio.operations.stderr(next.value);
+          stderr.send(next.value);
+          next = yield* io.stderr.next();
+        }
+      } catch (error) {
+        processResult.resolve(Err(error as Error));
+      } finally {
+        stderr.close();
+        io.stderrDone.resolve();
       }
-      stderr.close();
-      io.stderrDone.resolve();
     });
 
     let stdin: Writable<string> = {
@@ -121,6 +134,10 @@ export function* createWin32Process(
 
     yield* spawn(function* () {
       let value = yield* once<ProcessResultValue>(childProcess, "close");
+      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
+      // stdout pump 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[] = [];