diff --git a/crates/tty7-core/src/client/pane.rs b/crates/tty7-core/src/client/pane.rs index 251514f3..4cfacae0 100644 --- a/crates/tty7-core/src/client/pane.rs +++ b/crates/tty7-core/src/client/pane.rs @@ -86,6 +86,20 @@ impl PaneClient { ClientMsg::Kill { pane_id }.encode(&mut stream) } + pub fn send_input(&self, pane_id: u64, bytes: &[u8]) -> io::Result<()> { + let mut stream = self.open()?; + ClientMsg::SendInput { + pane_id, + bytes: bytes.to_vec(), + } + .encode(&mut stream)?; + match DaemonMsg::read(&mut stream)? { + DaemonMsg::InputAck { .. } => Ok(()), + DaemonMsg::Error(message) => Err(io::Error::other(message)), + other => Err(unexpected_reply("SendInput", &other)), + } + } + pub fn procs(&self, pane_id: u64) -> io::Result { let mut stream = self.open()?; ClientMsg::QueryProcs { pane_id }.encode(&mut stream)?; diff --git a/crates/tty7-core/src/core/agent_hooks.rs b/crates/tty7-core/src/core/agent_hooks.rs index 084380d0..c32841b9 100644 --- a/crates/tty7-core/src/core/agent_hooks.rs +++ b/crates/tty7-core/src/core/agent_hooks.rs @@ -136,7 +136,7 @@ fn write_to_controlling_tty(bytes: &[u8]) -> bool { .iter() .find(|p| p.pid == pid) .and_then(|p| name_of(p.parent)) - .is_some_and(|n| n == "tty7-app.exe") + .is_some_and(|n| is_tty7_host_exe(&n)) }); if let Some(pid) = shell { @@ -152,6 +152,11 @@ fn write_to_controlling_tty(bytes: &[u8]) -> bool { any } +#[cfg(any(not(unix), test))] +fn is_tty7_host_exe(name: &str) -> bool { + matches!(name, "tty7-app.exe" | "tty7-server.exe" | "tty7.exe") +} + #[cfg(not(unix))] fn attach_and_write(pid: u32, bytes: &[u8]) -> bool { use windows_sys::Win32::System::Console::{AttachConsole, FreeConsole}; @@ -869,6 +874,16 @@ export default function (pi: ExtensionAPI) {{ mod tests { use super::*; + #[test] + fn every_tty7_daemon_host_takes_the_console_fast_path() { + for name in ["tty7-app.exe", "tty7-server.exe", "tty7.exe"] { + assert!(is_tty7_host_exe(name), "{name} hosts tty7 shells"); + } + for name in ["explorer.exe", "cmd.exe", "tty7", "tty7-app", "wt.exe"] { + assert!(!is_tty7_host_exe(name), "{name} is not a tty7 host process"); + } + } + #[test] fn hook_sequence_round_trips_through_the_daemon_parser() { use crate::core::cli_agent::{AgentEventKind, CLIAgent, parse_agent_event}; diff --git a/crates/tty7-core/src/daemon/pane.rs b/crates/tty7-core/src/daemon/pane.rs index 4e0924fc..5f0dfaa7 100644 --- a/crates/tty7-core/src/daemon/pane.rs +++ b/crates/tty7-core/src/daemon/pane.rs @@ -398,6 +398,10 @@ impl OutputGate { self.drained.notify_all(); } + pub(crate) fn queued_bytes(&self) -> i64 { + self.queued.load(Ordering::Relaxed) + } + fn wait_below_high_water(&self) { if self.queued.load(Ordering::Relaxed) < Self::HIGH_WATER { return; @@ -415,12 +419,20 @@ impl OutputGate { } } +pub(crate) const OBSERVER_BUDGET: i64 = 8 * 1024 * 1024; + +struct Observer { + id: u64, + tx: Sender, + gate: Arc, +} + struct PaneState { id: u64, ring: ReplayRing, subscriber: Option>, subscriber_epoch: u64, - observers: Vec<(u64, Sender)>, + observers: Vec, observer_seq: u64, cwd: Option, shell: ShellState, @@ -436,7 +448,25 @@ fn notify(st: &mut PaneState, msg: DaemonMsg) { if let Some(sub) = &st.subscriber { let _ = sub.send(msg.clone()); } - st.observers.retain(|(_, tx)| tx.send(msg.clone()).is_ok()); + st.observers.retain(|obs| obs.tx.send(msg.clone()).is_ok()); +} + +fn fan_out_output(st: &mut PaneState, bytes: &[u8], gate: &OutputGate) { + if let Some(sub) = &st.subscriber { + if sub.send(DaemonMsg::Output(bytes.to_vec())).is_ok() { + gate.add(bytes.len()); + } + } + st.observers.retain(|obs| { + if obs.gate.queued_bytes() + bytes.len() as i64 > OBSERVER_BUDGET { + return false; + } + if obs.tx.send(DaemonMsg::Output(bytes.to_vec())).is_err() { + return false; + } + obs.gate.add(bytes.len()); + true + }); } enum PaneBackend { @@ -840,14 +870,7 @@ impl DaemonPane { let mut st = state.lock().unwrap(); let facts_before = may_change_facts.then(|| observed_facts(&st)); st.ring.append(bytes); - if let Some(sub) = &st.subscriber { - if sub.send(DaemonMsg::Output(bytes.to_vec())).is_ok() { - gate.add(n); - } - } - st.observers.retain(|(_, tx)| { - tx.send(DaemonMsg::Output(bytes.to_vec())).is_ok() - }); + fan_out_output(&mut st, bytes, &gate); apply_signals(&mut st, signals); if let Some(remote) = remote { apply_remote_context(&mut st, remote); @@ -905,14 +928,18 @@ impl DaemonPane { !st.alive && st.subscriber.is_none() } - pub fn observe(&self, observer: Sender) -> u64 { + pub fn observe(&self, observer: Sender, gate: Arc) -> u64 { let mut st = self.state.lock().unwrap(); - observe_subscriber(&mut st, observer) + observe_subscriber(&mut st, observer, gate) } pub fn unobserve(&self, observer_id: u64) { let mut st = self.state.lock().unwrap(); - st.observers.retain(|(id, _)| *id != observer_id); + st.observers.retain(|obs| obs.id != observer_id); + } + + pub fn controls(&self, epoch: u64) -> bool { + self.state.lock().unwrap().subscriber_epoch == epoch } pub fn agent_state(&self) -> Option { @@ -955,7 +982,7 @@ impl DaemonPane { let mut st = self.state.lock().unwrap(); st.ring.resize(size); st.observers - .retain(|(_, tx)| tx.send(DaemonMsg::Size(size)).is_ok()); + .retain(|obs| obs.tx.send(DaemonMsg::Size(size)).is_ok()); } match &self.backend { PaneBackend::Pty(p) => { @@ -967,7 +994,6 @@ impl DaemonPane { } } - #[allow(dead_code)] pub fn alive(&self) -> bool { self.state.lock().unwrap().alive } @@ -1301,10 +1327,14 @@ fn attach_subscriber(st: &mut PaneState, subscriber: Sender) -> u64 { st.subscriber_epoch } -fn observe_subscriber(st: &mut PaneState, observer: Sender) -> u64 { +fn observe_subscriber(st: &mut PaneState, observer: Sender, gate: Arc) -> u64 { st.observer_seq += 1; replay_state(st, &observer); - st.observers.push((st.observer_seq, observer)); + st.observers.push(Observer { + id: st.observer_seq, + tx: observer, + gate, + }); st.observer_seq } @@ -2979,7 +3009,7 @@ mod tests { drain(&controller_rx); let (observer_tx, observer_rx) = mpsc::channel(); - let id = observe_subscriber(&mut st, observer_tx); + let id = observe_subscriber(&mut st, observer_tx, Arc::new(OutputGate::new())); assert_eq!(st.subscriber_epoch, epoch, "observing must not bump the controller epoch"); assert!(st.subscriber.is_some(), "the controller keeps its seat"); @@ -2995,7 +3025,7 @@ mod tests { assert!(matches!(controller_rx.try_recv(), Ok(DaemonMsg::Output(b)) if b == b"tick")); assert!(matches!(observer_rx.try_recv(), Ok(DaemonMsg::Output(b)) if b == b"tick")); - st.observers.retain(|(oid, _)| *oid != id); + st.observers.retain(|obs| obs.id != id); notify(&mut st, DaemonMsg::Output(b"tock".to_vec())); assert!(matches!(controller_rx.try_recv(), Ok(DaemonMsg::Output(_)))); assert!(observer_rx.try_recv().is_err(), "a departed observer hears nothing"); @@ -3010,7 +3040,7 @@ mod tests { drain(&first_rx); let (observer_tx, observer_rx) = mpsc::channel(); - observe_subscriber(&mut st, observer_tx); + observe_subscriber(&mut st, observer_tx, Arc::new(OutputGate::new())); drain(&observer_rx); let (second_tx, second_rx) = mpsc::channel(); @@ -3042,18 +3072,92 @@ mod tests { fn a_gone_observer_is_pruned_on_the_next_broadcast() { let mut st = test_state(true); let (observer_tx, observer_rx) = mpsc::channel(); - observe_subscriber(&mut st, observer_tx); + observe_subscriber(&mut st, observer_tx, Arc::new(OutputGate::new())); drop(observer_rx); notify(&mut st, DaemonMsg::Output(b"x".to_vec())); assert!(st.observers.is_empty(), "a dead observer must not accumulate"); } + #[test] + fn a_stalled_observer_is_dropped_at_its_budget_while_the_controller_streams_on() { + let mut st = test_state(true); + let (controller_tx, controller_rx) = mpsc::channel(); + attach_subscriber(&mut st, controller_tx); + drain(&controller_rx); + + let (observer_tx, observer_rx) = mpsc::channel(); + observe_subscriber(&mut st, observer_tx, Arc::new(OutputGate::new())); + drain(&observer_rx); + + let pane_gate = OutputGate::new(); + let chunk = vec![b'x'; 1024 * 1024]; + let sends = (OBSERVER_BUDGET / chunk.len() as i64) as usize + 2; + for _ in 0..sends { + fan_out_output(&mut st, &chunk, &pane_gate); + pane_gate.sub(chunk.len()); + } + + assert!( + st.observers.is_empty(), + "an observer past its budget must be pruned" + ); + let mut controller_bytes = 0usize; + while let Ok(DaemonMsg::Output(b)) = controller_rx.try_recv() { + controller_bytes += b.len(); + } + assert_eq!( + controller_bytes, + sends * chunk.len(), + "the controller stream must stay complete" + ); + let mut observer_bytes = 0i64; + while let Ok(DaemonMsg::Output(b)) = observer_rx.try_recv() { + observer_bytes += b.len() as i64; + } + assert!( + observer_bytes <= OBSERVER_BUDGET, + "a stalled observer must never hold more than its budget, held {observer_bytes}" + ); + } + + #[test] + fn a_draining_observer_under_the_cap_stays_subscribed() { + let mut st = test_state(true); + let (observer_tx, observer_rx) = mpsc::channel(); + let observer_gate = Arc::new(OutputGate::new()); + observe_subscriber(&mut st, observer_tx, observer_gate.clone()); + drain(&observer_rx); + + let pane_gate = OutputGate::new(); + let chunk = vec![b'y'; 1024 * 1024]; + let sends = (OBSERVER_BUDGET / chunk.len() as i64) as usize * 3; + let mut got = 0usize; + for _ in 0..sends { + fan_out_output(&mut st, &chunk, &pane_gate); + pane_gate.sub(chunk.len()); + while let Ok(DaemonMsg::Output(b)) = observer_rx.try_recv() { + observer_gate.sub(b.len()); + got += b.len(); + } + } + assert_eq!( + st.observers.len(), + 1, + "an observer that keeps draining must stay subscribed" + ); + assert_eq!(got, sends * chunk.len(), "and must miss no bytes"); + } + #[test] fn death_notifies_observers_but_only_controllers_defer_the_reap() { let with_observer_only = Arc::new(Mutex::new(test_state(true))); let (observer_tx, observer_rx) = mpsc::channel(); - observe_subscriber(&mut with_observer_only.lock().unwrap(), observer_tx); + observe_subscriber( + &mut with_observer_only.lock().unwrap(), + observer_tx, + Arc::new(OutputGate::new()), + ); drain(&observer_rx); let (dead_tx, dead_rx) = mpsc::channel(); DeathReporter::new(move || dead_tx.send(()).unwrap()) @@ -3070,7 +3174,7 @@ mod tests { { let mut st = with_both.lock().unwrap(); attach_subscriber(&mut st, controller_tx); - observe_subscriber(&mut st, observer_tx); + observe_subscriber(&mut st, observer_tx, Arc::new(OutputGate::new())); } drain(&controller_rx); drain(&observer_rx); diff --git a/crates/tty7-core/src/daemon/protocol.rs b/crates/tty7-core/src/daemon/protocol.rs index c6cf4fd2..2f84909d 100644 --- a/crates/tty7-core/src/daemon/protocol.rs +++ b/crates/tty7-core/src/daemon/protocol.rs @@ -584,6 +584,10 @@ pub enum ClientMsg { size: WinSize, }, Input(Vec), + SendInput { + pane_id: u64, + bytes: Vec, + }, Resize(WinSize), Detach, Kill { @@ -656,6 +660,9 @@ pub enum DaemonMsg { code: Option, }, PaneList(Vec), + InputAck { + pane_id: u64, + }, RemoteContext(Option), Agent(Option), AgentStatus(Option), @@ -711,6 +718,7 @@ mod kind { pub const ON_WORKSPACE: u8 = 52; pub const SPAWN_OWNED: u8 = 53; pub const OBSERVE: u8 = 54; + pub const SEND_INPUT: u8 = 55; pub const SPAWNED: u8 = 1; pub const SNAPSHOT: u8 = 2; @@ -736,6 +744,7 @@ mod kind { pub const AGENT_STATUS: u8 = 22; pub const VERSION_REPLY: u8 = 40; pub const PROCS: u8 = 50; + pub const INPUT_ACK: u8 = 51; } pub fn write_frame(w: &mut W, kind: u8, payload: &[u8]) -> io::Result<()> { @@ -860,6 +869,9 @@ impl ClientMsg { write_frame(w, kind::OBSERVE, &to_json(&(pane_id, size))?) } ClientMsg::Input(bytes) => write_frame(w, kind::INPUT, bytes), + ClientMsg::SendInput { pane_id, bytes } => { + write_frame(w, kind::SEND_INPUT, &to_json(&(pane_id, bytes))?) + } ClientMsg::Resize(size) => write_frame(w, kind::RESIZE, &to_json(size)?), ClientMsg::Detach => write_frame(w, kind::DETACH, &[]), ClientMsg::Kill { pane_id } => write_frame(w, kind::KILL, &to_json(pane_id)?), @@ -963,6 +975,10 @@ impl ClientMsg { ClientMsg::Observe { pane_id, size } } kind::INPUT => ClientMsg::Input(payload), + kind::SEND_INPUT => { + let (pane_id, bytes) = from_json(&payload)?; + ClientMsg::SendInput { pane_id, bytes } + } kind::RESIZE => ClientMsg::Resize(from_json(&payload)?), kind::DETACH => ClientMsg::Detach, kind::KILL => ClientMsg::Kill { @@ -1050,6 +1066,7 @@ impl DaemonMsg { } => write_frame(w, kind::PROMPT, &to_json(&(active, at_prompt, last_exit))?), DaemonMsg::Exited { code } => write_frame(w, kind::EXITED, &to_json(code)?), DaemonMsg::PaneList(list) => write_frame(w, kind::PANE_LIST, &to_json(list)?), + DaemonMsg::InputAck { pane_id } => write_frame(w, kind::INPUT_ACK, &to_json(pane_id)?), DaemonMsg::RemoteContext(remote) => { write_frame(w, kind::REMOTE_CONTEXT, &to_json(remote)?) } @@ -1108,6 +1125,9 @@ impl DaemonMsg { code: from_json(&payload)?, }, kind::PANE_LIST => DaemonMsg::PaneList(from_json(&payload)?), + kind::INPUT_ACK => DaemonMsg::InputAck { + pane_id: from_json(&payload)?, + }, kind::REMOTE_CONTEXT => DaemonMsg::RemoteContext(from_json(&payload)?), kind::AGENT => DaemonMsg::Agent(from_json(&payload)?), kind::AGENT_STATUS => DaemonMsg::AgentStatus(from_json(&payload)?), @@ -1269,6 +1289,14 @@ mod tests { size: SIZE, }, ClientMsg::Input(vec![0x1b, b'[', b'A', 0, 255]), + ClientMsg::SendInput { + pane_id: 42, + bytes: vec![b'l', b's', b'\r', 0, 255], + }, + ClientMsg::SendInput { + pane_id: 7, + bytes: Vec::new(), + }, ClientMsg::Resize(SIZE), ClientMsg::Detach, ClientMsg::Kill { pane_id: 7 }, @@ -1405,6 +1433,7 @@ mod tests { owner: Some("ffe038d0-9ad6-40c0-815d-1fcc43c17ec0".into()), }, ]), + DaemonMsg::InputAck { pane_id: 42 }, DaemonMsg::RemoteContext(Some(RemoteContext { kind: RemoteKind::Ssh, argv: vec!["ssh".into(), "-p".into(), "2222".into(), "dev".into()], diff --git a/crates/tty7-core/src/daemon/server.rs b/crates/tty7-core/src/daemon/server.rs index b39ae261..dd84d936 100644 --- a/crates/tty7-core/src/daemon/server.rs +++ b/crates/tty7-core/src/daemon/server.rs @@ -556,6 +556,21 @@ fn handle_conn(stream: Stream, registry: Arc) -> anyhow::Result<()> { Ok(()) } + ClientMsg::SendInput { pane_id, bytes } => { + let mut w = write_stream; + match registry.get(pane_id) { + Some(pane) if pane.alive() => { + pane.write_input(&bytes); + DaemonMsg::InputAck { pane_id }.encode(&mut w)?; + } + Some(_) => { + DaemonMsg::Error(format!("pane {pane_id} is not running")).encode(&mut w)? + } + None => DaemonMsg::Error(format!("no such pane {pane_id}")).encode(&mut w)?, + } + Ok(()) + } + ClientMsg::ListForwards { pane_id } => { let mut w = write_stream; let list = crate::daemon::ssh::SshManager::global().list_forwards(pane_id); @@ -618,12 +633,9 @@ fn stream_observer( ) -> anyhow::Result<()> { let (tx, rx) = mpsc::channel::(); let refusals = tx.clone(); - let observer_id = pane.observe(tx); - let writer = spawn_writer( - rx, - write_stream, - Arc::new(crate::daemon::pane::OutputGate::new()), - ); + let gate = Arc::new(crate::daemon::pane::OutputGate::new()); + let observer_id = pane.observe(tx, gate.clone()); + let writer = spawn_writer(rx, write_stream, gate); observe_loop(&mut read_stream, &refusals); @@ -651,6 +663,8 @@ fn observe_loop(read_stream: &mut R, refusals: &mpsc::Sender, id: u64, @@ -660,32 +674,81 @@ fn run_stream( write_stream: Stream, registry: Arc, ) -> anyhow::Result<()> { + use std::io::Read as _; + let writer = spawn_writer(rx, write_stream, pane.gate()); + let _ = read_stream.set_read_timeout(Some(CONTROL_POLL_INTERVAL)); let mut killed = false; - loop { - match ClientMsg::read(&mut read_stream) { - Ok(ClientMsg::Input(bytes)) => pane.write_input(&bytes), - Ok(ClientMsg::Resize(size)) => pane.resize(size), - Ok(ClientMsg::AuthResponse { - request_id, - response, - }) => pane.deliver_auth_response(request_id, response), - Ok(ClientMsg::Detach) => break, - Ok(ClientMsg::Kill { pane_id }) => { - if pane_id == id { - killed = true; - break; - } else if let Some(other) = registry.remove(pane_id) { - other.kill(); + let mut displaced = false; + let mut pending: Vec = Vec::new(); + let mut chunk = [0u8; 65536]; + 'conn: loop { + if !pane.controls(epoch) { + displaced = true; + break; + } + loop { + let (kind, payload) = match crate::daemon::protocol::take_frame(&mut pending) { + Ok(Some(frame)) => frame, + Ok(None) => break, + Err(_) => break 'conn, + }; + let Ok(msg) = ClientMsg::from_frame(kind, payload) else { + break 'conn; + }; + match msg { + ClientMsg::Input(bytes) => { + if !pane.controls(epoch) { + displaced = true; + break 'conn; + } + pane.write_input(&bytes); } + ClientMsg::Resize(size) => { + if !pane.controls(epoch) { + displaced = true; + break 'conn; + } + pane.resize(size); + } + ClientMsg::AuthResponse { + request_id, + response, + } => pane.deliver_auth_response(request_id, response), + ClientMsg::Detach => break 'conn, + ClientMsg::Kill { pane_id } => { + if pane_id == id { + killed = true; + break 'conn; + } else if let Some(other) = registry.remove(pane_id) { + other.kill(); + } + } + _ => {} + } + } + match read_stream.read(&mut chunk) { + Ok(0) => break, + Ok(n) => pending.extend_from_slice(&chunk[..n]), + Err(e) + if matches!( + e.kind(), + std::io::ErrorKind::WouldBlock + | std::io::ErrorKind::TimedOut + | std::io::ErrorKind::Interrupted + ) => + { + continue; } - Ok(_) => {} Err(_) => break, } } let reclaimable = pane.detach(epoch); + if displaced { + let _ = read_stream.shutdown(std::net::Shutdown::Both); + } let _ = writer.join(); if killed { diff --git a/crates/tty7-core/src/daemon/spawn.rs b/crates/tty7-core/src/daemon/spawn.rs index 5410c8c8..3540a69e 100644 --- a/crates/tty7-core/src/daemon/spawn.rs +++ b/crates/tty7-core/src/daemon/spawn.rs @@ -49,6 +49,35 @@ enum VersionProbe { Unresponsive, } +const DAEMON_EXE_STEMS: [&str; 3] = ["tty7-app", "tty7-server", "tty7"]; + +fn strip_exe_suffix(name: &str) -> &str { + match name.len().checked_sub(4) { + Some(i) if name.is_char_boundary(i) && name[i..].eq_ignore_ascii_case(".exe") => { + &name[..i] + } + _ => name, + } +} + +fn exe_names_equal(a: &str, b: &str) -> bool { + let a = strip_exe_suffix(a); + let b = strip_exe_suffix(b); + if cfg!(windows) { + a.eq_ignore_ascii_case(b) + } else { + a == b + } +} + +fn is_reapable_daemon_name(name: &str) -> bool { + let own = std::env::current_exe() + .ok() + .and_then(|p| p.file_name().map(|n| n.to_string_lossy().into_owned())); + own.as_deref().is_some_and(|own| exe_names_equal(own, name)) + || DAEMON_EXE_STEMS.iter().any(|stem| exe_names_equal(stem, name)) +} + pub fn ensure_running() -> anyhow::Result<()> { if let Ok(mut stream) = transport::connect() { match query_daemon_version(&mut stream) { @@ -174,7 +203,7 @@ fn reap_recorded_daemon() { pidfile::remove(); return; } - if process_matches_own_exe(pid as libc::pid_t) { + if process_matches_daemon_exe(pid as libc::pid_t) { log::warn!("reaping unreachable daemon (pid {pid}); its sessions will be hung up"); reap_process(pid as libc::pid_t); } @@ -182,12 +211,10 @@ fn reap_recorded_daemon() { } #[cfg(any(target_os = "macos", target_os = "linux"))] -fn process_matches_own_exe(pid: libc::pid_t) -> bool { - let ours = std::env::current_exe() - .ok() - .and_then(|p| p.file_name().map(|n| n.to_os_string())); - let theirs = process_path(pid).and_then(|p| p.file_name().map(|n| n.to_os_string())); - matches!((ours, theirs), (Some(a), Some(b)) if a == b) +fn process_matches_daemon_exe(pid: libc::pid_t) -> bool { + process_path(pid) + .and_then(|p| p.file_name().map(|n| n.to_string_lossy().into_owned())) + .is_some_and(|name| is_reapable_daemon_name(&name)) } #[cfg(any(target_os = "macos", target_os = "linux"))] @@ -228,14 +255,10 @@ fn reap_recorded_daemon() { return; } let procs = winproc::snapshot(); - let ours = std::env::current_exe() - .ok() - .and_then(|p| p.file_name().map(|n| n.to_string_lossy().into_owned())); let matches = procs .iter() .find(|p| p.pid == pid) - .zip(ours) - .is_some_and(|(entry, name)| entry.name.eq_ignore_ascii_case(&name)); + .is_some_and(|entry| is_reapable_daemon_name(&entry.name)); if matches { log::warn!("reaping unreachable daemon (pid {pid}); its sessions will be hung up"); for descendant in winproc::descendants(&procs, pid) { @@ -345,6 +368,74 @@ fn detach(cmd: &mut Command) { cmd.creation_flags(DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW); } +#[cfg(test)] +mod exe_name_tests { + use super::*; + + #[test] + fn every_legitimate_daemon_name_is_reapable_with_and_without_exe() { + for name in [ + "tty7-app", + "tty7-server", + "tty7", + "tty7-app.exe", + "tty7-server.exe", + "tty7.exe", + ] { + assert!(is_reapable_daemon_name(name), "{name} is a daemon of ours"); + } + } + + #[test] + fn the_current_executable_name_remains_reapable() { + let own = std::env::current_exe() + .unwrap() + .file_name() + .unwrap() + .to_string_lossy() + .into_owned(); + assert!( + is_reapable_daemon_name(&own), + "{own} launched this process and must stay in the set" + ); + } + + #[test] + fn foreign_process_names_are_never_reapable() { + for name in [ + "explorer.exe", + "sleep", + "tty7d", + "nottty7", + "tty7-app2", + "tty7.", + "", + ] { + assert!( + !is_reapable_daemon_name(name), + "{name:?} must be protected from the reap" + ); + } + } + + #[cfg(windows)] + #[test] + fn windows_matches_daemon_names_case_insensitively() { + assert!(is_reapable_daemon_name("TTY7-APP.EXE")); + assert!(is_reapable_daemon_name("Tty7-Server")); + assert!(is_reapable_daemon_name("TTY7")); + } + + #[test] + fn strip_exe_suffix_only_strips_a_trailing_exe() { + assert_eq!(strip_exe_suffix("tty7-app.exe"), "tty7-app"); + assert_eq!(strip_exe_suffix("tty7-app.EXE"), "tty7-app"); + assert_eq!(strip_exe_suffix("tty7-app"), "tty7-app"); + assert_eq!(strip_exe_suffix(".exe"), ""); + assert_eq!(strip_exe_suffix("exe"), "exe"); + } +} + #[cfg(all(test, unix))] mod tests { use super::*; @@ -387,8 +478,8 @@ mod tests { std::thread::sleep(Duration::from_millis(10)); } assert!( - !process_matches_own_exe(pid), - "sleep must not match the test binary; matching here would mean the reap could kill it" + !process_matches_daemon_exe(pid), + "sleep must not match any daemon name; matching here would mean the reap could kill it" ); let _ = child.kill(); diff --git a/crates/tty7-server/tests/pane_send_input.rs b/crates/tty7-server/tests/pane_send_input.rs new file mode 100644 index 00000000..56a40668 --- /dev/null +++ b/crates/tty7-server/tests/pane_send_input.rs @@ -0,0 +1,267 @@ +use std::io; +use std::path::PathBuf; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +use tty7_core::client::PaneClient; +use tty7_core::daemon::protocol::{DaemonMsg, ShellSpec, WinSize}; + +const READY_WITHIN: Duration = Duration::from_secs(30); +const STREAM_WITHIN: Duration = Duration::from_secs(30); +const EOF_WITHIN: Duration = Duration::from_secs(10); + +struct Daemon { + child: Child, + dir: tempfile::TempDir, +} + +impl Daemon { + fn start() -> Daemon { + let dir = tempfile::TempDir::new().unwrap(); + let child = Command::new(env!("CARGO_BIN_EXE_tty7-server")) + .arg("--daemon") + .arg("--config-dir") + .arg(dir.path()) + .env("TTY7_DATA_DIR", dir.path()) + .env("TTY7_CONTROL_SOCK", dir.path().join("control.sock")) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("start tty7-server --daemon"); + let daemon = Daemon { child, dir }; + daemon.await_ready(); + daemon + } + + fn pane_endpoint(&self) -> PathBuf { + let file = if cfg!(windows) { + "daemon.port" + } else { + "daemon.sock" + }; + self.dir.path().join(file) + } + + fn panes(&self) -> PaneClient { + PaneClient::at(self.pane_endpoint()) + } + + fn await_ready(&self) { + let deadline = Instant::now() + READY_WITHIN; + loop { + if self.panes().version().is_ok() { + return; + } + assert!( + Instant::now() < deadline, + "tty7-server did not open its pane endpoint within {READY_WITHIN:?}" + ); + std::thread::sleep(Duration::from_millis(50)); + } + } +} + +impl Drop for Daemon { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +fn size() -> WinSize { + WinSize { + cols: 100, + rows: 30, + cell_w: 8, + cell_h: 16, + } +} + +fn one_shot_shell(command: &str) -> ShellSpec { + if cfg!(windows) { + ShellSpec { + program: "cmd.exe".into(), + args: vec!["/d".into(), "/c".into(), command.into()], + args_are_tty7_defaults: false, + } + } else { + ShellSpec { + program: "/bin/sh".into(), + args: vec!["-c".into(), command.into()], + args_are_tty7_defaults: false, + } + } +} + +fn interactive_shell() -> ShellSpec { + if cfg!(windows) { + ShellSpec { + program: "cmd.exe".into(), + args: vec!["/d".into()], + args_are_tty7_defaults: false, + } + } else { + ShellSpec { + program: "/bin/sh".into(), + args: Vec::new(), + args_are_tty7_defaults: false, + } + } +} + +fn windows_contain(haystack: &[u8], needle: &[u8]) -> bool { + haystack.windows(needle.len()).any(|w| w == needle) +} + +fn collect_until(session: &mut tty7_core::client::PaneSession, marker: &[u8]) -> Vec { + let mut seen: Vec = Vec::new(); + loop { + match session.recv() { + Ok(DaemonMsg::Output(bytes)) | Ok(DaemonMsg::Snapshot(bytes)) => { + seen.extend_from_slice(&bytes); + if windows_contain(&seen, marker) { + return seen; + } + } + Ok(DaemonMsg::Exited { code }) => panic!( + "pane exited ({code:?}) before {:?} appeared; saw {:?}", + String::from_utf8_lossy(marker), + String::from_utf8_lossy(&seen) + ), + Ok(_) => {} + Err(e) => panic!( + "pane stream ended early: {e}; saw {:?}", + String::from_utf8_lossy(&seen) + ), + } + } +} + +#[test] +fn send_input_reaches_the_shell_without_displacing_the_controller() { + let daemon = Daemon::start(); + let panes = daemon.panes(); + let mut session = panes + .spawn(None, size(), Some(interactive_shell()), None, None) + .expect("spawn an interactive pane"); + let pane_id = session.pane_id(); + session + .set_recv_timeout(Some(STREAM_WITHIN)) + .expect("bound the stream reads"); + + panes + .send_input(pane_id, b"echo tty7_send_oneshot\r") + .expect("one-shot input is acknowledged"); + collect_until(&mut session, b"tty7_send_oneshot"); + + session + .input(b"echo tty7_still_controller\r") + .expect("the controller keeps its seat"); + collect_until(&mut session, b"tty7_still_controller"); + + session.kill().expect("kill the pane"); +} + +#[test] +fn send_input_to_a_missing_pane_answers_an_error() { + let daemon = Daemon::start(); + let err = daemon + .panes() + .send_input(u64::MAX, b"echo lost\r") + .expect_err("input into a pane that never existed must fail"); + assert!( + err.to_string().contains("no such pane"), + "the refusal was {err}" + ); +} + +#[test] +fn send_input_to_a_dead_pane_answers_an_error() { + let daemon = Daemon::start(); + let panes = daemon.panes(); + let mut session = panes + .spawn(None, size(), Some(one_shot_shell("exit 0")), None, None) + .expect("spawn a one-shot pane"); + let pane_id = session.pane_id(); + session + .set_recv_timeout(Some(STREAM_WITHIN)) + .expect("bound the stream reads"); + loop { + match session.recv() { + Ok(DaemonMsg::Exited { .. }) => break, + Ok(_) => {} + Err(e) => panic!("pane stream ended before Exited: {e}"), + } + } + + let err = panes + .send_input(pane_id, b"echo too_late\r") + .expect_err("input into an exited pane must fail"); + assert!( + err.to_string().contains("not running"), + "the refusal was {err}" + ); +} + +#[test] +fn a_preempting_attach_closes_the_displaced_controller_and_drops_its_input() { + let daemon = Daemon::start(); + let panes = daemon.panes(); + let mut first = panes + .spawn(None, size(), Some(interactive_shell()), None, None) + .expect("spawn an interactive pane"); + let pane_id = first.pane_id(); + first + .set_recv_timeout(Some(STREAM_WITHIN)) + .expect("bound the stream reads"); + first + .input(b"echo tty7_first_seated\r") + .expect("the first controller types"); + collect_until(&mut first, b"tty7_first_seated"); + + let mut second = panes.attach(pane_id, size()).expect("preempting attach"); + second + .set_recv_timeout(Some(STREAM_WITHIN)) + .expect("bound the second stream reads"); + + let _ = first.input(b"echo tty7_stale_input\r"); + + first + .set_recv_timeout(Some(EOF_WITHIN)) + .expect("bound the displaced reads"); + let deadline = Instant::now() + EOF_WITHIN + Duration::from_secs(5); + let eof = loop { + match first.recv() { + Ok(_) => { + assert!( + Instant::now() < deadline, + "the displaced controller's stream never ended" + ); + } + Err(e) => break e, + } + }; + assert_ne!( + eof.kind(), + io::ErrorKind::TimedOut, + "the displaced connection must be closed, not left dangling: {eof}" + ); + assert_ne!( + eof.kind(), + io::ErrorKind::WouldBlock, + "the displaced connection must be closed, not left dangling: {eof}" + ); + + second + .input(b"echo tty7_second_alive\r") + .expect("the new controller types"); + let seen = collect_until(&mut second, b"tty7_second_alive"); + assert!( + !windows_contain(&seen, b"tty7_stale_input"), + "input from the displaced controller must never reach the shell: {:?}", + String::from_utf8_lossy(&seen) + ); + + second.kill().expect("kill the pane"); +}