fix(node): Rework node:child_process IPC (#24763)

Fixes https://github.com/denoland/deno/issues/24756. Fixes
https://github.com/denoland/deno/issues/24796.

This also gets vitest working when using
[`--pool=forks`](https://vitest.dev/guide/improving-performance#pool)
(which is the default as of vitest 2.0). Ref
https://github.com/denoland/deno/issues/23882.

---

This PR resolves a handful of issues with child_process IPC. In
particular:

- We didn't support sending typed array views over IPC
- Opening an IPC channel resulted in the event loop never exiting
- Sending a `null` over IPC would terminate the channel
- There was some UB in the read implementation (transmuting an `&[u8]`
to `&mut [u8]`)
- The `send` method wasn't returning anything, so there was no way to
signal backpressure (this also resulted in the benchmark
`child_process_ipc.mjs` being misleading, as it tried to respect
backpressure. That gave node much worse results at larger message sizes,
and gave us much worse results at smaller message sizes).
- We weren't setting up the `channel` property on the `process` global
(or on the `ChildProcess` object), and also didn't have a way to
ref/unref the channel
- Calling `kill` multiple times (or disconnecting the channel, then
calling kill) would throw an error
- Node couldn't spawn a deno subprocess and communicate with it over IPC

1 files changed, 17 insertions, 15 deletions

diff --git a/runtime/ops/process.rs b/runtime/ops/process.rs
index ecf6ef49b..69fb5cf29 100644
--- a/runtime/ops/process.rs
+++ b/runtime/ops/process.rs
@@ -345,14 +345,15 @@ fn create_command(
       });
 
       /* One end returned to parent process (this) */
-      let pipe_rid = Some(
-        state
-          .resource_table
-          .add(deno_node::IpcJsonStreamResource::new(fd1 as _)?),
-      );
+      let pipe_rid = Some(state.resource_table.add(
+        deno_node::IpcJsonStreamResource::new(
+          fd1 as _,
+          deno_node::IpcRefTracker::new(state.external_ops_tracker.clone()),
+        )?,
+      ));
 
-      /* The other end passed to child process via DENO_CHANNEL_FD */
-      command.env("DENO_CHANNEL_FD", format!("{}", ipc));
+      /* The other end passed to child process via NODE_CHANNEL_FD */
+      command.env("NODE_CHANNEL_FD", format!("{}", ipc));
 
       return Ok((command, pipe_rid));
     }
@@ -470,14 +471,15 @@ fn create_command(
       }
 
       /* One end returned to parent process (this) */
-      let pipe_fd = Some(
-        state
-          .resource_table
-          .add(deno_node::IpcJsonStreamResource::new(hd1 as i64)?),
-      );
-
-      /* The other end passed to child process via DENO_CHANNEL_FD */
-      command.env("DENO_CHANNEL_FD", format!("{}", hd2 as i64));
+      let pipe_fd = Some(state.resource_table.add(
+        deno_node::IpcJsonStreamResource::new(
+          hd1 as i64,
+          deno_node::IpcRefTracker::new(state.external_ops_tracker.clone()),
+        )?,
+      ));
+
+      /* The other end passed to child process via NODE_CHANNEL_FD */
+      command.env("NODE_CHANNEL_FD", format!("{}", hd2 as i64));
 
       return Ok((command, pipe_fd));
     }