diff --git a/crates/tty7-core/src/daemon/pane.rs b/crates/tty7-core/src/daemon/pane.rs index 5f0dfaa7..83b42485 100644 --- a/crates/tty7-core/src/daemon/pane.rs +++ b/crates/tty7-core/src/daemon/pane.rs @@ -421,6 +421,15 @@ impl OutputGate { pub(crate) const OBSERVER_BUDGET: i64 = 8 * 1024 * 1024; +/// How long the death path will chase a child's exit status before giving up +/// and reporting `Exited { code: None }`. The pty hits EOF when the last slave +/// fd closes, which can land a few milliseconds ahead of the child becoming +/// reapable — that race is all this window exists to cover. Everything past it +/// is a pane that looks frozen to every attached client, so it stays short. +const EXIT_CODE_PROBE_WINDOW: Duration = Duration::from_millis(500); + +const EXIT_CODE_PROBE_INTERVAL: Duration = Duration::from_millis(10); + struct Observer { id: u64, tx: Sender, @@ -448,7 +457,13 @@ fn notify(st: &mut PaneState, msg: DaemonMsg) { if let Some(sub) = &st.subscriber { let _ = sub.send(msg.clone()); } - st.observers.retain(|obs| obs.tx.send(msg.clone()).is_ok()); + // Status traffic is charged the same budget as output. These messages are + // small, but an agent pane emits AgentStatus often enough that an observer + // which stopped draining would still queue without bound. Being over the + // line already disqualifies it; the message is not sized individually. + st.observers.retain(|obs| { + obs.gate.queued_bytes() < OBSERVER_BUDGET && obs.tx.send(msg.clone()).is_ok() + }); } fn fan_out_output(st: &mut PaneState, bytes: &[u8], gate: &OutputGate) { @@ -622,16 +637,25 @@ impl DaemonPane { death.probe_exit_code({ let child = child.clone(); move || { - let deadline = std::time::Instant::now() + Duration::from_secs(2); + let deadline = std::time::Instant::now() + EXIT_CODE_PROBE_WINDOW; loop { - let status = child.lock().ok()?.try_wait().ok()?; - if let Some(status) = status { - return Some(status.exit_code() as i32); + // try_lock, never lock: DaemonPane::drop holds this very + // mutex across a blocking child.wait(), and this probe runs + // on the reader thread on its way to announcing the death. + // Blocking for the lock would park the Exited notification + // behind a wait() with no bound of its own; the deadline + // below is the only thing clients should ever wait on. + if let Ok(mut child) = child.try_lock() { + match child.try_wait() { + Ok(Some(status)) => return Some(status.exit_code() as i32), + Ok(None) => {} + Err(_) => return None, + } } if std::time::Instant::now() >= deadline { return None; } - std::thread::sleep(Duration::from_millis(10)); + std::thread::sleep(EXIT_CODE_PROBE_INTERVAL); } } }); @@ -1327,9 +1351,18 @@ fn attach_subscriber(st: &mut PaneState, subscriber: Sender) -> u64 { st.subscriber_epoch } -fn observe_subscriber(st: &mut PaneState, observer: Sender, gate: Arc) -> u64 { +fn observe_subscriber( + st: &mut PaneState, + observer: Sender, + gate: Arc, +) -> u64 { st.observer_seq += 1; replay_state(st, &observer); + // The replay just queued the whole ring into this channel. Charge it, or + // the first budget check would read zero while a full scrollback is already + // sitting there unread — an observer that never drains would be allowed a + // ring plus a full budget before anyone noticed. + gate.add(st.ring.len); st.observers.push(Observer { id: st.observer_seq, tx: observer, @@ -3010,12 +3043,17 @@ mod tests { let (observer_tx, observer_rx) = mpsc::channel(); 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_eq!( + st.subscriber_epoch, epoch, + "observing must not bump the controller epoch" + ); assert!(st.subscriber.is_some(), "the controller keeps its seat"); assert!(matches!(observer_rx.try_recv(), Ok(DaemonMsg::Size(_)))); assert!(matches!(observer_rx.try_recv(), Ok(DaemonMsg::Snapshot(b)) if b == b"screen")); - assert!(matches!(observer_rx.try_recv(), Ok(DaemonMsg::Cwd(p)) if p == PathBuf::from("/work"))); + assert!( + matches!(observer_rx.try_recv(), Ok(DaemonMsg::Cwd(p)) if p == PathBuf::from("/work")) + ); assert!( controller_rx.try_recv().is_err(), "an observer joining must be invisible to the controller" @@ -3028,7 +3066,10 @@ mod tests { 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"); + assert!( + observer_rx.try_recv().is_err(), + "a departed observer hears nothing" + ); } #[test] @@ -3076,7 +3117,10 @@ mod tests { drop(observer_rx); notify(&mut st, DaemonMsg::Output(b"x".to_vec())); - assert!(st.observers.is_empty(), "a dead observer must not accumulate"); + assert!( + st.observers.is_empty(), + "a dead observer must not accumulate" + ); } #[test] @@ -3162,7 +3206,10 @@ mod tests { let (dead_tx, dead_rx) = mpsc::channel(); DeathReporter::new(move || dead_tx.send(()).unwrap()) .report(&with_observer_only, &AtomicBool::new(false)); - assert!(matches!(observer_rx.try_recv(), Ok(DaemonMsg::Exited { code: None }))); + assert!(matches!( + observer_rx.try_recv(), + Ok(DaemonMsg::Exited { code: None }) + )); assert!( dead_rx.try_recv().is_ok(), "read-only observers must not keep a dead pane in the registry" @@ -3181,8 +3228,14 @@ mod tests { let (dead_tx, dead_rx) = mpsc::channel(); DeathReporter::new(move || dead_tx.send(()).unwrap()) .report(&with_both, &AtomicBool::new(false)); - assert!(matches!(controller_rx.try_recv(), Ok(DaemonMsg::Exited { code: None }))); - assert!(matches!(observer_rx.try_recv(), Ok(DaemonMsg::Exited { code: None }))); + assert!(matches!( + controller_rx.try_recv(), + Ok(DaemonMsg::Exited { code: None }) + )); + assert!(matches!( + observer_rx.try_recv(), + Ok(DaemonMsg::Exited { code: None }) + )); assert!( dead_rx.try_recv().is_err(), "an attached death is still the detach path's to reclaim" diff --git a/crates/tty7-core/src/daemon/server.rs b/crates/tty7-core/src/daemon/server.rs index dd84d936..98de2333 100644 --- a/crates/tty7-core/src/daemon/server.rs +++ b/crates/tty7-core/src/daemon/server.rs @@ -372,6 +372,11 @@ fn handle_conn(stream: Stream, registry: Arc) -> anyhow::Result<()> { } }, + // The size is accepted but deliberately not applied: an observer is + // read-only, and resizing the pty for it would reflow the grid under + // the controller. The replay tells the observer the pane's real size; + // the field stays on the wire so a future observer-side viewport does + // not need another message kind. ClientMsg::Observe { pane_id, size: _ } => match registry.get(pane_id) { Some(pane) => stream_observer(pane, read_stream, write_stream), None => { @@ -663,8 +668,6 @@ fn observe_loop(read_stream: &mut R, refusals: &mpsc::Sender, id: u64, @@ -676,18 +679,15 @@ fn run_stream( ) -> anyhow::Result<()> { use std::io::Read as _; + // Blocking reads: when this controller is displaced its writer's channel + // closes, and the writer shuts our read side down on its way out (see + // spawn_writer), so the read below returns instead of hanging. let writer = spawn_writer(rx, write_stream, pane.gate()); - let _ = read_stream.set_read_timeout(Some(CONTROL_POLL_INTERVAL)); let mut killed = false; - 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, @@ -700,14 +700,12 @@ fn run_stream( 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); @@ -731,20 +729,14 @@ fn run_stream( 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; - } + Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue, Err(_) => break, } } + // Asked after the loop, before detach clears the seat: whoever no longer + // holds the epoch was preempted rather than having hung up on their own. + let displaced = !pane.controls(epoch); let reclaimable = pane.detach(epoch); if displaced { let _ = read_stream.shutdown(std::net::Shutdown::Both); @@ -809,6 +801,14 @@ fn spawn_writer( } let _ = write_stream.flush(); } + // This half and the connection's read half are try_clone'd views of + // one socket, so shutting the read side down here unblocks whoever + // is parked in read() on the other handle. The writer's channel + // closing IS the handover signal — a new controller replaces the + // subscriber, the old Sender drops, rx.recv() fails, and we land + // here — so the displaced reader wakes at once and nobody has to + // poll for it. + let _ = write_stream.shutdown(std::net::Shutdown::Read); }) .expect("spawn daemon writer thread") } @@ -1132,5 +1132,40 @@ mod tests { writer.join().unwrap(); drop(tx); } + + #[test] + fn a_closed_writer_channel_wakes_the_controller_parked_in_read() { + let (client, server) = UnixStream::pair().unwrap(); + let mut reader = server.try_clone().expect("clone the read half"); + let (tx, rx) = mpsc::channel::(); + let writer = spawn_writer(rx, server, Arc::new(crate::daemon::pane::OutputGate::new())); + + // Exactly where run_stream sits between frames. + let parked = thread::spawn(move || { + let mut buf = [0u8; 64]; + std::io::Read::read(&mut reader, &mut buf).ok() + }); + + // A handover: the pane seats a new controller and drops this one's + // Sender. Nothing else happens — no poll, no timeout. + drop(tx); + + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + while !parked.is_finished() && std::time::Instant::now() < deadline { + thread::sleep(std::time::Duration::from_millis(5)); + } + assert!( + parked.is_finished(), + "a displaced controller must be woken by the writer's shutdown; \ + if this hangs, run_stream is back to polling for the handover" + ); + assert_eq!( + parked.join().unwrap(), + Some(0), + "the woken read must report EOF, not a partial frame" + ); + writer.join().unwrap(); + drop(client); + } } } diff --git a/crates/tty7-core/src/daemon/spawn.rs b/crates/tty7-core/src/daemon/spawn.rs index 3540a69e..e544b3d0 100644 --- a/crates/tty7-core/src/daemon/spawn.rs +++ b/crates/tty7-core/src/daemon/spawn.rs @@ -53,9 +53,7 @@ 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] - } + Some(i) if name.is_char_boundary(i) && name[i..].eq_ignore_ascii_case(".exe") => &name[..i], _ => name, } } @@ -75,7 +73,9 @@ fn is_reapable_daemon_name(name: &str) -> bool { .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)) + || DAEMON_EXE_STEMS + .iter() + .any(|stem| exe_names_equal(stem, name)) } pub fn ensure_running() -> anyhow::Result<()> { diff --git a/crates/tty7-core/src/daemon/ssh/mod.rs b/crates/tty7-core/src/daemon/ssh/mod.rs index 9f465741..f93baa11 100644 --- a/crates/tty7-core/src/daemon/ssh/mod.rs +++ b/crates/tty7-core/src/daemon/ssh/mod.rs @@ -379,10 +379,16 @@ impl SshManager { let mut routes: Vec<_> = conns .iter() .map(|(key, slot)| { - let connected = slot - .try_lock() - .map(|weak| weak.upgrade().is_some_and(|conn| conn.is_alive())) - .unwrap_or(false); + let connected = match slot.try_lock() { + Ok(weak) => weak.upgrade().is_some_and(|conn| conn.is_alive()), + // The slot is held by whoever is opening or using this link + // right now. Busy is not down — and callers act on this: + // the CLI's `-m ` refuses to route over a link it + // is told is down, so guessing false here fails a perfectly + // live connection the moment it gets used. SshConnection:: + // is_alive resolves its own lock contention the same way. + Err(_) => true, + }; crate::daemon::control::RouteInfo { key: key.as_str().to_string(), kind: "ssh".to_string(), @@ -691,6 +697,36 @@ mod tests { } } + #[test] + fn a_busy_link_reads_as_connected_rather_than_down() { + let runtime = tokio::runtime::Builder::new_current_thread() + .build() + .expect("build test runtime"); + let mgr = SshManager { + runtime, + conns: Mutex::new(HashMap::new()), + forwards: SshForwardRegistry::default(), + probes: Mutex::new(HashMap::new()), + }; + let slot: ConnSlot = Arc::new(tokio::sync::Mutex::new(Weak::new())); + mgr.conns + .lock() + .unwrap() + .insert(ConnectionKey::from_spec(&base_spec()), slot.clone()); + + // Someone is mid-operation on this link: opening a channel, running the + // remote bootstrap probe, anything that holds the slot for a moment. + let _busy = slot.try_lock().expect("nobody else holds it in this test"); + + let routes = mgr.routes(); + assert_eq!(routes.len(), 1, "a busy link must still be listed"); + assert!( + routes[0].connected, + "a link whose slot is momentarily held is busy, not down — calling it \ + down makes `tty7 -m ` refuse to route over a live connection" + ); + } + #[test] fn connection_key_includes_jump_chain() { let mut with_jump = base_spec(); diff --git a/crates/tty7-core/src/host/server.rs b/crates/tty7-core/src/host/server.rs index 61c6a5cf..e1218ddb 100644 --- a/crates/tty7-core/src/host/server.rs +++ b/crates/tty7-core/src/host/server.rs @@ -1403,6 +1403,11 @@ mod sock { host: SharedHost, services: Services, ) -> io::Result { + // Anchor uptime here, not at the first Status request. The GUI can host + // the control listener too, and there `run_with` never runs — so + // without this the OnceLock is first set by whoever asks, and every + // answer reports a server that just started. + super::server_started(); let path = control_socket_path()?; let listener = bind_control_socket(&path)?; std::thread::Builder::new() @@ -1436,6 +1441,9 @@ mod wsock { host: SharedHost, services: Services, ) -> io::Result { + // See the unix arm: uptime is anchored where the listener opens, so a + // GUI-hosted control server does not report itself as freshly started. + super::server_started(); let path = control_endpoint_path()?; if let Ok(live) = transport::connect_endpoint(CONTROL_PORT_FILE) { drop(live); @@ -1585,8 +1593,7 @@ mod aggregate_tests { }; let client = client_with(services); - let ReplyOk::AgentStates(states) = client.call(ControlRequest::AgentStates).unwrap() - else { + let ReplyOk::AgentStates(states) = client.call(ControlRequest::AgentStates).unwrap() else { panic!("AgentStates must answer with ReplyOk::AgentStates"); }; assert_eq!(states.len(), 1); @@ -1600,8 +1607,7 @@ mod aggregate_tests { fn aggregates_still_answer_when_this_process_serves_no_panes() { let client = client_with(Services::none()); - let ReplyOk::AgentStates(states) = client.call(ControlRequest::AgentStates).unwrap() - else { + let ReplyOk::AgentStates(states) = client.call(ControlRequest::AgentStates).unwrap() else { panic!("AgentStates must answer with ReplyOk::AgentStates"); }; assert!(states.is_empty()); diff --git a/crates/tty7-server/tests/pane_send_input.rs b/crates/tty7-server/tests/pane_send_input.rs index 56a40668..49322b85 100644 --- a/crates/tty7-server/tests/pane_send_input.rs +++ b/crates/tty7-server/tests/pane_send_input.rs @@ -227,9 +227,12 @@ fn a_preempting_attach_closes_the_displaced_controller_and_drops_its_input() { let _ = first.input(b"echo tty7_stale_input\r"); - first - .set_recv_timeout(Some(EOF_WITHIN)) - .expect("bound the displaced reads"); + // Best effort: the handover closes this connection immediately, and + // setsockopt on a socket whose peer is gone fails (EINVAL on macOS). Not a + // problem — the earlier STREAM_WITHIN bound is still in force, so the reads + // below stay bounded either way, and an already-closed connection is + // precisely what this test wants to see. + let _ = first.set_recv_timeout(Some(EOF_WITHIN)); let deadline = Instant::now() + EOF_WITHIN + Duration::from_secs(5); let eof = loop { match first.recv() {