fix(core): handle blocking stdin (#21672)

Cargo.lock

@@ -1536,6 +1536,7 @@ dependencies = [
  "ignore",
  "ignore-files 2.0.0",
  "itertools",
+ "mio",
  "napi",
  "napi-build",
  "napi-derive",

packages/nx/Cargo.toml

@@ -47,6 +47,9 @@ crossterm = "0.27.0"
 [target.'cfg(windows)'.dependencies]
 winapi = { version = "0.3", features = ["fileapi"] }
 
+[target.'cfg(not(windows))'.dependencies]
+mio = "0.8"
+
 [lib]
 crate-type = ['cdylib']
 

packages/nx/src/native/command.rs

@@ -14,8 +14,12 @@ use napi::{Env, JsFunction};
 use portable_pty::{ChildKiller, CommandBuilder, NativePtySystem, PtySize, PtySystem};
 use tracing::trace;
 
+#[cfg_attr(windows, path = "command/windows.rs")]
+#[cfg_attr(not(windows), path = "command/unix.rs")]
+mod os;
+
 fn command_builder() -> CommandBuilder {
-    if cfg!(target_os = "windows") {
+    if cfg!(windows) {
         let comspec = std::env::var("COMSPEC");
         let shell = comspec
             .as_ref()
@@ -191,12 +195,13 @@ pub fn run_command(
 
     // Stdin -> pty stdin
     if std::io::stdout().is_tty() {
-        std::thread::spawn(move || {
         enable_raw_mode().expect("Failed to enter raw terminal mode");
+        std::thread::spawn(move || {
             let mut stdin = std::io::stdin();
-            #[allow(clippy::redundant_pattern_matching)]
-            // ignore errors that come from copying the stream
-            if let Ok(_) = std::io::copy(&mut stdin, &mut writer) {}
+
+            if let Err(e) = os::write_to_pty(&mut stdin, &mut writer) {
+                trace!("Error writing to pty: {:?}", e);
+            }
         });
     }
 
@@ -224,7 +229,7 @@ pub fn nx_fork(
 ) -> napi::Result<ChildProcess> {
     let command = format!(
         "node {} {} {}",
-        handle_path_space(fork_script),
+        os::handle_path_space(fork_script),
         psuedo_ipc_path,
         id
     );
@@ -232,37 +237,3 @@ pub fn nx_fork(
     trace!("nx_fork command: {}", &command);
     run_command(command, command_dir, js_env, Some(quiet))
 }
-
-#[cfg(target_os = "windows")]
-pub fn handle_path_space(path: String) -> String {
-    use std::os::windows::ffi::OsStrExt;
-    use std::{ffi::OsString, os::windows::ffi::OsStringExt};
-
-    use winapi::um::fileapi::GetShortPathNameW;
-    let wide: Vec<u16> = std::path::PathBuf::from(&path)
-        .as_os_str()
-        .encode_wide()
-        .chain(Some(0))
-        .collect();
-    let mut buffer: Vec<u16> = vec![0; wide.len() * 2];
-    let result =
-        unsafe { GetShortPathNameW(wide.as_ptr(), buffer.as_mut_ptr(), buffer.len() as u32) };
-    if result == 0 {
-        path
-    } else {
-        let len = buffer.iter().position(|&x| x == 0).unwrap();
-        let short_path: String = OsString::from_wide(&buffer[..len])
-            .to_string_lossy()
-            .into_owned();
-        short_path
-    }
-}
-
-#[cfg(not(target_os = "windows"))]
-fn handle_path_space(path: String) -> String {
-    if path.contains(' ') {
-        format!("'{}'", path)
-    } else {
-        path
-    }
-}

packages/nx/src/native/command/unix.rs

@@ -0,0 +1,67 @@
+use std::{
+    io::{Read, Stdin, Write},
+    os::fd::AsRawFd,
+};
+
+use mio::{unix::SourceFd, Events};
+use tracing::trace;
+
+pub fn handle_path_space(path: String) -> String {
+    if path.contains(' ') {
+        format!("'{}'", path)
+    } else {
+        path
+    }
+}
+
+pub fn write_to_pty(stdin: &mut Stdin, writer: &mut impl Write) -> anyhow::Result<()> {
+    let mut buffer = [0; 1024];
+
+    let mut poll = mio::Poll::new()?;
+    let mut events = Events::with_capacity(3);
+
+    // Register stdin to the poll instance
+    let token = mio::Token(0);
+    poll.registry()
+        .register(
+            &mut SourceFd(&stdin.as_raw_fd()),
+            token,
+            mio::Interest::READABLE,
+        )
+        .map_err(|e| anyhow::anyhow!("Failed to register stdin to poll: {}", e))?;
+
+    loop {
+        // Poll for events
+        if let Err(e) = poll.poll(&mut events, None) {
+            if e.kind() == std::io::ErrorKind::Interrupted {
+                continue;
+            }
+            trace!("poll error: {:?}", e);
+            anyhow::bail!("Failed to poll for events: {}", e);
+        }
+
+        for event in events.iter().map(|x| x.token()) {
+            match event {
+                mio::Token(0) => {
+                    // Read data from stdin
+                    loop {
+                        match stdin.read(&mut buffer) {
+                            Ok(n) => {
+                                writer.write_all(&buffer[..n])?;
+                                writer.flush()?;
+                            }
+                            Err(e) => {
+                                if e.kind() == std::io::ErrorKind::WouldBlock {
+                                    break;
+                                } else if e.kind() == std::io::ErrorKind::Interrupted {
+                                    continue;
+                                }
+                            }
+                        }
+                    }
+                }
+                _ => unreachable!(),
+            }
+        }
+    }
+}

packages/nx/src/native/command/windows.rs

@@ -0,0 +1,31 @@
+use std::io::{Stdin, Write};
+use std::os::windows::ffi::OsStrExt;
+use std::{ffi::OsString, os::windows::ffi::OsStringExt};
+
+use winapi::um::fileapi::GetShortPathNameW;
+
+pub fn handle_path_space(path: String) -> String {
+    let wide: Vec<u16> = std::path::PathBuf::from(&path)
+        .as_os_str()
+        .encode_wide()
+        .chain(Some(0))
+        .collect();
+    let mut buffer: Vec<u16> = vec![0; wide.len() * 2];
+    let result =
+        unsafe { GetShortPathNameW(wide.as_ptr(), buffer.as_mut_ptr(), buffer.len() as u32) };
+    if result == 0 {
+        path
+    } else {
+        let len = buffer.iter().position(|&x| x == 0).unwrap();
+        let short_path: String = OsString::from_wide(&buffer[..len])
+            .to_string_lossy()
+            .into_owned();
+        short_path
+    }
+}
+
+pub fn write_to_pty(stdin: &mut Stdin, writer: &mut impl Write) -> anyhow::Result<()> {
+    std::io::copy(stdin, writer)
+        .map_err(|e| anyhow::anyhow!(e))
+        .map(|_| ())
+}