fix(node): Rework node:child_process IPC

cli/args/mod.rs

@@ -1050,10 +1050,10 @@ impl CliOptions {
   }
 
   pub fn node_ipc_fd(&self) -> Option<i64> {
-    let maybe_node_channel_fd = std::env::var("DENO_CHANNEL_FD").ok();
+    let maybe_node_channel_fd = std::env::var("NODE_CHANNEL_FD").ok();
     if let Some(node_channel_fd) = maybe_node_channel_fd {
       // Remove so that child processes don't inherit this environment variable.
-      std::env::remove_var("DENO_CHANNEL_FD");
+      std::env::remove_var("NODE_CHANNEL_FD");
       node_channel_fd.parse::<i64>().ok()
     } else {
       None

ext/node/Cargo.toml

@@ -55,6 +55,7 @@ libc.workspace = true
 libz-sys.workspace = true
 md-5 = { version = "0.10.5", features = ["oid"] }
 md4 = "0.10.2"
+memchr = "2.7.4"
 node_resolver.workspace = true
 num-bigint.workspace = true
 num-bigint-dig = "0.8.2"

ext/node/benchmarks/child_process_ipc.mjs

@@ -5,10 +5,20 @@ import { setImmediate } from "node:timers";
 if (process.env.CHILD) {
   const len = +process.env.CHILD;
   const msg = ".".repeat(len);
+  let waiting = false;
   const send = () => {
-    while (process.send(msg));
+    while (
+      process.send(msg, undefined, undefined, (_e) => {
+        if (waiting) {
+          waiting = false;
+          setImmediate(send);
+        }
+      })
+    );
     // Wait: backlog of unsent messages exceeds threshold
-    setImmediate(send);
+    // once the message is sent, the callback will be called
+    // and we'll resume
+    waiting = true;
   };
   send();
 } else {

ext/node/lib.rs

@@ -30,6 +30,7 @@ pub use deno_package_json::PackageJson;
 pub use node_resolver::PathClean;
 pub use ops::ipc::ChildPipeFd;
 pub use ops::ipc::IpcJsonStreamResource;
+pub use ops::ipc::IpcRefTracker;
 use ops::vm;
 pub use ops::vm::create_v8_context;
 pub use ops::vm::init_global_template;
@@ -378,6 +379,8 @@ deno_core::extension!(deno_node,
     ops::ipc::op_node_child_ipc_pipe,
     ops::ipc::op_node_ipc_write,
     ops::ipc::op_node_ipc_read,
+    ops::ipc::op_node_ipc_ref,
+    ops::ipc::op_node_ipc_unref,
     ops::process::op_node_process_kill,
     ops::process::op_process_abort,
   ],