diff --git a/Cargo.toml b/Cargo.toml index e6236ab..2436e88 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,7 +36,7 @@ windows-sys = { version = "0.61.2", features = [ ] } [target.'cfg(windows)'.dev-dependencies] -windows-sys = { version = "0.61.2", features = ["Win32_System_Console"] } +windows-sys = { version = "0.61.2", features = ["Win32_System_Console", "Win32_System_Threading"] } [profile.release] codegen-units = 1 diff --git a/README.md b/README.md index 09fe4db..fed4ddc 100644 --- a/README.md +++ b/README.md @@ -269,6 +269,23 @@ multiple connections, namespace ownership, cancellation, and final-frame completion. These checks do not yet establish Windows daemon support or complete ConPTY shutdown behavior. +On Windows, normal root-child exit hands the ConPTY master to the existing +child-wait worker for closing. The actor and reader continue draining until the +real output-pipe EOF; only then is the final exit event published. The worker +checks the root's wait handle at 10 ms intervals. After the master is handed +off, Windows input and resize requests fail with a child-exited error; final +snapshots, rows, and replay remain readable until the terminal is removed. +Unix post-exit PTY operations retain their existing behavior. + +### Known Linux cleanup limitation + +A child that exits without consuming a large queued PTY write can leave the +Linux master write blocked after the child, actor, reader, and waiter have +already stopped. `TerminalService::terminate` or service drop can then wait +indefinitely for the writer thread. This pre-existing Unix I/O limitation is +not repaired by the Windows cleanup work. A test watchdog or the daemon's +forced-exit deadline does not prove that those worker threads were joined. + Once the workflow is on `master`, it can also be run manually against a branch: ```sh diff --git a/src/service.rs b/src/service.rs index 5478922..92c71f7 100644 --- a/src/service.rs +++ b/src/service.rs @@ -16,6 +16,9 @@ use serde::{Deserialize, Serialize}; use crate::ghostty::{Format, Terminal, TerminalOptions}; use crate::protocol::AttachmentRole; +#[cfg(windows)] +mod windows; + const DEFAULT_COLS: u16 = 100; const DEFAULT_ROWS: u16 = 30; const REPLAY_CAPACITY: usize = 2 * 1024 * 1024; @@ -400,6 +403,7 @@ impl Drop for TerminalService { struct TerminalHandle { actor_tx: Sender, + shutdown_tx: Mutex>>, killer: Mutex>, info: Arc>, joins: Mutex>>>, @@ -426,7 +430,7 @@ impl TerminalHandle { command.env("TERM", "xterm-256color"); command.env("COLORTERM", "truecolor"); - let mut child = pair + let child = pair .slave .spawn_command(command) .with_context(|| format!("failed to spawn {}", request.program))?; @@ -461,17 +465,28 @@ impl TerminalHandle { let (actor_tx, actor_rx) = bounded(ACTOR_QUEUE_CAPACITY); let (writer_tx, writer_rx) = bounded(WRITER_QUEUE_CAPACITY); + // Disconnecting this channel wakes every cancellable queue operation, + // even when the actor cannot consume an ordinary shutdown message. + let (shutdown_tx, shutdown) = bounded(0); + #[cfg(windows)] + let (close_tx, close_rx) = bounded(1); let (init_tx, init_rx) = std_mpsc::sync_channel(1); let actor_info = Arc::clone(&info); let actor_writer_tx = writer_tx.clone(); + let actor_shutdown = shutdown.clone(); let actor_thread = thread::Builder::new() .name(format!("opencode-pty-actor-{id}")) .spawn(move || { let result = run_actor(ActorConfig { - master: pair.master, + master: ActorMaster { + pty: Some(pair.master), + #[cfg(windows)] + close_tx, + }, messages: actor_rx, writes: actor_writer_tx, + shutdown: actor_shutdown, info: actor_info, replay_capacity, cols: request.cols, @@ -500,7 +515,7 @@ impl TerminalHandle { let writer_actor_tx = actor_tx.clone(); let writer_thread = thread::Builder::new() .name(format!("opencode-pty-writer-{id}")) - .spawn(move || run_writer(writer, writer_rx, writer_actor_tx)) + .spawn(move || run_writer(writer, writer_rx, writer_actor_tx, shutdown)) .context("failed to spawn terminal writer")?; let reader_actor_tx = actor_tx.clone(); @@ -508,6 +523,7 @@ impl TerminalHandle { .name(format!("opencode-pty-reader-{id}")) .spawn(move || { let mut buffer = [0_u8; 8192]; + let mut forwarding = true; loop { match reader.read(&mut buffer) { Ok(0) => { @@ -515,12 +531,18 @@ impl TerminalHandle { break; } Ok(length) => { - if reader_actor_tx - .send(ActorMessage::Output(buffer[..length].to_vec())) - .is_err() - { - break; + if forwarding { + forwarding = reader_actor_tx + .send(ActorMessage::Output(buffer[..length].to_vec())) + .is_ok(); + #[cfg(unix)] + if !forwarding { + break; + } } + // On Windows, ClosePseudoConsole can wait for its + // output pipe to drain. Once the actor stops, keep + // this sole reader draining without forwarding. } Err(error) if error.kind() == io::ErrorKind::Interrupted => continue, #[cfg(unix)] @@ -542,15 +564,22 @@ impl TerminalHandle { let wait_thread = thread::Builder::new() .name(format!("opencode-pty-wait-{id}")) .spawn(move || { - let result = child.wait().map(|status| Some(status.exit_code())); - let _ = wait_actor_tx.send(ActorMessage::ChildExited( - result.map_err(|error| error.to_string()), - )); + #[cfg(windows)] + windows::wait_and_close(child, wait_actor_tx, close_rx); + #[cfg(unix)] + { + let mut child = child; + let result = child.wait().map(|status| Some(status.exit_code())); + let _ = wait_actor_tx.send(ActorMessage::ChildExited( + result.map_err(|error| error.to_string()), + )); + } }) .context("failed to spawn terminal child waiter")?; Ok(Self { actor_tx, + shutdown_tx: Mutex::new(Some(shutdown_tx)), killer: Mutex::new(killer), info, joins: Mutex::new(Some(vec![ @@ -581,17 +610,23 @@ impl TerminalHandle { } fn shutdown(&self) -> Result<()> { + // Keep this guard through the joins: concurrent shutdown callers must + // not return early or signal the child again after it has been reaped. + let mut joins = self + .joins + .lock() + .map_err(|_| anyhow!("terminal join lock poisoned"))?; + let Some(handles) = joins.take() else { + return Ok(()); + }; + self.shutdown_tx + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take(); if let Ok(mut killer) = self.killer.lock() { let _ = killer.kill(); } - let _ = self.actor_tx.send(ActorMessage::Shutdown); - let joins = self - .joins - .lock() - .map_err(|_| anyhow!("terminal join lock poisoned"))? - .take() - .unwrap_or_default(); - for join in joins { + for join in handles { let _ = join.join(); } Ok(()) @@ -646,7 +681,6 @@ enum ActorMessage { Detach { attachment_id: String, }, - Shutdown, } struct Attached { @@ -665,10 +699,47 @@ enum WriterMessage { Bytes(Vec), } +struct ActorMaster { + pty: Option>, + #[cfg(windows)] + close_tx: Sender>, +} + +impl ActorMaster { + fn get(&self) -> Result<&(dyn MasterPty + Send)> { + self.pty + .as_deref() + .ok_or_else(|| anyhow!("terminal child has exited")) + } + + #[cfg(windows)] + fn close(&mut self) -> Result<()> { + if let Some(master) = self.pty.take() { + // A capacity-one channel receives this unique master exactly once; + // handoff cannot wait on ClosePseudoConsole or PTY output drainage. + if let Err(error) = self.close_tx.send(master) { + // Keep ownership on failure. The actor stops receiving before + // dropping this fallback master, so its reader can still drain. + self.pty = Some(error.0); + bail!("terminal close worker stopped"); + } + } + Ok(()) + } +} + +impl Drop for ActorMaster { + fn drop(&mut self) { + #[cfg(windows)] + let _ = self.close(); + } +} + struct ActorConfig { - master: Box, + master: ActorMaster, messages: Receiver, writes: Sender, + shutdown: Receiver<()>, info: Arc>, replay_capacity: usize, cols: u16, @@ -681,12 +752,15 @@ fn run_actor(config: ActorConfig) -> Result<()> { master, messages, writes, + shutdown, info, replay_capacity, cols, rows, init_tx, } = config; + #[cfg(windows)] + let mut master = master; let mut terminal = Terminal::new(TerminalOptions { cols, rows, @@ -703,24 +777,30 @@ fn run_actor(config: ActorConfig) -> Result<()> { let mut pending_message = None; let _ = init_tx.send(Ok(())); - loop { - match receive_actor_message(&messages, &mut pending_message) { + let result = loop { + match receive_actor_message(&messages, &mut pending_message, &shutdown) { Ok(ActorMessage::RefreshForegroundProcess { reply }) => { - publish_foreground_process( - &*master, - &info, - &mut subscribers, - &mut controller, - &mut controller_generation, - ); + if let Ok(master) = master.get() { + publish_foreground_process( + &shutdown, + master, + &info, + &mut subscribers, + &mut controller, + &mut controller_generation, + ); + } let _ = reply.send(Ok(())); } Ok(ActorMessage::Output(bytes)) => { let (start, end) = replay.append(&bytes); terminal.vt_write(&bytes); - forward_replies(&mut terminal, &writes)?; + if let Err(error) = forward_replies(&mut terminal, &writes, &shutdown) { + break Err(error); + } update_offsets(&info, &replay); broadcast( + &shutdown, &mut subscribers, &mut controller, &mut controller_generation, @@ -739,6 +819,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { }; if changed { broadcast( + &shutdown, &mut subscribers, &mut controller, &mut controller_generation, @@ -754,9 +835,9 @@ fn run_actor(config: ActorConfig) -> Result<()> { }) => { let result = authorize_controller(&controller, attachment_id.as_deref()).and_then(|_| { - writes - .send(WriterMessage::Bytes(bytes)) - .map_err(|_| anyhow!("terminal writer stopped")) + #[cfg(windows)] + master.get()?; + queue_write(&writes, &shutdown, bytes) }); let _ = reply.send(result); } @@ -768,14 +849,14 @@ fn run_actor(config: ActorConfig) -> Result<()> { }) => { let result = (|| { authorize_controller(&controller, attachment_id.as_deref())?; - master.resize(PtySize { + master.get()?.resize(PtySize { rows, cols, pixel_width: 0, pixel_height: 0, })?; let resized = terminal.resize(cols, rows); - forward_replies(&mut terminal, &writes)?; + forward_replies(&mut terminal, &writes, &shutdown)?; resized?; if let Ok(mut value) = info.write() { value.cols = cols; @@ -784,6 +865,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { let generation = controller_generation; let checkpoint = format_terminal(&terminal, Format::Vt)?.into_bytes(); broadcast( + &shutdown, &mut subscribers, &mut controller, &mut controller_generation, @@ -823,6 +905,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { } let generation = controller_generation; broadcast( + &shutdown, &mut subscribers, &mut controller, &mut controller_generation, @@ -837,14 +920,14 @@ fn run_actor(config: ActorConfig) -> Result<()> { .map(|value| value.cols != cols || value.rows != rows) .unwrap_or(true); if changed { - master.resize(PtySize { + master.get()?.resize(PtySize { rows, cols, pixel_width: 0, pixel_height: 0, })?; let resized = terminal.resize(cols, rows); - forward_replies(&mut terminal, &writes)?; + forward_replies(&mut terminal, &writes, &shutdown)?; resized?; if let Ok(mut value) = info.write() { value.cols = cols; @@ -853,6 +936,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { let generation = controller_generation; let checkpoint = format_terminal(&terminal, Format::Vt)?.into_bytes(); broadcast( + &shutdown, &mut subscribers, &mut controller, &mut controller_generation, @@ -865,9 +949,9 @@ fn run_actor(config: ActorConfig) -> Result<()> { ); } if let Some(bytes) = bytes { - writes - .send(WriterMessage::Bytes(bytes)) - .map_err(|_| anyhow!("terminal writer stopped"))?; + #[cfg(windows)] + master.get()?; + queue_write(&writes, &shutdown, bytes)?; } Ok(()) })(); @@ -935,6 +1019,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { let controller_id = controller.clone(); let generation = controller_generation; broadcast( + &shutdown, &mut subscribers, &mut controller, &mut controller_generation, @@ -960,6 +1045,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { let generation = controller_generation; let attachment_id = controller.clone(); broadcast( + &shutdown, &mut subscribers, &mut controller, &mut controller_generation, @@ -972,15 +1058,24 @@ fn run_actor(config: ActorConfig) -> Result<()> { } Ok(ActorMessage::ChildExited(result)) => { child_exit = Some(result); + // ConPTY keeps its output pipe open until HPCON is closed. + // The wait worker closes it while this actor and the reader + // keep processing output, including the real final EOF. + #[cfg(windows)] + if let Err(error) = master.close() { + break Err(error); + } } Ok(ActorMessage::ReaderFailed(message)) | Ok(ActorMessage::WriterFailed(message)) => { - if let Ok(mut value) = info.write() { + // Closing ConPTY can finish a pending write with BrokenPipe + // after the final exit event. Never overwrite a published exit. + if !exit_published && let Ok(mut value) = info.write() { value.lifecycle = TerminalLifecycle::Failed { message }; } } Ok(ActorMessage::ReaderEof) => reader_eof = true, - Ok(ActorMessage::Shutdown) | Err(crossbeam_channel::RecvTimeoutError::Disconnected) => { - break; + Err(crossbeam_channel::RecvTimeoutError::Disconnected) => { + break Ok(()); } Err(crossbeam_channel::RecvTimeoutError::Timeout) => continue, } @@ -996,6 +1091,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { }; } broadcast( + &shutdown, &mut subscribers, &mut controller, &mut controller_generation, @@ -1006,22 +1102,41 @@ fn run_actor(config: ActorConfig) -> Result<()> { ); exit_published = true; } - } + }; + // Release senders blocked on the actor queue before closing ConPTY. Its + // synchronous close may need the reader to continue draining final output. + drop(messages); drop(terminal); drop(master); drop(writes); - Ok(()) + if matches!( + shutdown.try_recv(), + Err(crossbeam_channel::TryRecvError::Disconnected) + ) { + // Interrupted terminal-generated replies are expected during shutdown. + Ok(()) + } else { + result + } } fn receive_actor_message( messages: &Receiver, pending: &mut Option, + shutdown: &Receiver<()>, ) -> std::result::Result { + if matches!( + shutdown.try_recv(), + Err(crossbeam_channel::TryRecvError::Disconnected) + ) { + return Err(crossbeam_channel::RecvTimeoutError::Disconnected); + } let message = match pending.take() { Some(message) => message, - None => messages - .recv() - .map_err(|_| crossbeam_channel::RecvTimeoutError::Disconnected)?, + None => crossbeam_channel::select_biased! { + recv(shutdown) -> _ => return Err(crossbeam_channel::RecvTimeoutError::Disconnected), + recv(messages) -> message => message.map_err(|_| crossbeam_channel::RecvTimeoutError::Disconnected)?, + }, }; let ActorMessage::Output(mut bytes) = message else { return Ok(message); @@ -1042,6 +1157,7 @@ fn receive_actor_message( } fn publish_foreground_process( + shutdown: &Receiver<()>, master: &dyn MasterPty, info: &Arc>, subscribers: &mut HashMap, @@ -1061,6 +1177,7 @@ fn publish_foreground_process( }; if changed { broadcast( + shutdown, subscribers, controller, controller_generation, @@ -1157,21 +1274,42 @@ fn process_depth(pid: i32, parents: &HashMap) -> usize { // Forward effects immediately after each native mutation, even if it reports an // error. In particular, resize notifications must precede any accompanying input. // Actual PTY I/O remains on the single writer thread, never inside a callback. -fn forward_replies(terminal: &mut Terminal, writes: &Sender) -> Result<()> { +fn forward_replies( + terminal: &mut Terminal, + writes: &Sender, + shutdown: &Receiver<()>, +) -> Result<()> { for response in terminal.take_writes() { - writes - .send(WriterMessage::Bytes(response)) - .map_err(|_| anyhow!("terminal writer stopped"))?; + queue_write(writes, shutdown, response)?; } Ok(()) } +fn queue_write( + writes: &Sender, + shutdown: &Receiver<()>, + bytes: Vec, +) -> Result<()> { + crossbeam_channel::select_biased! { + recv(shutdown) -> _ => bail!("terminal is stopping"), + send(writes, WriterMessage::Bytes(bytes)) -> result => result.map_err(|_| anyhow!("terminal writer stopped")), + } +} + fn run_writer( mut writer: Box, writer_rx: Receiver, actor_tx: Sender, + shutdown: Receiver<()>, ) { - for message in writer_rx { + loop { + let message = crossbeam_channel::select_biased! { + recv(shutdown) -> _ => break, + recv(writer_rx) -> message => match message { + Ok(message) => message, + Err(_) => break, + }, + }; let result = match message { WriterMessage::Bytes(bytes) => writer.write_all(&bytes).and_then(|_| writer.flush()), }; @@ -1306,21 +1444,33 @@ impl ReplayBuffer { } fn broadcast( + shutdown: &Receiver<()>, subscribers: &mut HashMap, controller: &mut Option, controller_generation: &mut u64, event: StreamEvent, ) { - let dropped = subscribers - .iter() - .filter_map(|(id, subscriber)| { - let failed = match subscriber.role { - AttachmentRole::Controller => subscriber.events.send(event.clone()).is_err(), - AttachmentRole::Observer => subscriber.events.try_send(event.clone()).is_err(), - }; - failed.then(|| id.clone()) - }) - .collect::>(); + let mut dropped = Vec::new(); + for (id, subscriber) in subscribers.iter() { + let failed = match subscriber.role { + AttachmentRole::Controller => crossbeam_channel::select_biased! { + // Quitting is not a subscriber disconnect: do not promote and + // recursively notify controllers while tearing the actor down. + recv(shutdown) -> _ => return, + send(subscriber.events, event.clone()) -> result => result.is_err(), + }, + AttachmentRole::Observer => subscriber.events.try_send(event.clone()).is_err(), + }; + if failed { + dropped.push(id.clone()); + } + } + if matches!( + shutdown.try_recv(), + Err(crossbeam_channel::TryRecvError::Disconnected) + ) { + return; + } for id in dropped { subscribers.remove(&id); if controller.as_ref() == Some(&id) { @@ -1329,6 +1479,7 @@ fn broadcast( let generation = *controller_generation; let attachment_id = controller.clone(); broadcast( + shutdown, subscribers, controller, controller_generation, @@ -1390,6 +1541,255 @@ fn default_shell() -> String { mod tests { use super::*; + struct TestMaster; + + impl MasterPty for TestMaster { + fn resize(&self, _: PtySize) -> Result<()> { + Ok(()) + } + fn get_size(&self) -> Result { + Ok(PtySize::default()) + } + fn try_clone_reader(&self) -> Result> { + unreachable!() + } + fn take_writer(&self) -> Result> { + unreachable!() + } + #[cfg(unix)] + fn process_group_leader(&self) -> Option { + None + } + #[cfg(unix)] + fn as_raw_fd(&self) -> Option { + None + } + #[cfg(unix)] + fn tty_name(&self) -> Option { + None + } + } + + #[derive(Debug)] + struct TestKiller; + + impl ChildKiller for TestKiller { + fn kill(&mut self) -> io::Result<()> { + Ok(()) + } + fn clone_killer(&self) -> Box { + Box::new(Self) + } + } + + // Real actor/parser and bounded channels, without OS pipe capacity or a + // large output flood. Native child/handle coverage lives in tests/runtime.rs. + fn test_actor() -> (TerminalHandle, Receiver) { + let (actor_tx, messages) = bounded(ACTOR_QUEUE_CAPACITY); + let (writes, writer_rx) = bounded(1); + let (shutdown_tx, shutdown) = bounded(0); + #[cfg(windows)] + let (close_tx, close_rx) = bounded(1); + #[cfg(windows)] + let closer = thread::spawn(move || { + if let Ok(master) = close_rx.recv() { + drop(master); + } + }); + let (init_tx, init_rx) = std_mpsc::sync_channel(1); + let info = Arc::new(RwLock::new(TerminalInfo { + id: 1, + pid: None, + title: "test".into(), + foreground_process: None, + group_id: "test".into(), + command: vec![], + cwd: PathBuf::new(), + cols: 80, + rows: 24, + lifecycle: TerminalLifecycle::Running, + output_head: 0, + output_tail: 0, + })); + let actor_info = Arc::clone(&info); + let actor = thread::spawn(move || { + run_actor(ActorConfig { + master: ActorMaster { + pty: Some(Box::new(TestMaster)), + #[cfg(windows)] + close_tx, + }, + messages, + writes, + shutdown, + info: actor_info, + replay_capacity: 4096, + cols: 80, + rows: 24, + init_tx, + }) + .unwrap() + }); + init_rx.recv().unwrap().unwrap(); + ( + TerminalHandle { + actor_tx, + shutdown_tx: Mutex::new(Some(shutdown_tx)), + killer: Mutex::new(Box::new(TestKiller)), + info, + joins: Mutex::new(Some(vec![ + actor, + #[cfg(windows)] + closer, + ])), + }, + writer_rx, + ) + } + + #[test] + fn child_exit_and_reader_eof_are_distinct_actor_signals() { + for eof_first in [false, true] { + let (actor, _writer) = test_actor(); + actor + .actor_tx + .send(if eof_first { + ActorMessage::ReaderEof + } else { + ActorMessage::ChildExited(Ok(Some(23))) + }) + .unwrap(); + let snapshot = actor + .request(|reply| ActorMessage::Snapshot { reply }) + .unwrap(); + assert_eq!(snapshot.info.lifecycle, TerminalLifecycle::Running); + if !eof_first { + actor + .actor_tx + .send(ActorMessage::Output(b"final".to_vec())) + .unwrap(); + } + actor + .actor_tx + .send(if eof_first { + ActorMessage::ChildExited(Ok(Some(23))) + } else { + ActorMessage::ReaderEof + }) + .unwrap(); + let snapshot = actor + .request(|reply| ActorMessage::Snapshot { reply }) + .unwrap(); + assert_eq!( + snapshot.info.lifecycle, + TerminalLifecycle::Exited { + exit_code: Some(23) + } + ); + if !eof_first { + assert_eq!(snapshot.text, "final"); + } + actor + .actor_tx + .send(ActorMessage::WriterFailed("late broken pipe".into())) + .unwrap(); + let after_write_error = actor + .request(|reply| ActorMessage::Snapshot { reply }) + .unwrap(); + assert_eq!(after_write_error.info.lifecycle, snapshot.info.lifecycle); + actor.shutdown().unwrap(); + actor.shutdown().unwrap(); + } + } + + #[test] + #[cfg(windows)] + fn failed_close_handoff_retains_master_ownership() { + let (close_tx, close_rx) = bounded(1); + drop(close_rx); + let mut master = ActorMaster { + pty: Some(Box::new(TestMaster)), + close_tx, + }; + assert!(master.close().is_err()); + assert!(master.get().is_ok()); + } + + #[test] + fn shutdown_interrupts_controller_backpressure() { + let (actor, _writer) = test_actor(); + let attached = actor + .request(|reply| ActorMessage::Attach { + offset: 0, + attachment_id: "controller".into(), + role: AttachmentRole::Controller, + takeover: false, + reply, + }) + .unwrap(); + // Attach already queued ControllerChanged. Force one output per batch + // with a following actor request, filling the queue with only 1 KiB. + for _ in 1..SUBSCRIBER_QUEUE_CAPACITY { + actor + .actor_tx + .send(ActorMessage::Output(b"x".to_vec())) + .unwrap(); + actor + .request(|reply| ActorMessage::Replay { offset: 0, reply }) + .unwrap(); + } + assert!(attached.events.is_full()); + actor + .actor_tx + .send(ActorMessage::Output(b"!".to_vec())) + .unwrap(); + let (done, finished) = std_mpsc::channel(); + let shutdown = thread::spawn(move || { + actor.shutdown().unwrap(); + done.send(()).unwrap(); + }); + let stopped = finished.recv_timeout(Duration::from_secs(1)).is_ok(); + // Release the test consumer even on failure so a red regression never + // leaves a thread stuck or depends on the runtime-test watchdog. + drop(attached); + shutdown.join().unwrap(); + assert!( + stopped, + "shutdown waited for a controller to consume output" + ); + } + + #[test] + fn shutdown_interrupts_actor_to_writer_backpressure() { + let (actor, writer) = test_actor(); + actor + .request(|reply| ActorMessage::Write { + attachment_id: None, + bytes: b"first".to_vec(), + reply, + }) + .unwrap(); + assert!(writer.is_full()); + let (reply, _reply_rx) = std_mpsc::sync_channel(1); + actor + .actor_tx + .send(ActorMessage::Write { + attachment_id: None, + bytes: b"blocked".to_vec(), + reply, + }) + .unwrap(); + let (done, finished) = std_mpsc::channel(); + let shutdown = thread::spawn(move || { + actor.shutdown().unwrap(); + done.send(()).unwrap(); + }); + let stopped = finished.recv_timeout(Duration::from_secs(1)).is_ok(); + drop(writer); + shutdown.join().unwrap(); + assert!(stopped, "shutdown waited for the full input queue"); + } + fn rows_terminal(cols: u16, rows: u16, input: &str) -> Terminal { let mut terminal = Terminal::new(TerminalOptions { cols, @@ -1405,7 +1805,8 @@ mod tests { fn forwarding_replies_drains_once_and_reports_writer_failure() { let mut terminal = rows_terminal(10, 3, "\x1b[5n\x1b[6n"); let (writes, reader) = bounded(4); - forward_replies(&mut terminal, &writes).unwrap(); + let (_shutdown_tx, shutdown) = bounded(0); + forward_replies(&mut terminal, &writes, &shutdown).unwrap(); for expected in [b"\x1b[0n".as_slice(), b"\x1b[1;1R".as_slice()] { let WriterMessage::Bytes(bytes) = reader.try_recv().unwrap(); assert_eq!(bytes, expected); @@ -1416,7 +1817,7 @@ mod tests { terminal.vt_write(b"\x1b[5n"); drop(reader); assert_eq!( - forward_replies(&mut terminal, &writes) + forward_replies(&mut terminal, &writes, &shutdown) .unwrap_err() .to_string(), "terminal writer stopped", @@ -1652,6 +2053,7 @@ mod tests { #[test] fn controller_backpressures_instead_of_disconnecting() { + let (_shutdown_tx, shutdown) = bounded(0); let (events_tx, events) = bounded(1); events_tx .send(StreamEvent::Output { @@ -1672,6 +2074,7 @@ mod tests { let mut controller = Some("controller".to_string()); let mut generation = 1; broadcast( + &shutdown, &mut subscribers, &mut controller, &mut generation, @@ -1697,6 +2100,7 @@ mod tests { #[test] fn slow_observer_is_disconnected() { + let (_shutdown_tx, shutdown) = bounded(0); let (events_tx, _events) = bounded(1); events_tx .send(StreamEvent::Output { @@ -1715,6 +2119,7 @@ mod tests { )]); broadcast( + &shutdown, &mut subscribers, &mut None, &mut 0, @@ -1730,6 +2135,7 @@ mod tests { #[test] fn output_is_batched_without_reordering_other_messages() { + let (_shutdown_tx, shutdown) = bounded(0); let (messages_tx, messages) = bounded(4); messages_tx .send(ActorMessage::Output(b"abc".to_vec())) @@ -1737,16 +2143,16 @@ mod tests { messages_tx .send(ActorMessage::Output(b"def".to_vec())) .unwrap(); - messages_tx.send(ActorMessage::Shutdown).unwrap(); + messages_tx.send(ActorMessage::ReaderEof).unwrap(); let mut pending = None; assert!(matches!( - receive_actor_message(&messages, &mut pending).unwrap(), + receive_actor_message(&messages, &mut pending, &shutdown).unwrap(), ActorMessage::Output(bytes) if bytes == b"abcdef" )); assert!(matches!( - receive_actor_message(&messages, &mut pending).unwrap(), - ActorMessage::Shutdown + receive_actor_message(&messages, &mut pending, &shutdown).unwrap(), + ActorMessage::ReaderEof )); } } diff --git a/src/service/windows.rs b/src/service/windows.rs new file mode 100644 index 0000000..81c8cd4 --- /dev/null +++ b/src/service/windows.rs @@ -0,0 +1,77 @@ +use std::time::Duration; + +use crossbeam_channel::{Receiver, RecvTimeoutError, Sender}; +use portable_pty::{Child, MasterPty}; +use windows_sys::Win32::{ + Foundation::{WAIT_OBJECT_0, WAIT_TIMEOUT}, + System::Threading::{INFINITE, WaitForSingleObject}, +}; + +use super::ActorMessage; + +/// Keep blocking ClosePseudoConsole off the actor/reader without introducing +/// another worker. The actor hands its unique master here on root exit or on +/// teardown; closing it then produces the real output-pipe EOF. +/// Root-exit detection uses a 10 ms poll interval. After reaping, the +/// worker blocks on the close channel rather than continuing to poll. +pub(super) fn wait_and_close( + mut child: Box, + events: Sender, + close: Receiver>, +) { + let mut reported = false; + loop { + if !reported { + // SAFETY: the child owns this process handle throughout this call; + // cloned killer handles do not close or replace it. Poll the wait + // handle rather than GetExitCodeProcess so exit code 259 is not + // confused with STILL_ACTIVE by portable-pty's try_wait. + let ready = child + .as_raw_handle() + .is_none_or(|handle| unsafe { WaitForSingleObject(handle, 0) != WAIT_TIMEOUT }); + if ready { + report_exit(&mut *child, &events); + reported = true; + } + } + let request = if reported { + close.recv().map_err(|_| RecvTimeoutError::Disconnected) + } else { + close.recv_timeout(Duration::from_millis(10)) + }; + match request { + Ok(master) => { + if !reported { + // Also cover actor failure/early teardown, not just normal + // exit or the service's already-issued termination request. + let _ = child.kill(); + } + drop(master); + break; + } + Err(RecvTimeoutError::Disconnected) => break, + Err(RecvTimeoutError::Timeout) => {} + } + } + if !reported { + report_exit(&mut *child, &events); + } +} + +fn report_exit(child: &mut dyn Child, events: &Sender) { + let result = (|| { + if let Some(handle) = child.as_raw_handle() { + // SAFETY: this worker owns the child and therefore its wait handle. + // GetExitCodeProcess can expose the termination status before the + // process object is signaled; portable-pty's wait fast path is not + // sufficient to prove that TerminateProcess teardown has finished. + if unsafe { WaitForSingleObject(handle, INFINITE) } != WAIT_OBJECT_0 { + return Err(std::io::Error::last_os_error()); + } + } + child.wait().map(|status| Some(status.exit_code())) + })(); + let _ = events.send(ActorMessage::ChildExited( + result.map_err(|error| error.to_string()), + )); +} diff --git a/tests/runtime.rs b/tests/runtime.rs index b8d3e5b..4a5cf15 100644 --- a/tests/runtime.rs +++ b/tests/runtime.rs @@ -7,7 +7,9 @@ use std::thread; use std::time::{Duration, Instant}; use opencode_pty::protocol::AttachmentRole; -use opencode_pty::service::{StreamEvent, TerminalId, TerminalService, TerminalSnapshot}; +use opencode_pty::service::{ + StreamEvent, TerminalId, TerminalLifecycle, TerminalService, TerminalSnapshot, +}; use terminal_fixture::{Command, Deadline, Fixture, TempDir}; fn wait_text(service: &TerminalService, id: TerminalId, expected: &str) -> TerminalSnapshot { @@ -231,3 +233,256 @@ fn independent_terminals_keep_bounded_replay() { service.terminate(beta.id).unwrap(); assert!(service.list().unwrap().is_empty()); } + +#[test] +fn normal_exit_follows_final_output_and_preserves_exit_code() { + check_normal_exit(23); +} + +#[test] +#[cfg(windows)] +fn windows_exit_code_259_is_not_mistaken_for_a_running_child() { + check_normal_exit(259); +} + +fn check_normal_exit(code: u32) { + let _deadline = Deadline::new(); + let fixture = Fixture::new(); + let service = TerminalService::default(); + let info = service.create(fixture.request()).unwrap(); + let mut child = fixture.connect(); + let observer = service + .attach( + info.id, + 0, + "exit-observer".into(), + AttachmentRole::Observer, + false, + ) + .unwrap(); + let mut bytes = observer.replay.bytes.clone(); + let mut offset = observer.replay.end_offset; + child.command(Command::Output("\r\nFINAL_OUTPUT_BEFORE_EXIT\r\n".into())); + child.command(Command::Exit(code as i32)); + loop { + match observer + .events + .recv_timeout(Duration::from_secs(15)) + .unwrap_or_else(|error| { + panic!( + "missing exit event: {error}; snapshot={:?}", + service.snapshot(info.id) + ) + }) { + StreamEvent::Output { + start, + end, + bytes: output, + } => { + assert_eq!(start, offset); + assert_eq!(end - start, output.len() as u64); + bytes.extend(output); + offset = end; + } + StreamEvent::Exited { + exit_code, + final_offset, + } => { + assert_eq!(exit_code, Some(code)); + assert_eq!(final_offset, offset); + break; + } + _ => {} + } + } + assert_eq!(service.replay(info.id, 0).unwrap().bytes, bytes); + let snapshot = service.snapshot(info.id).unwrap(); + assert!( + snapshot.text.contains("FINAL_OUTPUT_BEFORE_EXIT"), + "{snapshot:?}" + ); + assert_eq!( + snapshot.info.lifecycle, + TerminalLifecycle::Exited { + exit_code: Some(code) + } + ); + assert_eq!(snapshot.info.output_tail, offset); + assert!(observer.events.try_recv().is_err()); + #[cfg(windows)] + { + assert!( + service + .resize(info.id, 100, 40) + .unwrap_err() + .to_string() + .contains("child has exited") + ); + assert!(service.write(info.id, b"late input".to_vec()).is_err()); + assert_eq!( + service.snapshot(info.id).unwrap().info.lifecycle, + snapshot.info.lifecycle + ); + assert!( + service + .read_rows(info.id, None) + .unwrap() + .lines + .iter() + .any(|line| line.contains("FINAL_OUTPUT_BEFORE_EXIT")) + ); + } + drop(observer); + service.terminate(info.id).unwrap(); +} + +#[test] +fn repeated_termination_and_drop_release_children() { + termination_cycles(false); +} + +// Linux has a pre-existing portable-pty blocking-write hang after child exit; +// that repro is retained outside this Windows cleanup suite, not claimed fixed. +#[test] +#[cfg(windows)] +fn repeated_termination_and_drop_release_nonreading_children() { + termination_cycles(true); +} + +fn termination_cycles(block_input: bool) { + let _deadline = Deadline::new(); + let service = TerminalService::default(); + for _ in 0..3 { + let fixture = Fixture::new(); + let info = service.create(fixture.request()).unwrap(); + let _child = fixture.connect(); + let process = ChildProcess::open(info.pid.unwrap()); + // Child waits on the private control channel, never draining PTY stdin. + if block_input { + service.write(info.id, vec![b'x'; 256 * 1024]).unwrap(); + } + service.terminate(info.id).unwrap(); + process.assert_exited(); + assert!(service.list().unwrap().is_empty()); + assert!(service.terminate(info.id).is_err()); + } + let fixture = Fixture::new(); + let info = service.create(fixture.request()).unwrap(); + let _child = fixture.connect(); + let process = ChildProcess::open(info.pid.unwrap()); + if block_input { + service.write(info.id, vec![b'x'; 256 * 1024]).unwrap(); + } + drop(service); + process.assert_exited(); +} + +struct ChildProcess { + #[cfg(unix)] + pid: u32, + #[cfg(windows)] + handle: std::os::windows::io::OwnedHandle, +} + +#[test] +#[cfg(windows)] +fn terminating_a_shell_releases_its_console_child() { + use base64::Engine; + + let _deadline = Deadline::new(); + let fixture = Fixture::new(); + let service = TerminalService::default(); + let mut request = fixture.request(); + request + .env + .insert("PTY_FIXTURE_EXE".into(), request.program); + request.program = std::path::PathBuf::from(std::env::var_os("SystemRoot").unwrap()) + .join("System32/WindowsPowerShell/v1.0/powershell.exe") + .to_str() + .unwrap() + .to_owned(); + // EncodedCommand avoids imposing cmd.exe's distinct quoting rules on the + // portable-pty CommandBuilder's normal Windows argv quoting. + let script = "& $env:PTY_FIXTURE_EXE --ignored --exact terminal_fixture::child --nocapture --test-threads=1; exit $LASTEXITCODE"; + let script = script + .encode_utf16() + .flat_map(u16::to_le_bytes) + .collect::>(); + request.args = vec![ + "-NoLogo".into(), + "-NoProfile".into(), + "-NonInteractive".into(), + "-EncodedCommand".into(), + base64::engine::general_purpose::STANDARD.encode(script), + ]; + let info = service.create(request).unwrap(); + let root = ChildProcess::open(info.pid.unwrap()); + let mut child = fixture.connect(); + let pid = u32::try_from(child.command(Command::Context)["pid"].as_u64().unwrap()).unwrap(); + assert_ne!(pid, info.pid.unwrap()); + let descendant = ChildProcess::open(pid); + child.command(Command::Output("\r\nSHELL_DESCENDANT_READY".into())); + wait_text(&service, info.id, "SHELL_DESCENDANT_READY"); + service.terminate(info.id).unwrap(); + root.assert_exited(); + descendant.assert_exited(); +} + +impl ChildProcess { + fn open(pid: u32) -> Self { + #[cfg(unix)] + { + Self { pid } + } + #[cfg(windows)] + { + use std::os::windows::io::FromRawHandle; + use windows_sys::Win32::System::Threading::{OpenProcess, PROCESS_SYNCHRONIZE}; + // SAFETY: request only wait access to this known-live test child. + let handle = unsafe { OpenProcess(PROCESS_SYNCHRONIZE, 0, pid) }; + assert!(!handle.is_null(), "{}", std::io::Error::last_os_error()); + Self { + handle: unsafe { std::os::windows::io::OwnedHandle::from_raw_handle(handle) }, + } + } + } + + fn assert_exited(&self) { + #[cfg(unix)] + { + // SAFETY: signal zero checks liveness without signalling a process. + assert_eq!(unsafe { libc::kill(self.pid as i32, 0) }, -1); + assert_eq!( + std::io::Error::last_os_error().raw_os_error(), + Some(libc::ESRCH) + ); + } + #[cfg(windows)] + { + use std::os::windows::io::AsRawHandle; + use windows_sys::Win32::{ + Foundation::WAIT_OBJECT_0, System::Threading::WaitForSingleObject, + }; + // SAFETY: the owned process handle has wait access and stays live. + assert_eq!( + unsafe { WaitForSingleObject(self.handle.as_raw_handle(), 0) }, + WAIT_OBJECT_0 + ); + } + } +} + +#[test] +#[cfg(windows)] +fn shutdown_drains_conpty_while_a_child_is_producing_output() { + let _deadline = Deadline::new(); + let fixture = Fixture::new(); + let service = TerminalService::default(); + let info = service.create(fixture.request()).unwrap(); + let mut child = fixture.connect(); + let process = ChildProcess::open(info.pid.unwrap()); + child.command(Command::Flood); + wait_text(&service, info.id, "FLOOD_OUTPUT"); + drop(service); + process.assert_exited(); +} diff --git a/tests/support/terminal_fixture.rs b/tests/support/terminal_fixture.rs index c5c3255..7ddc237 100644 --- a/tests/support/terminal_fixture.rs +++ b/tests/support/terminal_fixture.rs @@ -26,6 +26,7 @@ pub enum Command { Size, Context, Exit(i32), + Flood, } pub struct Fixture { @@ -162,6 +163,7 @@ fn child() { } Command::Size => serde_json::json!(console_size()), Command::Context => serde_json::json!({ + "pid": std::process::id(), "cwd": env::current_dir().unwrap(), "args": env::args().skip(1).collect::>(), "value": env::var("PTY_FIXTURE_VALUE").ok(), @@ -172,6 +174,15 @@ fn child() { send(channel.get_mut(), &serde_json::Value::Null); std::process::exit(code); } + Command::Flood => { + send(channel.get_mut(), &serde_json::Value::Null); + let mut stdout = io::stdout().lock(); + let chunk = "FLOOD_OUTPUT\r\n".repeat(256); + loop { + stdout.write_all(chunk.as_bytes()).unwrap(); + stdout.flush().unwrap(); + } + } }; send(channel.get_mut(), &result); }