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, 116 insertions, 197 deletions

diff --git a/runtime/ops/process.rs b/runtime/ops/process.rs
index 69fb5cf29..9d166a801 100644
--- a/runtime/ops/process.rs
+++ b/runtime/ops/process.rs
@@ -154,6 +154,8 @@ pub struct SpawnArgs {
 
   #[serde(flatten)]
   stdio: ChildStdio,
+
+  extra_stdio: Vec<Stdio>,
 }
 
 #[derive(Deserialize)]
@@ -215,7 +217,12 @@ pub struct SpawnOutput {
   stderr: Option<ToJsBuffer>,
 }
 
-type CreateCommand = (std::process::Command, Option<ResourceId>);
+type CreateCommand = (
+  std::process::Command,
+  Option<ResourceId>,
+  Vec<Option<ResourceId>>,
+  Vec<deno_io::RawBiPipeHandle>,
+);
 
 fn create_command(
   state: &mut OpState,
@@ -277,216 +284,103 @@ fn create_command(
   // TODO(bartlomieju):
   #[allow(clippy::undocumented_unsafe_blocks)]
   unsafe {
+    let mut extra_pipe_rids = Vec::new();
+    let mut fds_to_dup = Vec::new();
+    let mut fds_to_close = Vec::new();
+    let mut ipc_rid = None;
     if let Some(ipc) = args.ipc {
-      if ipc < 0 {
-        return Ok((command, None));
-      }
-      // SockFlag is broken on macOS
-      // https://github.com/nix-rust/nix/issues/861
-      let mut fds = [-1, -1];
-      #[cfg(not(target_os = "macos"))]
-      let flags = libc::SOCK_CLOEXEC | libc::SOCK_NONBLOCK;
-
-      #[cfg(target_os = "macos")]
-      let flags = 0;
-
-      let ret = libc::socketpair(
-        libc::AF_UNIX,
-        libc::SOCK_STREAM | flags,
-        0,
-        fds.as_mut_ptr(),
-      );
-      if ret != 0 {
-        return Err(std::io::Error::last_os_error().into());
+      if ipc >= 0 {
+        let (ipc_fd1, ipc_fd2) = deno_io::bi_pipe_pair_raw()?;
+        fds_to_dup.push((ipc_fd2, ipc));
+        fds_to_close.push(ipc_fd2);
+        /* One end returned to parent process (this) */
+        let pipe_rid =
+          state
+            .resource_table
+            .add(deno_node::IpcJsonStreamResource::new(
+              ipc_fd1 as _,
+              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!("{}", ipc));
+        ipc_rid = Some(pipe_rid);
       }
+    }
 
-      if cfg!(target_os = "macos") {
-        let fcntl =
-          |fd: i32, flag: libc::c_int| -> Result<(), std::io::Error> {
-            let flags = libc::fcntl(fd, libc::F_GETFL, 0);
-
-            if flags == -1 {
-              return Err(fail(fds));
+    for (i, stdio) in args.extra_stdio.into_iter().enumerate() {
+      // index 0 in `extra_stdio` actually refers to fd 3
+      // because we handle stdin,stdout,stderr specially
+      let fd = (i + 3) as i32;
+      // TODO(nathanwhit): handle inherited, but this relies on the parent process having
+      // fds open already. since we don't generally support dealing with raw fds,
+      // we can't properly support this
+      if matches!(stdio, Stdio::Piped) {
+        let (fd1, fd2) = deno_io::bi_pipe_pair_raw()?;
+        fds_to_dup.push((fd2, fd));
+        fds_to_close.push(fd2);
+        let rid = state.resource_table.add(
+          match deno_io::BiPipeResource::from_raw_handle(fd1) {
+            Ok(v) => v,
+            Err(e) => {
+              log::warn!("Failed to open bidirectional pipe for fd {fd}: {e}");
+              extra_pipe_rids.push(None);
+              continue;
             }
-            let ret = libc::fcntl(fd, libc::F_SETFL, flags | flag);
-            if ret == -1 {
-              return Err(fail(fds));
-            }
-            Ok(())
-          };
-
-        fn fail(fds: [i32; 2]) -> std::io::Error {
-          unsafe {
-            libc::close(fds[0]);
-            libc::close(fds[1]);
-          }
-          std::io::Error::last_os_error()
-        }
-
-        // SOCK_NONBLOCK is not supported on macOS.
-        (fcntl)(fds[0], libc::O_NONBLOCK)?;
-        (fcntl)(fds[1], libc::O_NONBLOCK)?;
-
-        // SOCK_CLOEXEC is not supported on macOS.
-        (fcntl)(fds[0], libc::FD_CLOEXEC)?;
-        (fcntl)(fds[1], libc::FD_CLOEXEC)?;
+          },
+        );
+        extra_pipe_rids.push(Some(rid));
+      } else {
+        extra_pipe_rids.push(None);
       }
+    }
 
-      let fd1 = fds[0];
-      let fd2 = fds[1];
-
-      command.pre_exec(move || {
-        if ipc >= 0 {
-          let _fd = libc::dup2(fd2, ipc);
-          libc::close(fd2);
+    command.pre_exec(move || {
+      for &(src, dst) in &fds_to_dup {
+        if src >= 0 && dst >= 0 {
+          let _fd = libc::dup2(src, dst);
+          libc::close(src);
         }
-        libc::setgroups(0, std::ptr::null());
-        Ok(())
-      });
-
-      /* One end returned to parent process (this) */
-      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 NODE_CHANNEL_FD */
-      command.env("NODE_CHANNEL_FD", format!("{}", ipc));
-
-      return Ok((command, pipe_rid));
-    }
+      }
+      libc::setgroups(0, std::ptr::null());
+      Ok(())
+    });
 
-    Ok((command, None))
+    Ok((command, ipc_rid, extra_pipe_rids, fds_to_close))
   }
 
   #[cfg(windows)]
-  // Safety: We setup a windows named pipe and pass one end to the child process.
-  unsafe {
-    use windows_sys::Win32::Foundation::CloseHandle;
-    use windows_sys::Win32::Foundation::DuplicateHandle;
-    use windows_sys::Win32::Foundation::DUPLICATE_SAME_ACCESS;
-    use windows_sys::Win32::Foundation::ERROR_ACCESS_DENIED;
-    use windows_sys::Win32::Foundation::ERROR_PIPE_CONNECTED;
-    use windows_sys::Win32::Foundation::GENERIC_READ;
-    use windows_sys::Win32::Foundation::GENERIC_WRITE;
-    use windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE;
-    use windows_sys::Win32::Security::SECURITY_ATTRIBUTES;
-    use windows_sys::Win32::Storage::FileSystem::CreateFileW;
-    use windows_sys::Win32::Storage::FileSystem::FILE_FLAG_FIRST_PIPE_INSTANCE;
-    use windows_sys::Win32::Storage::FileSystem::FILE_FLAG_OVERLAPPED;
-    use windows_sys::Win32::Storage::FileSystem::OPEN_EXISTING;
-    use windows_sys::Win32::Storage::FileSystem::PIPE_ACCESS_DUPLEX;
-    use windows_sys::Win32::System::Pipes::ConnectNamedPipe;
-    use windows_sys::Win32::System::Pipes::CreateNamedPipeW;
-    use windows_sys::Win32::System::Pipes::PIPE_READMODE_BYTE;
-    use windows_sys::Win32::System::Pipes::PIPE_TYPE_BYTE;
-    use windows_sys::Win32::System::Threading::GetCurrentProcess;
-
-    use std::io;
-    use std::os::windows::ffi::OsStrExt;
-    use std::path::Path;
-    use std::ptr;
-
+  {
+    let mut ipc_rid = None;
+    let mut handles_to_close = Vec::with_capacity(1);
     if let Some(ipc) = args.ipc {
-      if ipc < 0 {
-        return Ok((command, None));
-      }
+      if ipc >= 0 {
+        let (hd1, hd2) = deno_io::bi_pipe_pair_raw()?;
 
-      let (path, hd1) = loop {
-        let name = format!("\\\\.\\pipe\\{}", uuid::Uuid::new_v4());
-        let mut path = Path::new(&name)
-          .as_os_str()
-          .encode_wide()
-          .collect::<Vec<_>>();
-        path.push(0);
-
-        let hd1 = CreateNamedPipeW(
-          path.as_ptr(),
-          PIPE_ACCESS_DUPLEX
-            | FILE_FLAG_FIRST_PIPE_INSTANCE
-            | FILE_FLAG_OVERLAPPED,
-          PIPE_TYPE_BYTE | PIPE_READMODE_BYTE,
-          1,
-          65536,
-          65536,
-          0,
-          std::ptr::null_mut(),
-        );
-
-        if hd1 == INVALID_HANDLE_VALUE {
-          let err = io::Error::last_os_error();
-          /* If the pipe name is already in use, try again. */
-          if err.raw_os_error() == Some(ERROR_ACCESS_DENIED as i32) {
-            continue;
-          }
+        /* One end returned to parent process (this) */
+        let pipe_rid = Some(state.resource_table.add(
+          deno_node::IpcJsonStreamResource::new(
+            hd1 as i64,
+            deno_node::IpcRefTracker::new(state.external_ops_tracker.clone()),
+          )?,
+        ));
 
-          return Err(err.into());
-        }
+        /* The other end passed to child process via NODE_CHANNEL_FD */
+        command.env("NODE_CHANNEL_FD", format!("{}", hd2 as i64));
 
-        break (path, hd1);
-      };
-
-      /* Create child pipe handle. */
-      let s = SECURITY_ATTRIBUTES {
-        nLength: std::mem::size_of::<SECURITY_ATTRIBUTES>() as u32,
-        lpSecurityDescriptor: ptr::null_mut(),
-        bInheritHandle: 1,
-      };
-      let mut hd2 = CreateFileW(
-        path.as_ptr(),
-        GENERIC_READ | GENERIC_WRITE,
-        0,
-        &s,
-        OPEN_EXISTING,
-        FILE_FLAG_OVERLAPPED,
-        0,
-      );
-      if hd2 == INVALID_HANDLE_VALUE {
-        return Err(io::Error::last_os_error().into());
-      }
+        handles_to_close.push(hd2);
 
-      // Will not block because we have create the pair.
-      if ConnectNamedPipe(hd1, ptr::null_mut()) == 0 {
-        let err = std::io::Error::last_os_error();
-        if err.raw_os_error() != Some(ERROR_PIPE_CONNECTED as i32) {
-          CloseHandle(hd2);
-          return Err(err.into());
-        }
+        ipc_rid = pipe_rid;
       }
+    }
 
-      // Duplicating the handle to allow the child process to use it.
-      if DuplicateHandle(
-        GetCurrentProcess(),
-        hd2,
-        GetCurrentProcess(),
-        &mut hd2,
-        0,
-        1,
-        DUPLICATE_SAME_ACCESS,
-      ) == 0
-      {
-        return Err(std::io::Error::last_os_error().into());
-      }
-
-      /* One end returned to parent process (this) */
-      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));
+    if args.extra_stdio.iter().any(|s| matches!(s, Stdio::Piped)) {
+      log::warn!(
+        "Additional stdio pipes beyond stdin/stdout/stderr are not currently supported on windows"
+      );
     }
-  }
 
-  #[cfg(not(unix))]
-  return Ok((command, None));
+    Ok((command, ipc_rid, vec![], handles_to_close))
+  }
 }
 
 #[derive(Serialize)]
@@ -497,13 +391,15 @@ struct Child {
   stdin_rid: Option<ResourceId>,
   stdout_rid: Option<ResourceId>,
   stderr_rid: Option<ResourceId>,
-  pipe_fd: Option<ResourceId>,
+  ipc_pipe_rid: Option<ResourceId>,
+  extra_pipe_rids: Vec<Option<ResourceId>>,
 }
 
 fn spawn_child(
   state: &mut OpState,
   command: std::process::Command,
-  pipe_fd: Option<ResourceId>,
+  ipc_pipe_rid: Option<ResourceId>,
+  extra_pipe_rids: Vec<Option<ResourceId>>,
 ) -> Result<Child, AnyError> {
   let mut command = tokio::process::Command::from(command);
   // TODO(@crowlkats): allow detaching processes.
@@ -585,10 +481,28 @@ fn spawn_child(
     stdin_rid,
     stdout_rid,
     stderr_rid,
-    pipe_fd,
+    ipc_pipe_rid,
+    extra_pipe_rids,
   })
 }
 
+fn close_raw_handle(handle: deno_io::RawBiPipeHandle) {
+  #[cfg(unix)]
+  {
+    // SAFETY: libc call
+    unsafe {
+      libc::close(handle);
+    }
+  }
+  #[cfg(windows)]
+  {
+    // SAFETY: win32 call
+    unsafe {
+      windows_sys::Win32::Foundation::CloseHandle(handle as _);
+    }
+  }
+}
+
 #[op2]
 #[serde]
 fn op_spawn_child(
@@ -596,8 +510,13 @@ fn op_spawn_child(
   #[serde] args: SpawnArgs,
   #[string] api_name: String,
 ) -> Result<Child, AnyError> {
-  let (command, pipe_rid) = create_command(state, args, &api_name)?;
-  spawn_child(state, command, pipe_rid)
+  let (command, pipe_rid, extra_pipe_rids, handles_to_close) =
+    create_command(state, args, &api_name)?;
+  let child = spawn_child(state, command, pipe_rid, extra_pipe_rids);
+  for handle in handles_to_close {
+    close_raw_handle(handle);
+  }
+  child
 }
 
 #[op2(async)]
@@ -626,7 +545,7 @@ fn op_spawn_sync(
 ) -> Result<SpawnOutput, AnyError> {
   let stdout = matches!(args.stdio.stdout, StdioOrRid::Stdio(Stdio::Piped));
   let stderr = matches!(args.stdio.stderr, StdioOrRid::Stdio(Stdio::Piped));
-  let (mut command, _) =
+  let (mut command, _, _, _) =
     create_command(state, args, "Deno.Command().outputSync()")?;
   let output = command.output().with_context(|| {
     format!(