Repository: denoland/deno
Pull request: fix(ext/node): make `child_process` stdio streams Socket instances

Closes #31961
Closes #25602

This PR fixes child process stdio streams (`stdin`, `stdout`, `stderr`) to be instances of `Socket` from `node:net`, matching Node.js behavior.

Changed files:
filename: ext/node/polyfills/internal/child_process.ts
patch: @@ -156,6 +156,8 @@ function flushStdio(subprocess: ChildProcess) {
 // StreamBase, so it can be used with node streams
 class StreamResource implements StreamBase {
   #rid: number;
+  #isUnref = false;
+  #pendingPromises: Set<Promise<number>> = new Set();
   constructor(rid: number) {
     this.#rid = rid;
   }
@@ -164,18 +166,40 @@ class StreamResource implements StreamBase {
   }
   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;
+    this.#pendingPromises.add(readPromise);
+    if (this.#isUnref) {
+      core.unrefOpPromise(readPromise);
+    }
+    try {
+      const nread = await readPromise;
+      return nread > 0 ? nread : null;
+    } finally {
+      this.#pendingPromises.delete(readPromise);
+    }
   }
   ref(): void {
-    return;
+    this.#isUnref = false;
+    for (const promise of this.#pendingPromises) {
+      core.refOpPromise(promise);
+    }
   }
   unref(): void {
-    return;
+    this.#isUnref = true;
+    for (const promise of this.#pendingPromises) {
+      core.unrefOpPromise(promise);
+    }
   }
-  write(p: Uint8Array): Promise<number> {
-    return core.write(this.#rid, p);
+  async write(p: Uint8Array): Promise<number> {
+    const writePromise = core.write(this.#rid, p);
+    this.#pendingPromises.add(writePromise);
+    if (this.#isUnref) {
+      core.unrefOpPromise(writePromise);
+    }
+    try {
+      return await writePromise;
+    } finally {
+      this.#pendingPromises.delete(writePromise);
+    }
   }
 }
 
@@ -377,9 +401,25 @@ export class ChildProcess extends EventEmitter {
       }).spawn();
       this.pid = this.#process.pid;
 
+      // Get stdio rids to create Socket instances
+      const stdioRids = internals.getStdioRids(this.#process);
+
       if (stdin === "pipe") {
         assert(this.#process.stdin);
-        this.stdin = Writable.fromWeb(this.#process.stdin);
+        if (stdioRids.stdinRid !== null) {
+          // Create Socket instance for stdin (like Node.js does)
+          this.stdin = new Socket({
+            handle: new Pipe(
+              socketType.SOCKET,
+              new StreamResource(stdioRids.stdinRid),
+            ),
+            writable: true,
+            readable: false,
+          });
+        } else {
+          // Fallback to web stream conversion
+          this.stdin = Writable.fromWeb(this.#process.stdin);
+        }
       }
 
       if (stdin instanceof Stream) {
@@ -395,7 +435,20 @@ export class ChildProcess extends EventEmitter {
       if (stdout === "pipe") {
         assert(this.#process.stdout);
         this[kClosesNeeded]++;
-        this.stdout = Readable.fromWeb(this.#process.stdout);
+        if (stdioRids.stdoutRid !== null) {
+          // Create Socket instance for stdout (like Node.js does)
+          this.stdout = new Socket({
+            handle: new Pipe(
+              socketType.SOCKET,
+              new StreamResource(stdioRids.stdoutRid),
+            ),
+            writable: false,
+            readable: true,
+          });
+        } else {
+          // Fallback to web stream conversion
+          this.stdout = Readable.fromWeb(this.#process.stdout);
+        }
         this.stdout.on("close", () => {
           maybeClose(this);
         });
@@ -404,7 +457,20 @@ export class ChildProcess extends EventEmitter {
       if (stderr === "pipe") {
         assert(this.#process.stderr);
         this[kClosesNeeded]++;
-        this.stderr = Readable.fromWeb(this.#process.stderr);
+        if (stdioRids.stderrRid !== null) {
+          // Create Socket instance for stderr (like Node.js does)
+          this.stderr = new Socket({
+            handle: new Pipe(
+              socketType.SOCKET,
+              new StreamResource(stdioRids.stderrRid),
+            ),
+            writable: false,
+            readable: true,
+          });
+        } else {
+          // Fallback to web stream conversion
+          this.stderr = Readable.fromWeb(this.#process.stderr);
+        }
         this.stderr.on("close", () => {
           maybeClose(this);
         });

filename: ext/process/40_process.js
patch: @@ -228,9 +228,17 @@ function collectOutput(readableStream) {
 
 const _ipcPipeRid = Symbol("[[ipcPipeRid]]");
 const _extraPipeRids = Symbol("[[_extraPipeRids]]");
+const _stdinRid = Symbol("[[stdinRid]]");
+const _stdoutRid = Symbol("[[stdoutRid]]");
+const _stderrRid = Symbol("[[stderrRid]]");
 
 internals.getIpcPipeRid = (process) => process[_ipcPipeRid];
 internals.getExtraPipeRids = (process) => process[_extraPipeRids];
+internals.getStdioRids = (process) => ({
+  stdinRid: process[_stdinRid],
+  stdoutRid: process[_stdoutRid],
+  stderrRid: process[_stderrRid],
+});
 internals.kExtraStdio = kExtraStdio;
 
 class ChildProcess {
@@ -240,6 +248,9 @@ class ChildProcess {
 
   [_ipcPipeRid];
   [_extraPipeRids];
+  [_stdinRid];
+  [_stdoutRid];
+  [_stderrRid];
 
   #pid;
   get pid() {
@@ -288,6 +299,9 @@ class ChildProcess {
     this.#pid = pid;
     this[_ipcPipeRid] = ipcPipeRid;
     this[_extraPipeRids] = extraPipeRids;
+    this[_stdinRid] = stdinRid;
+    this[_stdoutRid] = stdoutRid;
+    this[_stderrRid] = stderrRid;
 
     if (stdinRid !== null) {
       this.#stdin = writableStreamForRid(stdinRid);

filename: tests/specs/node/child_process_stdio_socket/__test__.jsonc
patch: @@ -0,0 +1,8 @@
+{
+  "tests": {
+    "stdio_is_socket_instance": {
+      "args": "run --allow-all main.ts",
+      "output": "main.out"
+    }
+  }
+}

filename: tests/specs/node/child_process_stdio_socket/main.out
patch: @@ -0,0 +1,7 @@
+stdout instanceof Socket: true
+stdout has unref: true
+stderr instanceof Socket: true
+stderr has unref: true
+stdin instanceof Socket: true
+stdin has unref: true
+unref() calls succeeded

filename: tests/specs/node/child_process_stdio_socket/main.ts
patch: @@ -0,0 +1,38 @@
+import { spawn } from "node:child_process";
+import { Socket } from "node:net";
+
+// Test that stdio streams are Socket instances (Node.js compatibility)
+// Issue: https://github.com/denoland/deno/issues/31961 and https://github.com/denoland/deno/issues/25602
+
+const cp = spawn("node", [], {
+  stdio: "pipe",
+});
+
+// stdout should be a Socket instance
+console.log("stdout instanceof Socket:", cp.stdout instanceof Socket);
+console.log(
+  "stdout has unref:",
+  "unref" in cp.stdout! && typeof cp.stdout!.unref === "function",
+);
+
+// stderr should be a Socket instance
+console.log("stderr instanceof Socket:", cp.stderr instanceof Socket);
+console.log(
+  "stderr has unref:",
+  "unref" in cp.stderr! && typeof cp.stderr!.unref === "function",
+);
+
+// stdin should be a Socket instance
+console.log("stdin instanceof Socket:", cp.stdin instanceof Socket);
+console.log(
+  "stdin has unref:",
+  "unref" in cp.stdin! && typeof cp.stdin!.unref === "function",
+);
+
+// Verify that unref() can be called without error
+cp.stdout!.unref();
+cp.stderr!.unref();
+cp.stdin!.unref();
+console.log("unref() calls succeeded");
+
+cp.kill();