diff --git a/crates/ide-core/Cargo.toml b/crates/ide-core/Cargo.toml index 3a4f5f6cf..086aa84fd 100644 --- a/crates/ide-core/Cargo.toml +++ b/crates/ide-core/Cargo.toml @@ -22,4 +22,4 @@ tempfile.workspace = true libc = "0.2" [target.'cfg(windows)'.dependencies] -windows-sys = { version = "0.61", features = ["Win32_Globalization", "Win32_Security", "Win32_System_Registry", "Win32_Foundation", "Win32_System_Diagnostics_ToolHelp", "Win32_System_Threading"] } +windows-sys = { version = "0.61", features = ["Win32_Globalization", "Win32_Security", "Win32_System_Registry", "Win32_Foundation", "Win32_System_Diagnostics_ToolHelp", "Win32_System_Threading", "Win32_System_IO"] } diff --git a/crates/ide-core/src/input.rs b/crates/ide-core/src/input.rs new file mode 100644 index 000000000..7d4a24961 --- /dev/null +++ b/crates/ide-core/src/input.rs @@ -0,0 +1,285 @@ +//! Bounded, ordered process input. Only the worker touches the OS writer. + +use std::collections::VecDeque; +use std::io::{self, Write}; +use std::sync::{Arc, Condvar, Mutex}; +use std::thread::JoinHandle; +use std::time::Duration; + +const MAX_BYTES: usize = 4 * 1024 * 1024; +const MAX_WRITES: usize = 1024; +const RETRY: Duration = Duration::from_millis(4); + +#[derive(Default)] +struct State { + queue: VecDeque>, + bytes: usize, + finished: bool, + aborted: bool, + error: Option, +} + +/// One worker per pipe/PTY. The byte budget includes the in-flight write. +/// Rejected writes enqueue nothing. Dropping this handle discards pending input +/// without joining the worker; `finish` instead drains accepted input before EOF. +pub struct InputWriter { + shared: Arc<(Mutex, Condvar)>, + worker: Option>, +} + +impl InputWriter { + /// Unix writers must first be made nonblocking with `nonblocking`. + pub fn new( + name: String, + mut writer: Box, + on_error: impl FnOnce(io::Error) + Send + 'static, + ) -> io::Result { + let shared = Arc::new((Mutex::new(State::default()), Condvar::new())); + let worker_shared = shared.clone(); + let worker = std::thread::Builder::new().name(name).spawn(move || { + let (lock, wake) = &*worker_shared; + let result = (|| { + loop { + let data = { + let mut state = lock.lock().unwrap(); + while state.queue.is_empty() && !state.finished && !state.aborted { + state = wake.wait(state).unwrap(); + } + if state.aborted { + return Ok(()); + } + let Some(data) = state.queue.pop_front() else { return Ok(()) }; + data + }; + let mut offset = 0; + loop { + if lock.lock().unwrap().aborted { + return Ok(()); + } + let flushing = offset == data.len(); + let result = if !flushing { + writer.write(&data[offset..]).and_then(|n| { + if n == 0 { + Err(io::ErrorKind::WriteZero.into()) + } else { + offset += n; + Ok(()) + } + }) + } else { + writer.flush() + }; + match result { + Ok(()) if flushing => break, + Ok(()) => {} + Err(e) if e.kind() == io::ErrorKind::Interrupted => continue, + Err(e) if e.kind() == io::ErrorKind::WouldBlock => { + let state = lock.lock().unwrap(); + if state.aborted { + return Ok(()); + } + drop(wake.wait_timeout(state, RETRY).unwrap()); + } + Err(e) => return Err(e), + } + } + let len = data.len(); + drop(data); + let mut state = lock.lock().unwrap(); + state.bytes = state.bytes.saturating_sub(len); + } + })(); + if let Err(e) = result { + let mut state = lock.lock().unwrap(); + if state.aborted { + return; + } + state.error = Some(e.to_string()); + state.queue.clear(); + state.bytes = 0; + drop(state); + on_error(e); + } + })?; + Ok(Self { shared, worker: Some(worker) }) + } + + pub fn write(&self, data: &[u8]) -> io::Result<()> { + let (lock, wake) = &*self.shared; + let mut state = lock.lock().unwrap(); + if let Some(error) = &state.error { + return Err(io::Error::new(io::ErrorKind::BrokenPipe, error.clone())); + } + if state.finished || state.aborted { + return Err(io::Error::new(io::ErrorKind::BrokenPipe, "stdin is closed")); + } + if data.is_empty() { + return Ok(()); + } + if data.len() > MAX_BYTES.saturating_sub(state.bytes) || state.queue.len() >= MAX_WRITES { + return Err(io::Error::new( + io::ErrorKind::WouldBlock, + "Input queue is full (4 MiB or 1024 writes); input was not sent.", + )); + } + state.bytes += data.len(); + state.queue.push_back(data.to_vec()); + wake.notify_one(); + Ok(()) + } + + /// Send EOF after the accepted writes, without waiting for the child to read. + pub fn finish(&self) { + let (lock, wake) = &*self.shared; + lock.lock().unwrap().finished = true; + wake.notify_one(); + } + + pub fn close(&self) { + let (lock, wake) = &*self.shared; + let mut state = lock.lock().unwrap(); + state.aborted = true; + state.queue.clear(); + wake.notify_one(); + } +} + +impl Drop for InputWriter { + fn drop(&mut self) { + self.close(); + let Some(worker) = self.worker.take() else { return }; + #[cfg(windows)] + { + use std::os::windows::io::AsRawHandle; + // A cancellation can race with entry into WriteFile, so retry off + // the caller's thread until the worker observes the closed queue. + let _ = std::thread::Builder::new().name("openidea-input-cancel".into()).spawn(move || { + while !worker.is_finished() { + unsafe { + windows_sys::Win32::System::IO::CancelSynchronousIo(worker.as_raw_handle()); + } + std::thread::sleep(RETRY); + } + let _ = worker.join(); + }); + } + #[cfg(not(windows))] + drop(worker); + } +} + +/// PTY clones share file status flags, so readers must also handle WouldBlock. +#[cfg(unix)] +pub fn nonblocking(fd: std::os::fd::RawFd) -> io::Result<()> { + let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) }; + if flags == -1 || unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) } == -1 { + return Err(io::Error::last_os_error()); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::mpsc; + #[cfg(unix)] + use std::time::Instant; + + struct SlowWriter { + ready: Arc, + output: Arc>>, + dropped: mpsc::Sender<()>, + } + + impl Write for SlowWriter { + fn write(&mut self, bytes: &[u8]) -> io::Result { + if !self.ready.load(Ordering::Acquire) { + return Err(io::ErrorKind::WouldBlock.into()); + } + let n = bytes.len().min(7); + self.output.lock().unwrap().extend_from_slice(&bytes[..n]); + Ok(n) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + + impl Drop for SlowWriter { + fn drop(&mut self) { + let _ = self.dropped.send(()); + } + } + + #[test] + fn bounds_include_inflight_bytes_and_reject_whole_writes() { + let ready = Arc::new(AtomicBool::new(false)); + let output = Arc::new(Mutex::new(Vec::new())); + let (tx, rx) = mpsc::channel(); + let writer = InputWriter::new( + "test-input".into(), + Box::new(SlowWriter { ready: ready.clone(), output: output.clone(), dropped: tx }), + |e| panic!("{e}"), + ) + .unwrap(); + let bytes = vec![b'x'; MAX_BYTES]; + writer.write(&bytes).unwrap(); + std::thread::sleep(RETRY * 2); + assert_eq!(writer.write(b"rejected").unwrap_err().kind(), io::ErrorKind::WouldBlock); + writer.close(); + rx.recv_timeout(Duration::from_secs(1)).expect("worker stops while peer still refuses input"); + assert!(output.lock().unwrap().is_empty()); + assert!(writer.write(b"closed").is_err()); + assert!(writer.shared.0.lock().unwrap().queue.is_empty()); + } + + #[test] + fn partial_writes_preserve_order_and_finish_drains_before_eof() { + let ready = Arc::new(AtomicBool::new(false)); + let output = Arc::new(Mutex::new(Vec::new())); + let (tx, rx) = mpsc::channel(); + let writer = InputWriter::new( + "test-input".into(), + Box::new(SlowWriter { ready: ready.clone(), output: output.clone(), dropped: tx }), + |e| panic!("{e}"), + ) + .unwrap(); + let paste = "abé".repeat(1024); + writer.write(paste.as_bytes()).unwrap(); + writer.write(b"tail").unwrap(); + writer.finish(); + assert!(writer.write(b"after EOF").is_err()); + ready.store(true, Ordering::Release); + rx.recv_timeout(Duration::from_secs(1)).unwrap(); + assert_eq!(*output.lock().unwrap(), format!("{paste}tail").as_bytes()); + } + + #[test] + fn tiny_writes_are_also_bounded() { + let shared = Arc::new((Mutex::new(State::default()), Condvar::new())); + let writer = InputWriter { shared, worker: None }; + for _ in 0..MAX_WRITES { + writer.write(b"x").unwrap(); + } + assert_eq!(writer.write(b"x").unwrap_err().kind(), io::ErrorKind::WouldBlock); + } + + #[cfg(unix)] + #[test] + fn close_interrupts_a_full_os_pipe_while_the_peer_stays_open() { + use std::os::fd::AsRawFd; + use std::os::unix::net::UnixStream; + let (pipe, _peer) = UnixStream::pair().unwrap(); + nonblocking(pipe.as_raw_fd()).unwrap(); + let mut writer = InputWriter::new("test-pipe-input".into(), Box::new(pipe), |e| panic!("{e}")).unwrap(); + writer.write(&vec![b'x'; MAX_BYTES]).unwrap(); + std::thread::sleep(Duration::from_millis(50)); + assert!(!writer.worker.as_ref().unwrap().is_finished()); + let started = Instant::now(); + writer.close(); + writer.worker.take().unwrap().join().unwrap(); + assert!(started.elapsed() < Duration::from_millis(250)); + } +} diff --git a/crates/ide-core/src/lib.rs b/crates/ide-core/src/lib.rs index 6234af032..05c3d5d87 100644 --- a/crates/ide-core/src/lib.rs +++ b/crates/ide-core/src/lib.rs @@ -10,6 +10,7 @@ pub mod event; pub mod go; pub mod ids; pub mod ignored; +pub mod input; pub mod inspection_settings; pub mod ipc_path; pub mod java_runtime; diff --git a/crates/ide-dap/src/session.rs b/crates/ide-dap/src/session.rs index fe2d6f8ec..b093ce134 100644 --- a/crates/ide-dap/src/session.rs +++ b/crates/ide-dap/src/session.rs @@ -600,15 +600,14 @@ fn on_reverse_request(mgr: &Arc, s: &Arc, seq: i64, comma /// A `runInTerminal` program and the pseudo-terminal it runs in. pub(crate) struct ProgramTerminal { - writer: Box, + writer: ide_core::input::InputWriter, child: Box, _master: Box, } impl ProgramTerminal { pub(crate) fn write(&mut self, text: &str) -> std::io::Result<()> { - self.writer.write_all(text.as_bytes())?; - self.writer.flush() + self.writer.write(text.as_bytes()) } } @@ -628,6 +627,10 @@ pub(crate) fn run_in_terminal(s: &Arc, args: &Value) -> Result, args: &Value) -> Result = Vec::new(); loop { match reader.read(&mut buf) { - Ok(0) | Err(_) => break, + Ok(0) => break, Ok(n) => pending.extend_from_slice(&buf[..n]), + Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue, + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { + std::thread::sleep(std::time::Duration::from_millis(4)); + continue; + } + #[cfg(unix)] + Err(e) if e.raw_os_error() == Some(libc::EIO) => break, + Err(e) => { + tracing::warn!(%session, "debug terminal read: {e}"); + break; + } } // Keep incomplete UTF-8 sequences, and a `\r` that may start a `\r\n`, for the next chunk. let mut valid = match std::str::from_utf8(&pending) { diff --git a/crates/ide-dap/src/tests.rs b/crates/ide-dap/src/tests.rs index c0ff9304f..6a989b968 100644 --- a/crates/ide-dap/src/tests.rs +++ b/crates/ide-dap/src/tests.rs @@ -554,6 +554,22 @@ async fn run_in_terminal_program_reads_typed_input() { assert!(h.manager.send_input(session, "x\n").is_err()); } +#[cfg(unix)] +#[tokio::test] +async fn blocked_program_input_does_not_hold_up_shutdown() { + let h = harness(Script { + run_in_terminal: Some(vec!["sh".into(), "-c".into(), "stty raw -echo; printf READY; exec sleep 60".into()]), + ..Script::default() + }); + let session = h.start(config()).await.unwrap(); + h.wait_event(|e| matches!(e, DebugEvent::Output { output, .. } if output.contains("READY"))).await; + let started = std::time::Instant::now(); + h.manager.send_input(session, &"x".repeat(512 * 1024)).unwrap(); + assert!(started.elapsed() < std::time::Duration::from_millis(250)); + h.manager.stop(session).await.unwrap(); + assert!(h.manager.send_input(session, "later").is_err()); +} + #[test] fn launch_arguments_merge_and_defaults() { let mut c = config(); diff --git a/crates/ide-run/src/process.rs b/crates/ide-run/src/process.rs index 9ffa51275..3c955eadc 100644 --- a/crates/ide-run/src/process.rs +++ b/crates/ide-run/src/process.rs @@ -51,7 +51,7 @@ struct Entry { config_id: String, name: String, pid: Mutex>, - stdin: Mutex>>, + stdin: Mutex>, /// The PTY master of a terminal-emulated run (resize; dropped on exit). pty: Option>>>, stop: Option, @@ -422,10 +422,23 @@ impl ProcessManager { })?; let id = RunId::next(); - let stdin: Option> = match child.stdin.take() { - Some(pipe) if !redirected => Some(Box::new(pipe)), - _ => None, - }; + let stdin = (|| -> std::io::Result<_> { + Ok(match child.stdin.take() { + Some(pipe) if !redirected => { + #[cfg(unix)] + { + use std::os::fd::AsRawFd; + ide_core::input::nonblocking(pipe.as_raw_fd())?; + } + Some(input_writer(id, Box::new(pipe))?) + } + _ => None, + }) + })() + .inspect_err(|_| { + let _ = child.kill(); + let _ = child.wait(); + })?; let entry = Arc::new(Entry { project_root: root.to_path_buf(), id, @@ -524,6 +537,10 @@ impl ProcessManager { for (k, v) in &env.vars { cmd.env(k, v); } + #[cfg(unix)] + if let Some(fd) = pair.master.as_raw_fd() { + ide_core::input::nonblocking(fd)?; + } let mut child = pair.slave.spawn_command(cmd).map_err(|e| { let text = e.to_string(); if text.contains("No such file") || text.contains("not found") { @@ -538,6 +555,10 @@ impl ProcessManager { let writer = pair.master.take_writer().map_err(|e| IdeError::internal(format!("pty writer: {e}")))?; let id = RunId::next(); + let writer = input_writer(id, writer).inspect_err(|_| { + let _ = child.kill(); + let _ = child.wait(); + })?; let entry = Arc::new(Entry { project_root: root.to_path_buf(), id, @@ -609,16 +630,9 @@ impl ProcessManager { if entry.exited.load(Ordering::Acquire) { return Err(IdeError::invalid("process has exited")); } - let mut stdin = entry.stdin.lock(); - let pipe = stdin.as_mut().ok_or_else(|| IdeError::invalid("stdin is closed"))?; - let res = pipe.write_all(data.as_bytes()).and_then(|_| pipe.flush()); - if let Err(e) = res { - if e.kind() == std::io::ErrorKind::BrokenPipe { - *stdin = None; - } - return Err(e.into()); - } - Ok(()) + let stdin = entry.stdin.lock(); + let pipe = stdin.as_ref().ok_or_else(|| IdeError::invalid("stdin is closed"))?; + pipe.write(data.as_bytes()).map_err(Into::into) } /// Closes the process's stdin (sends EOF; on a terminal, types ^D). @@ -627,7 +641,9 @@ impl ProcessManager { if entry.pty.is_some() { return self.write_stdin(id, "\u{4}"); } - entry.stdin.lock().take(); + if let Some(stdin) = entry.stdin.lock().as_ref() { + stdin.finish(); + } Ok(()) } @@ -769,7 +785,23 @@ fn run_followup( (entry.output)(OutputChunk::System { text: format!("{}\n", inv.display) }); let mut child = cmd.spawn()?; *pid = Some(child.id()); - *entry.stdin.lock() = child.stdin.take().map(|p| Box::new(p) as Box); + *entry.stdin.lock() = (|| -> std::io::Result<_> { + Ok(match child.stdin.take() { + Some(pipe) => { + #[cfg(unix)] + { + use std::os::fd::AsRawFd; + ide_core::input::nonblocking(pipe.as_raw_fd())?; + } + Some(input_writer(entry.id, Box::new(pipe))?) + } + None => None, + }) + })() + .inspect_err(|_| { + let _ = child.kill(); + let _ = child.wait(); + })?; drop(pid); let readers = (|| { if let Some(out) = child.stdout.take() { @@ -963,7 +995,13 @@ fn spawn_reader( } } Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue, - Err(_) => break, + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => std::thread::sleep(BATCH_INTERVAL), + #[cfg(unix)] + Err(e) if e.raw_os_error() == Some(libc::EIO) => break, + Err(e) => { + tracing::warn!(%id, "run read: {e}"); + break; + } } } })?; @@ -1180,6 +1218,12 @@ fn signal_number(_description: &str) -> Option { None } +fn input_writer(id: RunId, writer: Box) -> std::io::Result { + ide_core::input::InputWriter::new(format!("openidea-run-input-{}", id.0), writer, move |e| { + tracing::warn!(%id, "run input: {e}"); + }) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/ide-run/tests/process.rs b/crates/ide-run/tests/process.rs index d2c9caa6d..ec475c246 100644 --- a/crates/ide-run/tests/process.rs +++ b/crates/ide-run/tests/process.rs @@ -150,6 +150,24 @@ fn stdin_echo_with_cat() { assert_eq!(wait_exit(&h, id, Duration::from_secs(10)), Some(0)); } +#[test] +fn full_stdin_does_not_block_input_or_stop_for_pipes_and_ptys() { + for pty in [false, true] { + let h = harness(); + let mut config = + cfg("t:blocked-input", "/bin/sh", &["-c", "stty raw -echo 2>/dev/null; printf READY; exec sleep 60"]); + config.extras.emulate_terminal = pty; + let (id, chunks) = start(&h, &config); + wait_for(&chunks, "READY"); + let started = Instant::now(); + h.pm.write_stdin(id, &"x".repeat(512 * 1024)).unwrap(); + assert!(started.elapsed() < Duration::from_millis(250)); + h.pm.stop(id).unwrap(); + h.pm.stop(id).unwrap(); + wait_exit(&h, id, Duration::from_secs(10)); + } +} + #[test] fn missing_command_and_cwd_are_errors() { let h = harness(); diff --git a/crates/ide-terminal/src/lib.rs b/crates/ide-terminal/src/lib.rs index 098ffcab8..d5c70ffa8 100644 --- a/crates/ide-terminal/src/lib.rs +++ b/crates/ide-terminal/src/lib.rs @@ -11,7 +11,7 @@ use portable_pty::{ChildKiller, CommandBuilder, MasterPty, PtySize, native_pty_s use serde::{Deserialize, Serialize}; use specta::Type; use std::collections::{BTreeMap, HashMap}; -use std::io::{Read, Write}; +use std::io::Read; use std::path::{Path, PathBuf}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; @@ -56,8 +56,9 @@ struct Session { /// The window that opened it: closing that window's project closes its terminals. key: String, info: TerminalInfo, + // Cancel input before dropping the master (ClosePseudoConsole on Windows). + writer: ide_core::input::InputWriter, master: Mutex>, - writer: Mutex>, killer: Mutex>, exited: Arc, started: Instant, @@ -161,7 +162,12 @@ impl TerminalManager { } apply_powershell_env(&shell, &mut cmd); - let child = pair + #[cfg(unix)] + if let Some(fd) = pair.master.as_raw_fd() { + ide_core::input::nonblocking(fd)?; + } + + let mut child = pair .slave .spawn_command(cmd) .map_err(|e| IdeError::unavailable(format!("failed to start shell `{shell}`: {e}")))?; @@ -174,11 +180,18 @@ impl TerminalManager { let id = TerminalId::next(); let info = TerminalInfo { id, title: shell_title(&shell), pid: child.process_id() }; let exited = Arc::new(AtomicBool::new(false)); + let writer = ide_core::input::InputWriter::new(format!("openidea-term-write-{}", id.0), writer, move |e| { + tracing::warn!(%id, "terminal input: {e}"); + }) + .inspect_err(|_| { + let _ = child.kill(); + let _ = child.wait(); + })?; let session = Arc::new(Session { key, info: info.clone(), master: Mutex::new(pair.master), - writer: Mutex::new(writer), + writer, killer: Mutex::new(child.clone_killer()), exited: exited.clone(), started: Instant::now(), @@ -209,9 +222,7 @@ impl TerminalManager { let ms = session.started.elapsed().as_millis(); tracing::info!(%id, ms, bytes = data.len(), cursor_report, "terminal first input"); } - let mut w = session.writer.lock(); - w.write_all(data.as_bytes())?; - w.flush()?; + session.writer.write(data.as_bytes())?; Ok(()) } @@ -364,6 +375,7 @@ mod win { /// Hangs up the shell (and its process group), escalating to SIGKILL if it lingers. fn terminate(session: &Session) { + session.writer.close(); if session.exited.load(Ordering::Acquire) { return; } @@ -407,8 +419,14 @@ fn spawn_reader(id: TerminalId, mut reader: Box, tx: mpsc::Send } } Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue, + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => std::thread::sleep(BATCH_INTERVAL), // EIO is how Linux/macOS report "slave side closed". - Err(_) => break, + #[cfg(unix)] + Err(e) if e.raw_os_error() == Some(libc::EIO) => break, + Err(e) => { + tracing::warn!(%id, "terminal read: {e}"); + break; + } } } })?; @@ -809,6 +827,40 @@ mod tests { all } + #[cfg(unix)] + #[test] + fn blocked_input_returns_promptly_and_kill_does_not_wait_for_it() { + let h = harness(); + let (info, output) = spawn_collect(&h, sh("stty raw -echo; printf READY; exec sleep 60")); + read_until(&output, "READY"); + let started = Instant::now(); + h.manager.write(info.id, &"x".repeat(512 * 1024)).unwrap(); + h.manager.write(info.id, "\u{3}").unwrap(); + assert!(started.elapsed() < Duration::from_millis(250)); + std::thread::sleep(Duration::from_millis(50)); + let started = Instant::now(); + h.manager.kill(info.id).unwrap(); + assert!(started.elapsed() < Duration::from_millis(250)); + wait_exit(&h, info.id); + assert!(h.manager.write(info.id, "later").is_err()); + } + + #[cfg(unix)] + #[test] + fn input_keeps_large_pastes_and_following_keystrokes_in_order() { + let h = harness(); + let (info, output) = spawn_collect(&h, sh("stty raw -echo; printf READY; sleep 1; exec cat")); + read_until(&output, "READY"); + let paste = "abé".repeat(128 * 1024); + let started = Instant::now(); + h.manager.write(info.id, &paste).unwrap(); + h.manager.write(info.id, "END-OF-INPUT").unwrap(); + assert!(started.elapsed() < Duration::from_millis(250)); + let actual = read_until(&output, "END-OF-INPUT"); + h.manager.kill(info.id).unwrap(); + assert_eq!(actual, format!("{paste}END-OF-INPUT")); + } + #[test] fn program_sessions_never_fall_back_to_another_executable_or_cwd() { let h = harness(); diff --git a/src/features/run/store.ts b/src/features/run/store.ts index 47f366586..7e393e8df 100644 --- a/src/features/run/store.ts +++ b/src/features/run/store.ts @@ -1257,7 +1257,7 @@ export async function sendStdin(sessionId: string, text: string): Promise } catch (e) { enqueue(sessionId, { kind: "system", - text: tx`Cannot write to stdin: ${e instanceof Error ? e.message : String(e)}\n`, + text: tx`Cannot write to stdin: ${e instanceof Error ? tx(e.message) : String(e)}\n`, }); } } @@ -1266,7 +1266,12 @@ export async function sendStdin(sessionId: string, text: string): Promise export function writeTerminal(sessionId: string, data: string) { const session = getSession(sessionId); if (!session || session.runId === null || session.status !== "running") return; - void call(commands.runStdin(session.runId, data)).catch(() => undefined); + void call(commands.runStdin(session.runId, data)).catch((e) => { + enqueue(sessionId, { + kind: "system", + text: tx`Cannot write to stdin: ${e instanceof Error ? tx(e.message) : String(e)}\n`, + }); + }); } /** The terminal console's size in cells; the process's terminal follows it. */ diff --git a/src/features/terminal/session.test.ts b/src/features/terminal/session.test.ts index 5daae0408..6bec93120 100644 --- a/src/features/terminal/session.test.ts +++ b/src/features/terminal/session.test.ts @@ -16,6 +16,7 @@ const ipc = vi.hoisted(() => ({ termKill: vi.fn(), termResize: vi.fn(), termIsBusy: vi.fn(), + termWrite: vi.fn(), }, })); const workbench = vi.hoisted(() => ({ @@ -122,6 +123,27 @@ describe("resolveStartCwd", () => { describe("terminal start directory fallback", () => { beforeEach(() => vi.clearAllMocks()); + it("shows rejected terminal input instead of silently discarding it", async () => { + ipc.commands.termSpawn.mockResolvedValue({ id: 9, title: "zsh" }); + ipc.commands.termWrite.mockRejectedValueOnce(new Error("input queue full")); + const session = new TerminalSession({ + key: "s-input-error", + cwd: "/repo", + onChange: () => undefined, + onFocus: () => undefined, + }); + await session.start(); + session.input("paste"); + await vi.waitFor(() => + expect(workbench.notify).toHaveBeenCalledWith({ + level: "error", + title: "Cannot write to terminal", + body: "input queue full", + }), + ); + session.dispose(); + }); + it("spawns a restored session in the project root when its saved directory is gone", async () => { ipc.commands.fsEntryKind.mockRejectedValue(new Error("gone")); ipc.commands.termSpawn.mockResolvedValue({ id: 7, title: "zsh" }); diff --git a/src/features/terminal/session.ts b/src/features/terminal/session.ts index 7c65ce958..8e61ab329 100644 --- a/src/features/terminal/session.ts +++ b/src/features/terminal/session.ts @@ -9,6 +9,7 @@ import { SearchAddon, type ISearchOptions } from "@xterm/addon-search"; import "@xterm/xterm/css/xterm.css"; import { Channel } from "@tauri-apps/api/core"; import { call, commands, isTauri, type TerminalId } from "@/platform/ipc"; +import { tx } from "@/platform/i18n"; import { editorService, notifications, registerFontZoomSurface } from "@/workbench/contributions"; import { fitScheduler, settledResize } from "./fit"; import { installTerminalDrop } from "./drop"; @@ -658,14 +659,20 @@ export class TerminalSession { this.pendingInput += data; return; } - void commands.termWrite(this.ptyId, data); + void call(commands.termWrite(this.ptyId, data)).catch((error) => { + notifications().notify({ + level: "error", + title: tx("Cannot write to terminal"), + body: error instanceof Error ? tx(error.message) : String(error), + }); + }); } /** Sends keystrokes buffered while the PTY was starting. */ private flushPendingInput() { const queued = this.pendingInput; this.pendingInput = ""; - if (queued && this.ptyId !== null) void commands.termWrite(this.ptyId, queued); + if (queued && this.ptyId !== null) this.input(queued); } /** Writes input immediately (shortcuts translated outside xterm). */ diff --git a/src/platform/locales/text/en.json b/src/platform/locales/text/en.json index c5d4989a1..dd6aee814 100644 --- a/src/platform/locales/text/en.json +++ b/src/platform/locales/text/en.json @@ -1,4 +1,6 @@ { + "Input queue is full (4 MiB or 1024 writes); input was not sent.": "Input queue is full (4 MiB or 1024 writes); input was not sent.", + "Cannot write to terminal": "Cannot write to terminal", "Docker": "Docker", "Shell": "Shell", "C/C++": "C/C++", diff --git a/src/platform/locales/text/zh.json b/src/platform/locales/text/zh.json index 095f90431..7b3531908 100644 --- a/src/platform/locales/text/zh.json +++ b/src/platform/locales/text/zh.json @@ -1,4 +1,6 @@ { + "Input queue is full (4 MiB or 1024 writes); input was not sent.": "输入队列已满（4 MiB 或 1024 次写入）；本次输入未发送。", + "Cannot write to terminal": "无法写入终端", "Docker": "Docker", "Shell": "Shell", "C/C++": "C/C++",