fix(node): Create additional pipes for child processes (#25016)

Linux/macos only currently.

Part of https://github.com/denoland/deno/issues/23524 (fixes it on
platforms other than windows).
Part of #16899  (fixes it on platforms other than windows).

After this PR, playwright is functional on mac/linux.

1 files changed, 85 insertions, 17 deletions

diff --git a/ext/node/polyfills/internal/child_process.ts b/ext/node/polyfills/internal/child_process.ts
index 0996806d7..edc11df73 100644
--- a/ext/node/polyfills/internal/child_process.ts
+++ b/ext/node/polyfills/internal/child_process.ts
@@ -25,7 +25,6 @@ import {
   StringPrototypeStartsWith,
   StringPrototypeToUpperCase,
 } from "ext:deno_node/internal/primordials.mjs";
-
 import { assert } from "ext:deno_node/_util/asserts.ts";
 import { EventEmitter } from "node:events";
 import { os } from "ext:deno_node/internal_binding/constants.ts";
@@ -54,6 +53,10 @@ import { kEmptyObject } from "ext:deno_node/internal/util.mjs";
 import { getValidatedPath } from "ext:deno_node/internal/fs/utils.mjs";
 import process from "node:process";
 import { StringPrototypeSlice } from "ext:deno_node/internal/primordials.mjs";
+import { StreamBase } from "ext:deno_node/internal_binding/stream_wrap.ts";
+import { Pipe, socketType } from "ext:deno_node/internal_binding/pipe_wrap.ts";
+import console from "node:console";
+import { Socket } from "node:net";
 
 export function mapValues<T, O>(
   record: Readonly<Record<string, T>>,
@@ -118,6 +121,47 @@ function maybeClose(child: ChildProcess) {
   }
 }
 
+function flushStdio(subprocess: ChildProcess) {
+  const stdio = subprocess.stdio;
+
+  if (stdio == null) return;
+
+  for (let i = 0; i < stdio.length; i++) {
+    const stream = stdio[i];
+    if (!stream || !stream.readable) {
+      continue;
+    }
+    stream.resume();
+  }
+}
+
+// Wraps a resource in a class that implements
+// StreamBase, so it can be used with node streams
+class StreamResource implements StreamBase {
+  #rid: number;
+  constructor(rid: number) {
+    this.#rid = rid;
+  }
+  close(): void {
+    core.close(this.#rid);
+  }
+  async read(p: Uint8Array): Promise<number | null> {
+    const readPromise = core.read(this.#rid, p);
+    core.unrefOpPromise(readPromise);
+    const nread = await readPromise;
+    return nread > 0 ? nread : null;
+  }
+  ref(): void {
+    return;
+  }
+  unref(): void {
+    return;
+  }
+  write(p: Uint8Array): Promise<number> {
+    return core.write(this.#rid, p);
+  }
+}
+
 export class ChildProcess extends EventEmitter {
   /**
    * The exit code of the child process. This property will be `null` until the child process exits.
@@ -201,7 +245,7 @@ export class ChildProcess extends EventEmitter {
       stdin = "pipe",
       stdout = "pipe",
       stderr = "pipe",
-      _channel, // TODO(kt3k): handle this correctly
+      ...extraStdio
     ] = normalizedStdio;
     const [cmd, cmdArgs] = buildCommand(
       command,
@@ -213,6 +257,15 @@ export class ChildProcess extends EventEmitter {
 
     const ipc = normalizedStdio.indexOf("ipc");
 
+    const extraStdioOffset = 3; // stdin, stdout, stderr
+
+    const extraStdioNormalized: DenoStdio[] = [];
+    for (let i = 0; i < extraStdio.length; i++) {
+      const fd = i + extraStdioOffset;
+      if (fd === ipc) extraStdioNormalized.push("null");
+      extraStdioNormalized.push(toDenoStdio(extraStdio[i]));
+    }
+
     const stringEnv = mapValues(env, (value) => value.toString());
     try {
       this.#process = new Deno.Command(cmd, {
@@ -224,6 +277,7 @@ export class ChildProcess extends EventEmitter {
         stderr: toDenoStdio(stderr),
         windowsRawArguments: windowsVerbatimArguments,
         ipc, // internal
+        extraStdio: extraStdioNormalized,
       }).spawn();
       this.pid = this.#process.pid;
 
@@ -259,13 +313,36 @@ export class ChildProcess extends EventEmitter {
           maybeClose(this);
         });
       }
-      // TODO(nathanwhit): once we impl > 3 stdio pipes make sure we also listen for their
-      // close events (like above)
 
       this.stdio[0] = this.stdin;
       this.stdio[1] = this.stdout;
       this.stdio[2] = this.stderr;
 
+      if (ipc >= 0) {
+        this.stdio[ipc] = null;
+      }
+
+      const pipeRids = internals.getExtraPipeRids(this.#process);
+      for (let i = 0; i < pipeRids.length; i++) {
+        const rid: number | null = pipeRids[i];
+        const fd = i + extraStdioOffset;
+        if (rid) {
+          this[kClosesNeeded]++;
+          this.stdio[fd] = new Socket(
+            {
+              handle: new Pipe(
+                socketType.IPC,
+                new StreamResource(rid),
+              ),
+              // deno-lint-ignore no-explicit-any
+            } as any,
+          );
+          this.stdio[fd]?.on("close", () => {
+            maybeClose(this);
+          });
+        }
+      }
+
       nextTick(() => {
         this.emit("spawn");
         this.#spawned.resolve();
@@ -292,9 +369,9 @@ export class ChildProcess extends EventEmitter {
         }
       }
 
-      const pipeFd = internals.getPipeFd(this.#process);
-      if (typeof pipeFd == "number") {
-        setupChannel(this, pipeFd);
+      const pipeRid = internals.getIpcPipeRid(this.#process);
+      if (typeof pipeRid == "number") {
+        setupChannel(this, pipeRid);
         this[kClosesNeeded]++;
         this.on("disconnect", () => {
           maybeClose(this);
@@ -312,6 +389,7 @@ export class ChildProcess extends EventEmitter {
           await this.#_waitForChildStreamsToClose();
           this.#closePipes();
           maybeClose(this);
+          nextTick(flushStdio, this);
         });
       })();
     } catch (err) {
@@ -395,16 +473,6 @@ export class ChildProcess extends EventEmitter {
       assert(this.stdin);
       this.stdin.destroy();
     }
-    /// TODO(nathanwhit): for some reason when the child process exits
-    /// and the child end of the named pipe closes, reads still just return `Pending`
-    /// instead of returning that 0 bytes were read (to signal the pipe died).
-    /// For now, just forcibly disconnect, but in theory I think we could miss messages
-    /// that haven't been read yet.
-    if (Deno.build.os === "windows") {
-      if (this.canDisconnect) {
-        this.disconnect?.();
-      }
-    }
   }
 }