diff --git a/crates/splice-pty/src/lib.rs b/crates/splice-pty/src/lib.rs index b267b0b..0714f72 100644 --- a/crates/splice-pty/src/lib.rs +++ b/crates/splice-pty/src/lib.rs @@ -215,7 +215,6 @@ pub struct PtySession { writer: std::sync::Mutex>>, killer: std::sync::Mutex>, pid: Option, - completed: std::sync::Mutex>, reader: std::sync::Mutex>>, waiter: std::sync::Mutex>>, running: std::sync::Arc, @@ -344,16 +343,20 @@ impl PtySession { let id = lifecycle.id(); let reader_lifecycle = std::sync::Arc::clone(&lifecycle); - let reader = - std::thread::spawn(move || read_output(reader, id, reader_lifecycle, on_output)); + let (reader_completed_tx, reader_completed) = std::sync::mpsc::channel(); + let reader = std::thread::spawn(move || { + read_output(reader, id, reader_lifecycle, on_output); + let _ = reader_completed_tx.send(()); + }); let waiter_lifecycle = std::sync::Arc::clone(&lifecycle); let waiter_running = std::sync::Arc::clone(&running); - let (completed_tx, completed) = std::sync::mpsc::channel(); let waiter = std::thread::spawn(move || { let exited = child.wait().is_ok(); waiter_running.store(false, std::sync::atomic::Ordering::SeqCst); - let _ = completed_tx.send(()); + // A Unix PTY returns EOF/EIO only after the child releases its slave + // side. Drain that final output before observers see natural exit. + let _ = reader_completed.recv_timeout(std::time::Duration::from_millis(250)); if exited && waiter_lifecycle.should_emit_natural_exit() { on_exit(waiter_lifecycle.id()); } @@ -365,7 +368,6 @@ impl PtySession { writer: std::sync::Mutex::new(Some(writer)), killer: std::sync::Mutex::new(killer), pid, - completed: std::sync::Mutex::new(completed), reader: std::sync::Mutex::new(Some(reader)), waiter: std::sync::Mutex::new(Some(waiter)), running, @@ -395,11 +397,28 @@ impl PtySession { } pub fn interrupt(&self) -> Result<(), PtyError> { - self.write("\u{3}") + if !self.lifecycle.should_emit_natural_exit() { + return Err(PtyError::SessionClosed); + } + match self.writer.try_lock() { + Ok(mut writer) => { + let writer = writer.as_mut().ok_or(PtyError::SessionClosed)?; + writer.write_all(b"\x03")?; + writer.flush()?; + Ok(()) + } + Err(std::sync::TryLockError::WouldBlock) if self.signal_process_group(libc::SIGINT) => { + Ok(()) + } + Err(_) => Err(PtyError::SessionClosed), + } } pub fn resize(&self, size: TerminalSize) -> Result<(), PtyError> { use portable_pty::PtySize; + if !self.lifecycle.should_emit_natural_exit() { + return Err(PtyError::SessionClosed); + } self.master .lock() .map_err(|_| PtyError::SessionClosed)? @@ -427,19 +446,17 @@ impl PtySession { pub fn close(&self) { if self.lifecycle.begin_close() { (self.on_closing)(); - if let Ok(mut killer) = self.killer.lock() { - let _ = killer.kill(); + for signal in [libc::SIGHUP, libc::SIGTERM, libc::SIGKILL] { + let _ = self.signal_process_group(signal); + if self.wait_for_teardown(std::time::Duration::from_millis(100)) { + break; + } } - let completed = self.completed.lock().is_ok_and(|completed| { - completed - .recv_timeout(std::time::Duration::from_millis(250)) - .is_ok() - }); - if !completed { - if let Some(pid) = self.pid { - // Root-only escalation; process-group teardown belongs to PR4. - unsafe { libc::kill(pid as libc::pid_t, libc::SIGKILL) }; + if !self.teardown_complete() { + if let Ok(mut killer) = self.killer.lock() { + let _ = killer.kill(); } + let _ = self.wait_for_teardown(std::time::Duration::from_millis(250)); } if let Ok(mut writer) = self.writer.lock() { writer.take(); @@ -453,6 +470,33 @@ impl PtySession { } } } + + fn signal_process_group(&self, signal: libc::c_int) -> bool { + self.pid.is_some_and(|pid| unsafe { + libc::kill(-(pid as libc::pid_t), signal) == 0 + || (self.running.load(std::sync::atomic::Ordering::SeqCst) + && libc::kill(pid as libc::pid_t, signal) == 0) + }) + } + + fn teardown_complete(&self) -> bool { + !self.process_group_exists() && !self.running.load(std::sync::atomic::Ordering::SeqCst) + } + + fn process_group_exists(&self) -> bool { + self.pid.is_some_and(|pid| unsafe { + libc::kill(-(pid as libc::pid_t), 0) == 0 + || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) + }) + } + + fn wait_for_teardown(&self, timeout: std::time::Duration) -> bool { + let deadline = std::time::Instant::now() + timeout; + while !self.teardown_complete() && std::time::Instant::now() < deadline { + std::thread::sleep(std::time::Duration::from_millis(10)); + } + self.teardown_complete() + } } #[cfg(unix)] @@ -543,8 +587,12 @@ fn read_output( let mut bytes = [0; 4096]; let mut pending = Vec::new(); while lifecycle.should_emit_natural_exit() { - let Ok(count) = reader.read(&mut bytes) else { - break; + let count = match reader.read(&mut bytes) { + Ok(count) => count, + // Linux PTYs report EIO once their slave side closes. It is EOF, + // not a session error, and still needs the buffered UTF-8 flush. + Err(error) if error.raw_os_error() == Some(libc::EIO) => break, + Err(_) => break, }; if count == 0 { break; @@ -572,6 +620,9 @@ fn read_output( } } } + if lifecycle.should_emit_natural_exit() && !pending.is_empty() { + on_output(id, String::from_utf8_lossy(&pending).into_owned()); + } } #[cfg(windows)] @@ -2342,4 +2393,51 @@ mod tests { }) ); } + + #[cfg(unix)] + #[test] + fn unix_interrupt_uses_process_group_fallback_while_writer_lock_is_held() { + use std::{ + sync::mpsc, + time::{Duration, Instant}, + }; + + let (sender, receiver) = mpsc::channel(); + let session = PtySession::spawn( + "/bin/sh", + &[ + "-c", + "trap 'printf interrupted; exit' INT; printf ready; while :; do :; done", + ], + TerminalSize::new(80, 24).unwrap(), + move |_, output| { + let _ = sender.send(output); + }, + |_| {}, + ) + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(1); + let mut output = String::new(); + while Instant::now() < deadline && !output.contains("ready") { + if let Ok(chunk) = receiver.recv_timeout(Duration::from_millis(50)) { + output.push_str(&chunk); + } + } + assert!(output.contains("ready")); + assert_eq!( + unsafe { libc::getpgid(session.pid.unwrap() as libc::pid_t) }, + session.pid.unwrap() as libc::pid_t + ); + + let writer = session.writer.lock().unwrap(); + let started = Instant::now(); + session.interrupt().unwrap(); + assert!(started.elapsed() < Duration::from_millis(100)); + assert!(receiver + .recv_timeout(Duration::from_secs(1)) + .unwrap() + .contains("interrupted")); + drop(writer); + session.close(); + } } diff --git a/crates/splice-pty/tests/unix_pty.rs b/crates/splice-pty/tests/unix_pty.rs index 032c459..44df2a1 100644 --- a/crates/splice-pty/tests/unix_pty.rs +++ b/crates/splice-pty/tests/unix_pty.rs @@ -61,4 +61,51 @@ fn pid(output: &str) -> String { output.split("pid=").nth(1).unwrap().trim().to_ assert!(receive_until(&receiver, "24 80").contains("24 80")); session.resize(TerminalSize::new(132, 43).unwrap()).unwrap(); assert!(receive_until(&receiver, "43 132").contains("43 132")); session.close(); } +#[test] fn unix_pty_close_escalates_for_resistant_descendants() { + let (session, receiver) = spawn("trap '' HUP TERM; sh -c 'trap \"\" HUP TERM; echo grandchild=$$; while :; do sleep 1; done' & echo root=$$; wait"); + let deadline = Instant::now() + Duration::from_secs(5); let mut output = String::new(); + while Instant::now() < deadline && (!output.contains("root=") || !output.contains("grandchild=")) { + if let Ok(chunk) = receiver.recv_timeout(Duration::from_millis(250)) { output.push_str(&chunk); } + } + let root = output.split("root=").nth(1).unwrap().lines().next().unwrap().trim().to_owned(); + let grandchild = output.split("grandchild=").nth(1).unwrap().lines().next().unwrap().trim().to_owned(); + let started = Instant::now(); session.close(); + assert!(started.elapsed() < Duration::from_secs(2) && !is_alive(&root) && !is_alive(&grandchild)); } +#[test] fn unix_pty_close_escalates_after_the_leader_exits_on_hup() { + let (session, receiver) = spawn("trap 'exit 0' HUP; sh -c 'trap \"\" HUP TERM; echo grandchild=$$; while :; do sleep 1; done' & echo leader=$$; wait"); + let deadline = Instant::now() + Duration::from_secs(5); let mut output = String::new(); + while Instant::now() < deadline && (!output.contains("leader=") || !output.contains("grandchild=")) { + if let Ok(chunk) = receiver.recv_timeout(Duration::from_millis(250)) { output.push_str(&chunk); } + } + let leader = output.split("leader=").nth(1).unwrap().lines().next().unwrap().trim().to_owned(); + let grandchild = output.split("grandchild=").nth(1).unwrap().lines().next().unwrap().trim().to_owned(); + let started = Instant::now(); session.close(); + assert!(started.elapsed() < Duration::from_secs(2) && !is_alive(&leader) && !is_alive(&grandchild)); } +#[test] fn unix_pty_flushes_partial_utf8_before_natural_exit() { + let (output_sender, output_receiver) = mpsc::channel(); let (exit_sender, exit_receiver) = mpsc::channel(); + let session = PtySession::spawn("/bin/sh", &["-c", "printf final; printf '\\303'"], size(), + move |_, output| { let _ = output_sender.send(output); }, move |_| { let _ = exit_sender.send(()); }).unwrap(); + assert!(exit_receiver.recv_timeout(Duration::from_secs(1)).is_ok()); let mut output = String::new(); + while let Ok(chunk) = output_receiver.recv_timeout(Duration::from_millis(100)) { output.push_str(&chunk); } + assert!(output.contains("final�")); session.close(); } +#[test] fn unix_pty_delivers_final_output_before_natural_exit() { + enum Event { Output(String), Exit } + let (sender, receiver) = mpsc::channel(); let output_sender = sender.clone(); + let session = PtySession::spawn("/bin/sh", &["-c", "printf final-output"], size(), + move |_, output| { std::thread::sleep(Duration::from_millis(100)); let _ = output_sender.send(Event::Output(output)); }, move |_| { let _ = sender.send(Event::Exit); }).unwrap(); + let first = receiver.recv_timeout(Duration::from_secs(1)).unwrap(); + assert!(matches!(first, Event::Output(output) if output.contains("final-output"))); + assert!(matches!(receiver.recv_timeout(Duration::from_secs(1)).unwrap(), Event::Exit)); session.close(); } +#[test] fn unix_pty_bounds_natural_exit_drain_when_descendant_retains_slave() { + enum Event { Output(String), Exit } + let (sender, receiver) = mpsc::channel(); let output_sender = sender.clone(); + let session = PtySession::spawn("/bin/sh", &["-c", "sh -c 'trap \"\" HUP TERM; echo descendant=$$; while :; do sleep 1; done' & sleep 0.1; printf final-output"], size(), + move |_, output| { let _ = output_sender.send(Event::Output(output)); }, move |_| { let _ = sender.send(Event::Exit); }).unwrap(); + let mut output = String::new(); + while !output.contains("final-output") { match receiver.recv_timeout(Duration::from_secs(1)).unwrap() { + Event::Output(chunk) => output.push_str(&chunk), Event::Exit => panic!("final output must precede exit"), + }} + let descendant = pid(&output.replace("descendant", "pid").replace("final-output", "")); + assert!(matches!(receiver.recv_timeout(Duration::from_secs(1)).unwrap(), Event::Exit)); + session.close(); assert!(!is_alive(&descendant)); } }