diff --git a/crates/tty7-core/src/client/control.rs b/crates/tty7-core/src/client/control.rs new file mode 100644 index 00000000..7f9967ff --- /dev/null +++ b/crates/tty7-core/src/client/control.rs @@ -0,0 +1,405 @@ +use std::io; +use std::path::Path; +use std::sync::Mutex; +use std::sync::mpsc::{Receiver, channel}; +use std::time::Duration; + +use crate::daemon::control::{ + ControlEvent, ControlHello, ControlHelloOk, ControlRequest, ControlResponse, EventSink, ReplyOk, +}; +use crate::daemon::router::{RouteAction, RouteChannel, RouteHeader, RouteTarget, negotiate}; +use crate::daemon::transport; + +pub struct ControlClient { + link: crate::daemon::control::ControlClient, + events: Mutex>, +} + +impl ControlClient { + pub fn connect(hello: &ControlHello) -> io::Result { + Self::over_stream(connect_local_control()?, hello) + } + + pub fn connect_at(endpoint: &Path, hello: &ControlHello) -> io::Result { + Self::over_stream(transport::connect_endpoint_at(endpoint)?, hello) + } + + pub fn routed(target: RouteTarget, hello: &ControlHello) -> io::Result { + Self::routed_over(transport::connect()?, target, hello) + } + + pub fn routed_over( + mut stream: transport::Stream, + target: RouteTarget, + hello: &ControlHello, + ) -> io::Result { + let header = RouteHeader { + target, + server_command: None, + channel: RouteChannel::Control, + action: RouteAction::Forward, + }; + negotiate(&mut stream, &header)?; + Self::over_stream(stream, hello) + } + + pub fn over_stream( + stream: transport::Stream, + hello: &ControlHello, + ) -> io::Result { + let (push, events) = channel(); + let sink: EventSink = Box::new(move |event| { + let _ = push.send(event); + }); + #[cfg(unix)] + let link = crate::daemon::control::ControlClient::over_unix(stream, hello, sink)?; + #[cfg(windows)] + let link = crate::daemon::control::ControlClient::over_tcp(stream, hello, sink)?; + Ok(ControlClient { + link, + events: Mutex::new(events), + }) + } + + pub fn hello(&self) -> &ControlHelloOk { + self.link.hello() + } + + pub fn is_connected(&self) -> bool { + self.link.is_connected() + } + + pub fn request(&self, req: ControlRequest) -> io::Result { + self.link.call(req) + } + + pub fn request_full(&self, req: ControlRequest, blob: &[u8]) -> io::Result { + self.link.call_full(req, blob) + } + + pub fn request_with_deadline( + &self, + req: ControlRequest, + blob: &[u8], + deadline: Duration, + ) -> io::Result { + self.link.call_with_deadline(req, blob, deadline) + } + + pub fn next_event(&self, wait: Duration) -> Option { + let events = self.events.lock().ok()?; + events.recv_timeout(wait).ok() + } + + pub fn events(&self) -> ControlEvents<'_> { + ControlEvents { client: self } + } + + pub fn close(&self) { + self.link.close(); + } +} + +impl std::fmt::Debug for ControlClient { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.link.fmt(f) + } +} + +pub struct ControlEvents<'a> { + client: &'a ControlClient, +} + +impl Iterator for ControlEvents<'_> { + type Item = ControlEvent; + + fn next(&mut self) -> Option { + let events = self.client.events.lock().ok()?; + events.recv().ok() + } +} + +fn connect_local_control() -> io::Result { + #[cfg(unix)] + { + let path = crate::host::server::control_socket_path()?; + transport::connect_endpoint_at(&path) + } + #[cfg(windows)] + { + crate::host::server::connect_control() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::client::stream_pair; + use crate::daemon::control::{ + CONTROL_VERSION, ControlClientMsg, ControlReply, ControlServerMsg, feature, + }; + use crate::daemon::protocol::{read_frame, write_frame}; + use crate::daemon::router::{ROUTE_KIND, RouteAck}; + use crate::host::{MTime, Meta}; + use std::io::{Read, Write}; + + const EVENT_WAIT: Duration = Duration::from_secs(10); + + fn hello() -> ControlHello { + ControlHello::host_rpc("unit-token", "unit-host") + } + + fn hello_ok() -> crate::daemon::control::ControlHelloOk { + crate::daemon::control::ControlHelloOk { + control_version: CONTROL_VERSION, + protocol_version: crate::daemon::protocol::PROTOCOL_VERSION, + build: "unit".into(), + separator: '/', + home: "/home/unit".into(), + features: vec![feature::CONTROL.into()], + instance: "unit-instance".into(), + } + } + + fn meta() -> Meta { + Meta { + is_dir: false, + is_symlink: false, + len: 8, + mtime: Some(MTime { + secs: 1_769_000_000, + nanos: 0, + }), + readonly: false, + } + } + + fn answer_hello(r: &mut impl Read, w: &mut impl Write) { + match ControlClientMsg::read(r).expect("read the client hello") { + ControlClientMsg::Hello(_) => {} + other => panic!("expected HELLO first, got {other:?}"), + } + ControlServerMsg::HelloOk(hello_ok()) + .encode(w) + .expect("answer HELLO_OK"); + } + + #[test] + fn requests_events_and_blobs_flow_through_the_wrapper() { + let (client_end, server_end) = stream_pair(); + let server = std::thread::spawn(move || { + let mut r = server_end.try_clone().expect("clone server end"); + let mut w = server_end; + answer_hello(&mut r, &mut w); + + match ControlClientMsg::read(&mut r).expect("read Ping") { + ControlClientMsg::Request { + req_id, + req: ControlRequest::Ping, + } => { + ControlServerMsg::Event(ControlEvent::LayoutResync) + .encode(&mut w) + .expect("push an event"); + ControlServerMsg::Response { + req_id, + reply: ControlReply::Ok(ReplyOk::Pong), + } + .encode(&mut w) + .expect("answer Ping"); + } + other => panic!("expected Ping, got {other:?}"), + } + + match ControlClientMsg::read(&mut r).expect("read WriteFile") { + ControlClientMsg::RequestBlob { + req_id, + req: ControlRequest::WriteFile { .. }, + blob, + } => { + assert_eq!(blob, b"payload", "the request blob rides the frame"); + ControlServerMsg::Response { + req_id, + reply: ControlReply::Ok(ReplyOk::Unit), + } + .encode(&mut w) + .expect("answer WriteFile"); + } + other => panic!("expected a WriteFile blob, got {other:?}"), + } + + match ControlClientMsg::read(&mut r).expect("read ReadFile") { + ControlClientMsg::Request { + req_id, + req: ControlRequest::ReadFile { .. }, + } => { + ControlServerMsg::ResponseBlob { + req_id, + reply: ControlReply::Ok(ReplyOk::FileMeta { meta: meta() }), + blob: b"contents".to_vec(), + } + .encode(&mut w) + .expect("answer ReadFile"); + } + other => panic!("expected ReadFile, got {other:?}"), + } + }); + + let client = ControlClient::over_stream(client_end, &hello()).expect("handshake"); + assert!(client.hello().has_feature(feature::CONTROL)); + + assert!(matches!( + client.request(ControlRequest::Ping).expect("ping"), + ReplyOk::Pong + )); + assert!(matches!( + client.next_event(EVENT_WAIT), + Some(ControlEvent::LayoutResync) + )); + + let wrote = client + .request_full( + ControlRequest::WriteFile { + path: "/tmp/x".into(), + }, + b"payload", + ) + .expect("write"); + assert!(matches!(wrote.reply, ReplyOk::Unit)); + + let read = client + .request_full( + ControlRequest::ReadFile { + path: "/tmp/x".into(), + max_bytes: 1024, + }, + &[], + ) + .expect("read"); + assert!(matches!(read.reply, ReplyOk::FileMeta { .. })); + assert_eq!(read.blob, b"contents", "the reply blob comes back intact"); + + client.close(); + server.join().expect("server thread"); + } + + #[test] + fn the_event_iterator_ends_when_the_link_goes_down() { + let (client_end, server_end) = stream_pair(); + let server = std::thread::spawn(move || { + let mut r = server_end.try_clone().expect("clone server end"); + let mut w = server_end; + answer_hello(&mut r, &mut w); + ControlServerMsg::Event(ControlEvent::PaneExited { + pane_id: 3, + code: Some(0), + }) + .encode(&mut w) + .expect("push an event"); + }); + + let client = ControlClient::over_stream(client_end, &hello()).expect("handshake"); + server.join().expect("server thread"); + + let mut events = client.events(); + assert!(matches!( + events.next(), + Some(ControlEvent::PaneExited { + pane_id: 3, + code: Some(0) + }) + )); + assert!( + events.next().is_none(), + "a dead link must end the iterator, not block it" + ); + } + + #[test] + fn routed_over_negotiates_before_the_handshake() { + let (client_end, server_end) = stream_pair(); + let server = std::thread::spawn(move || { + let mut r = server_end.try_clone().expect("clone server end"); + let mut w = server_end; + + let (kind, payload) = read_frame(&mut r).expect("read the route header"); + assert_eq!(kind, ROUTE_KIND, "the ROUTE frame must come first"); + let header = RouteHeader::decode(&payload).expect("decode the header"); + assert_eq!(header.channel, RouteChannel::Control); + assert!(matches!(header.target, RouteTarget::Wsl { ref distro } if distro == "Ubuntu")); + + let ack = serde_json::to_vec(&RouteAck { + ok: true, + link: Some("unit".into()), + action: Some(RouteAction::Forward), + error: None, + }) + .expect("encode the ack"); + write_frame(&mut w, ROUTE_KIND, &ack).expect("send the ack"); + + answer_hello(&mut r, &mut w); + match ControlClientMsg::read(&mut r).expect("read Ping") { + ControlClientMsg::Request { + req_id, + req: ControlRequest::Ping, + } => { + ControlServerMsg::Response { + req_id, + reply: ControlReply::Ok(ReplyOk::Pong), + } + .encode(&mut w) + .expect("answer Ping"); + } + other => panic!("expected Ping, got {other:?}"), + } + }); + + let client = ControlClient::routed_over( + client_end, + RouteTarget::Wsl { + distro: "Ubuntu".into(), + }, + &hello(), + ) + .expect("routed handshake"); + assert!(matches!( + client.request(ControlRequest::Ping).expect("ping"), + ReplyOk::Pong + )); + + client.close(); + server.join().expect("server thread"); + } + + #[test] + fn a_refused_route_surfaces_the_servers_reason() { + let (client_end, server_end) = stream_pair(); + let server = std::thread::spawn(move || { + let mut r = server_end.try_clone().expect("clone server end"); + let mut w = server_end; + let (kind, _) = read_frame(&mut r).expect("read the route header"); + assert_eq!(kind, ROUTE_KIND); + let ack = serde_json::to_vec(&RouteAck { + ok: false, + link: None, + action: None, + error: Some("no such distro".into()), + }) + .expect("encode the refusal"); + write_frame(&mut w, ROUTE_KIND, &ack).expect("send the refusal"); + }); + + let err = ControlClient::routed_over( + client_end, + RouteTarget::Wsl { + distro: "Nowhere".into(), + }, + &hello(), + ) + .expect_err("a refused route must not hand back a client"); + assert!( + err.to_string().contains("no such distro"), + "the refusal reason was lost: {err}" + ); + server.join().expect("server thread"); + } +} diff --git a/crates/tty7-core/src/client/mod.rs b/crates/tty7-core/src/client/mod.rs new file mode 100644 index 00000000..e2e654bb --- /dev/null +++ b/crates/tty7-core/src/client/mod.rs @@ -0,0 +1,31 @@ +mod control; +mod pane; + +pub use control::{ControlClient, ControlEvents}; +pub use pane::{PaneClient, PaneInput, PaneOutput, PaneSession}; + +pub use crate::daemon::control::{ + ControlEvent, ControlHello, ControlHelloOk, ControlRequest, ControlResponse, ReplyOk, +}; +pub use crate::daemon::protocol::{DaemonMsg, DaemonVersion, PaneInfo, ShellSpec, WinSize}; +pub use crate::daemon::router::RouteTarget; + +#[cfg(test)] +pub(crate) fn stream_pair() -> ( + crate::daemon::transport::Stream, + crate::daemon::transport::Stream, +) { + #[cfg(unix)] + { + std::os::unix::net::UnixStream::pair().expect("socketpair") + } + #[cfg(windows)] + { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind loopback"); + let addr = listener.local_addr().expect("bound addr"); + let connecting = + std::thread::spawn(move || std::net::TcpStream::connect(addr).expect("connect back")); + let (accepted, _) = listener.accept().expect("accept"); + (connecting.join().expect("connector thread"), accepted) + } +} diff --git a/crates/tty7-core/src/client/pane.rs b/crates/tty7-core/src/client/pane.rs new file mode 100644 index 00000000..bcff2147 --- /dev/null +++ b/crates/tty7-core/src/client/pane.rs @@ -0,0 +1,434 @@ +use std::io::{self, Read as _}; +use std::path::PathBuf; +use std::time::Duration; + +use crate::daemon::protocol::{ + ClientMsg, DaemonMsg, DaemonVersion, PaneInfo, ShellSpec, WinSize, is_error_kind, + peek_frame_kind, take_frame, +}; +use crate::daemon::transport; + +const OPEN_REPLY_WAIT: Duration = Duration::from_secs(15); + +#[derive(Clone, Debug, Default)] +enum PaneEndpoint { + #[default] + Local, + At(PathBuf), +} + +#[derive(Clone, Debug, Default)] +pub struct PaneClient { + endpoint: PaneEndpoint, +} + +impl PaneClient { + pub fn local() -> PaneClient { + PaneClient { + endpoint: PaneEndpoint::Local, + } + } + + pub fn at(endpoint: impl Into) -> PaneClient { + PaneClient { + endpoint: PaneEndpoint::At(endpoint.into()), + } + } + + fn open(&self) -> io::Result { + match &self.endpoint { + PaneEndpoint::Local => transport::connect(), + PaneEndpoint::At(path) => transport::connect_endpoint_at(path), + } + } + + pub fn list(&self) -> io::Result> { + let mut stream = self.open()?; + ClientMsg::List.encode(&mut stream)?; + match DaemonMsg::read(&mut stream)? { + DaemonMsg::PaneList(panes) => Ok(panes), + DaemonMsg::Error(message) => Err(io::Error::other(message)), + other => Err(unexpected_reply("List", &other)), + } + } + + pub fn version(&self) -> io::Result { + let mut stream = self.open()?; + ClientMsg::Version.encode(&mut stream)?; + match DaemonMsg::read(&mut stream)? { + DaemonMsg::Version(version) => Ok(version), + DaemonMsg::Error(message) => Err(io::Error::other(message)), + other => Err(unexpected_reply("Version", &other)), + } + } + + pub fn kill(&self, pane_id: u64) -> io::Result<()> { + let mut stream = self.open()?; + ClientMsg::Kill { pane_id }.encode(&mut stream) + } + + pub fn spawn( + &self, + cwd: Option, + size: WinSize, + shell: Option, + owner: Option, + ) -> io::Result { + PaneSession::spawn_over(self.open()?, cwd, size, shell, owner, OPEN_REPLY_WAIT) + } + + pub fn attach(&self, pane_id: u64, size: WinSize) -> io::Result { + PaneSession::attach_over(self.open()?, pane_id, size, OPEN_REPLY_WAIT) + } +} + +#[derive(Debug)] +pub struct PaneSession { + input: PaneInput, + output: PaneOutput, +} + +impl PaneSession { + pub(crate) fn spawn_over( + mut stream: transport::Stream, + cwd: Option, + size: WinSize, + shell: Option, + owner: Option, + reply_wait: Duration, + ) -> io::Result { + ClientMsg::Spawn { + cwd, + size, + shell, + owner, + } + .encode(&mut stream)?; + let mut session = PaneSession::over(stream, 0)?; + session.set_recv_timeout(Some(reply_wait))?; + let first = session.recv(); + session.set_recv_timeout(None)?; + match first { + Ok(DaemonMsg::Spawned { pane_id }) => { + session.input.pane_id = pane_id; + Ok(session) + } + Ok(DaemonMsg::Error(message)) => { + Err(io::Error::other(format!("daemon refused Spawn: {message}"))) + } + Ok(other) => Err(unexpected_reply("Spawn", &other)), + Err(e) if would_block(&e) => Err(io::Error::new( + io::ErrorKind::TimedOut, + format!("no answer to Spawn within {reply_wait:?}"), + )), + Err(e) => Err(e), + } + } + + pub(crate) fn attach_over( + mut stream: transport::Stream, + pane_id: u64, + size: WinSize, + reply_wait: Duration, + ) -> io::Result { + ClientMsg::Attach { pane_id, size }.encode(&mut stream)?; + let mut session = PaneSession::over(stream, pane_id)?; + session.set_recv_timeout(Some(reply_wait))?; + let verdict = session.output.refusal_check(pane_id); + session.set_recv_timeout(None)?; + verdict?; + Ok(session) + } + + fn over(stream: transport::Stream, pane_id: u64) -> io::Result { + let reader = stream.try_clone()?; + Ok(PaneSession { + input: PaneInput { + writer: stream, + pane_id, + }, + output: PaneOutput { + reader, + buffered: Vec::new(), + }, + }) + } + + pub fn pane_id(&self) -> u64 { + self.input.pane_id() + } + + pub fn input(&mut self, bytes: &[u8]) -> io::Result<()> { + self.input.input(bytes) + } + + pub fn resize(&mut self, size: WinSize) -> io::Result<()> { + self.input.resize(size) + } + + pub fn detach(self) -> io::Result<()> { + self.input.detach() + } + + pub fn kill(self) -> io::Result<()> { + self.input.kill() + } + + pub fn recv(&mut self) -> io::Result { + self.output.recv() + } + + pub fn set_recv_timeout(&self, wait: Option) -> io::Result<()> { + self.output.set_recv_timeout(wait) + } + + pub fn split(self) -> (PaneInput, PaneOutput) { + (self.input, self.output) + } +} + +#[derive(Debug)] +pub struct PaneInput { + writer: transport::Stream, + pane_id: u64, +} + +impl PaneInput { + pub fn pane_id(&self) -> u64 { + self.pane_id + } + + pub fn input(&mut self, bytes: &[u8]) -> io::Result<()> { + ClientMsg::Input(bytes.to_vec()).encode(&mut self.writer) + } + + pub fn resize(&mut self, size: WinSize) -> io::Result<()> { + ClientMsg::Resize(size).encode(&mut self.writer) + } + + pub fn detach(mut self) -> io::Result<()> { + ClientMsg::Detach.encode(&mut self.writer) + } + + pub fn kill(mut self) -> io::Result<()> { + ClientMsg::Kill { + pane_id: self.pane_id, + } + .encode(&mut self.writer) + } +} + +#[derive(Debug)] +pub struct PaneOutput { + reader: transport::Stream, + buffered: Vec, +} + +impl PaneOutput { + pub fn recv(&mut self) -> io::Result { + loop { + if let Some((kind, payload)) = take_frame(&mut self.buffered)? { + return DaemonMsg::from_frame(kind, payload); + } + let mut scratch = [0u8; 16 * 1024]; + match self.reader.read(&mut scratch) { + Ok(0) => { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "the daemon closed the pane connection", + )); + } + Ok(n) => self.buffered.extend_from_slice(&scratch[..n]), + Err(e) if e.kind() == io::ErrorKind::Interrupted => continue, + Err(e) => return Err(e), + } + } + } + + pub fn set_recv_timeout(&self, wait: Option) -> io::Result<()> { + self.reader.set_read_timeout(wait) + } + + fn refusal_check(&mut self, pane_id: u64) -> io::Result<()> { + let mut scratch = [0u8; 4096]; + loop { + if let Some(kind) = peek_frame_kind(&self.buffered) { + if !is_error_kind(kind) { + return Ok(()); + } + return Err(self.refusal(pane_id)); + } + match self.reader.read(&mut scratch) { + Ok(0) => { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + format!( + "the daemon closed the connection without answering \ + Attach for pane {pane_id}" + ), + )); + } + Ok(n) => self.buffered.extend_from_slice(&scratch[..n]), + Err(e) if e.kind() == io::ErrorKind::Interrupted => continue, + Err(e) if would_block(&e) => return Ok(()), + Err(e) => return Err(e), + } + } + } + + fn refusal(&mut self, pane_id: u64) -> io::Error { + match self.recv() { + Ok(DaemonMsg::Error(message)) => { + io::Error::other(format!("daemon refused Attach: {message}")) + } + Ok(other) => unexpected_reply("Attach", &other), + Err(_) => io::Error::other(format!("daemon refused Attach for pane {pane_id}")), + } + } +} + +fn would_block(e: &io::Error) -> bool { + matches!( + e.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ) +} + +fn unexpected_reply(request: &str, got: &DaemonMsg) -> io::Error { + io::Error::new( + io::ErrorKind::InvalidData, + format!("unexpected daemon reply to {request}: {got:?}"), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::client::stream_pair; + + const REPLY_WAIT: Duration = Duration::from_secs(10); + + fn size() -> WinSize { + WinSize { + cols: 80, + rows: 24, + cell_w: 8, + cell_h: 17, + } + } + + #[test] + fn a_session_speaks_the_pane_protocol_end_to_end() { + let (client_end, server_end) = stream_pair(); + let server = std::thread::spawn(move || { + let mut r = server_end.try_clone().expect("clone server end"); + let mut w = server_end; + + match ClientMsg::read(&mut r).expect("read Spawn") { + ClientMsg::Spawn { + owner: Some(owner), .. + } => assert_eq!(owner, "unit"), + other => panic!("expected an owned Spawn, got {other:?}"), + } + DaemonMsg::Spawned { pane_id: 7 } + .encode(&mut w) + .expect("confirm the spawn"); + DaemonMsg::Output(b"hello from the pane".to_vec()) + .encode(&mut w) + .expect("stream output"); + + match ClientMsg::read(&mut r).expect("read Input") { + ClientMsg::Input(bytes) => assert_eq!(bytes, b"ls\r"), + other => panic!("expected Input, got {other:?}"), + } + match ClientMsg::read(&mut r).expect("read Resize") { + ClientMsg::Resize(new_size) => assert_eq!(new_size.cols, 120), + other => panic!("expected Resize, got {other:?}"), + } + DaemonMsg::Exited { code: Some(0) } + .encode(&mut w) + .expect("report the exit"); + match ClientMsg::read(&mut r).expect("read Detach") { + ClientMsg::Detach => {} + other => panic!("expected Detach, got {other:?}"), + } + }); + + let mut session = + PaneSession::spawn_over(client_end, None, size(), None, Some("unit".into()), REPLY_WAIT) + .expect("spawn"); + assert_eq!(session.pane_id(), 7); + + match session.recv().expect("first stream message") { + DaemonMsg::Output(bytes) => assert_eq!(bytes, b"hello from the pane"), + other => panic!("expected Output, got {other:?}"), + } + + session.input(b"ls\r").expect("send input"); + session + .resize(WinSize { + cols: 120, + ..size() + }) + .expect("send resize"); + match session.recv().expect("exit message") { + DaemonMsg::Exited { code: Some(0) } => {} + other => panic!("expected Exited, got {other:?}"), + } + session.detach().expect("detach"); + server.join().expect("server thread"); + } + + #[test] + fn an_attach_refusal_carries_the_daemons_message() { + let (client_end, server_end) = stream_pair(); + let server = std::thread::spawn(move || { + let mut r = server_end.try_clone().expect("clone server end"); + let mut w = server_end; + match ClientMsg::read(&mut r).expect("read Attach") { + ClientMsg::Attach { pane_id, .. } => assert_eq!(pane_id, 42), + other => panic!("expected Attach, got {other:?}"), + } + DaemonMsg::Error("no such pane 42".into()) + .encode(&mut w) + .expect("refuse the attach"); + }); + + let err = PaneSession::attach_over(client_end, 42, size(), REPLY_WAIT) + .expect_err("attaching to a missing pane must fail"); + assert!( + err.to_string().contains("no such pane 42"), + "the daemon's reason was lost: {err}" + ); + server.join().expect("server thread"); + } + + #[test] + fn an_accepted_attach_keeps_the_replay_it_peeked_at() { + let (client_end, server_end) = stream_pair(); + let server = std::thread::spawn(move || { + let mut r = server_end.try_clone().expect("clone server end"); + let mut w = server_end; + match ClientMsg::read(&mut r).expect("read Attach") { + ClientMsg::Attach { pane_id, .. } => assert_eq!(pane_id, 9), + other => panic!("expected Attach, got {other:?}"), + } + DaemonMsg::Size(size()).encode(&mut w).expect("replay size"); + DaemonMsg::Snapshot(b"screen contents".to_vec()) + .encode(&mut w) + .expect("replay snapshot"); + }); + + let mut session = + PaneSession::attach_over(client_end, 9, size(), REPLY_WAIT).expect("attach"); + match session.recv().expect("replayed size") { + DaemonMsg::Size(_) => {} + other => panic!("expected Size, got {other:?}"), + } + match session.recv().expect("replayed snapshot") { + DaemonMsg::Snapshot(bytes) => assert_eq!(bytes, b"screen contents"), + other => panic!("expected Snapshot, got {other:?}"), + } + server.join().expect("server thread"); + } +} diff --git a/crates/tty7-core/src/daemon/transport.rs b/crates/tty7-core/src/daemon/transport.rs index 153f2d5b..29b8bf83 100644 --- a/crates/tty7-core/src/daemon/transport.rs +++ b/crates/tty7-core/src/daemon/transport.rs @@ -63,6 +63,10 @@ mod imp_unix { let path = socket_path().ok_or_else(|| { io::Error::other("could not resolve daemon socket path (no config dir)") })?; + connect_endpoint_at(&path) + } + + pub fn connect_endpoint_at(path: &Path) -> io::Result { let stream = UnixStream::connect(path)?; tune(&stream); Ok(stream) @@ -336,10 +340,7 @@ mod imp_windows { let (port, token) = read_port_file() .filter(|(p, _)| *p != 0) .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "no daemon port file"))?; - let mut stream = TcpStream::connect(loopback(port))?; - tune(&stream); - stream.write_all(&token)?; - Ok(stream) + connect_with_token(port, &token) } pub fn authenticate(stream: &mut Stream) -> io::Result<()> { @@ -412,9 +413,26 @@ mod imp_windows { let (port, token) = read_port_file_named(file) .filter(|(p, _)| *p != 0) .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, format!("no {file} file")))?; + connect_with_token(port, &token) + } + + pub fn connect_endpoint_at(path: &std::path::Path) -> io::Result { + let contents = std::fs::read_to_string(path)?; + let (port, token) = parse_port_file(&contents) + .filter(|(p, _)| *p != 0) + .ok_or_else(|| { + io::Error::new( + io::ErrorKind::NotFound, + format!("no listener recorded at {}", path.display()), + ) + })?; + connect_with_token(port, &token) + } + + fn connect_with_token(port: u16, token: &Token) -> io::Result { let mut stream = TcpStream::connect(loopback(port))?; tune(&stream); - stream.write_all(&token)?; + stream.write_all(token)?; Ok(stream) } diff --git a/crates/tty7-core/src/lib.rs b/crates/tty7-core/src/lib.rs index 606bf796..3b96b2be 100644 --- a/crates/tty7-core/src/lib.rs +++ b/crates/tty7-core/src/lib.rs @@ -1,3 +1,4 @@ +pub mod client; pub mod core; pub mod daemon; pub mod host; diff --git a/crates/tty7-server/tests/client_lib.rs b/crates/tty7-server/tests/client_lib.rs new file mode 100644 index 00000000..d77bd8d0 --- /dev/null +++ b/crates/tty7-server/tests/client_lib.rs @@ -0,0 +1,343 @@ +use std::path::PathBuf; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +use tty7_core::client::{ControlClient, PaneClient}; +use tty7_core::core::machine::{LayoutDelta, PaneSeed}; +use tty7_core::daemon::control::{ControlEvent, ControlHello, ControlRequest, ReplyOk, feature}; +use tty7_core::daemon::protocol::{DaemonMsg, PROTOCOL_VERSION, ShellSpec, WinSize}; + +const READY_WITHIN: Duration = Duration::from_secs(30); +const STREAM_WITHIN: Duration = Duration::from_secs(30); + +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 control_endpoint(&self) -> PathBuf { + let file = if cfg!(windows) { + "control.port" + } else { + "control.sock" + }; + self.dir.path().join(file) + } + + fn panes(&self) -> PaneClient { + PaneClient::at(self.pane_endpoint()) + } + + fn control(&self, name: &str) -> ControlClient { + ControlClient::connect_at(&self.control_endpoint(), &hello(name)) + .expect("control handshake with the spawned server") + } + + fn await_ready(&self) { + let deadline = Instant::now() + READY_WITHIN; + loop { + let control_up = + ControlClient::connect_at(&self.control_endpoint(), &hello("probe")).is_ok(); + let panes_up = self.panes().version().is_ok(); + if control_up && panes_up { + return; + } + assert!( + Instant::now() < deadline, + "tty7-server did not open its endpoints 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 hello(name: &str) -> ControlHello { + ControlHello::host_rpc(name, name) +} + +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 seed(pane: u64) -> PaneSeed { + PaneSeed { + pane, + cwd: Some("/home/me/proj".into()), + ssh_spec: None, + agent: None, + } +} + +fn collect_until( + session: &mut tty7_core::client::PaneSession, + marker: &[u8], +) -> (Vec, Option>) { + 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, None); + } + } + Ok(DaemonMsg::Exited { code }) => return (seen, Some(code)), + Ok(_) => {} + Err(e) => panic!( + "pane stream ended early: {e}; saw {:?}", + String::from_utf8_lossy(&seen) + ), + } + } +} + +fn drain_until_exit(session: &mut tty7_core::client::PaneSession) -> Vec { + let mut seen: Vec = Vec::new(); + loop { + match session.recv() { + Ok(DaemonMsg::Output(bytes)) | Ok(DaemonMsg::Snapshot(bytes)) => { + seen.extend_from_slice(&bytes); + } + Ok(DaemonMsg::Exited { .. }) => return seen, + Ok(_) => {} + Err(e) => panic!( + "pane stream ended before Exited: {e}; saw {:?}", + String::from_utf8_lossy(&seen) + ), + } + } +} + +fn windows_contain(haystack: &[u8], needle: &[u8]) -> bool { + haystack.windows(needle.len()).any(|w| w == needle) +} + +#[test] +fn the_pane_daemon_reports_its_dialect() { + let daemon = Daemon::start(); + let version = daemon.panes().version().expect("query the version"); + assert_eq!(version.protocol, PROTOCOL_VERSION); +} + +#[test] +fn control_requests_build_the_tree_and_events_reach_the_other_client() { + let daemon = Daemon::start(); + let writer = daemon.control("writer"); + assert!( + writer.hello().has_feature(feature::MACHINE_TREE), + "features were {:?}", + writer.hello().features + ); + + let watcher = daemon.control("watcher"); + watcher + .request(ControlRequest::Ping) + .expect("the watcher is live before the writer acts"); + + let ws = match writer + .request(ControlRequest::WorkspaceCreate { + name: Some("api".into()), + workspace: None, + }) + .expect("create a workspace") + { + ReplyOk::WorkspaceTree(ws) => *ws, + other => panic!("expected WorkspaceTree, got {other:?}"), + }; + match writer + .request(ControlRequest::TabCreate { + workspace: ws.id, + at: None, + pane: seed(1), + tab: None, + }) + .expect("create a tab") + { + ReplyOk::TabTree(_) => {} + other => panic!("expected TabTree, got {other:?}"), + } + + let machine = match writer + .request(ControlRequest::MachineGet) + .expect("fetch the machine tree") + { + ReplyOk::MachineTree(m) => *m, + other => panic!("expected MachineTree, got {other:?}"), + }; + assert_eq!(machine.workspaces.len(), 1); + assert_eq!(machine.workspaces[0].tabs[0].root.pane_ids(), vec![1]); + + let key = ws.id.to_string(); + let deadline = Instant::now() + STREAM_WITHIN; + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + assert!( + !remaining.is_zero(), + "the watcher never saw the WorkspaceCreated delta for {key}" + ); + match watcher.next_event(remaining) { + Some(ControlEvent::Layout { workspace, delta }) + if workspace == key + && matches!(delta, LayoutDelta::WorkspaceCreated { .. }) => + { + break; + } + Some(_) => {} + None => {} + } + } +} + +#[test] +fn a_spawned_pane_streams_its_output_and_its_exit() { + let daemon = Daemon::start(); + let mut session = daemon + .panes() + .spawn( + None, + size(), + Some(one_shot_shell("echo tty7_pane_roundtrip")), + Some("client-lib-test".into()), + ) + .expect("spawn a one-shot pane"); + assert_ne!(session.pane_id(), 0, "the daemon must name the pane"); + session + .set_recv_timeout(Some(STREAM_WITHIN)) + .expect("bound the stream reads"); + + let seen = drain_until_exit(&mut session); + assert!( + windows_contain(&seen, b"tty7_pane_roundtrip"), + "output was {:?}", + String::from_utf8_lossy(&seen) + ); +} + +#[test] +fn input_reaches_the_shell_and_a_reattach_replays_it() { + let daemon = Daemon::start(); + let panes = daemon.panes(); + let mut session = panes + .spawn(None, size(), Some(interactive_shell()), None) + .expect("spawn an interactive pane"); + let pane_id = session.pane_id(); + session + .set_recv_timeout(Some(STREAM_WITHIN)) + .expect("bound the stream reads"); + + session + .input(b"echo tty7_attach_replay\r") + .expect("type into the pane"); + let (_, exit) = collect_until(&mut session, b"tty7_attach_replay"); + assert!(exit.is_none(), "the shell must still be running"); + session.detach().expect("detach"); + + let listed = panes.list().expect("list panes"); + let entry = listed + .iter() + .find(|p| p.pane_id == pane_id) + .expect("the detached pane is still listed"); + assert!(entry.alive, "the detached pane is still alive"); + + let mut reattached = panes.attach(pane_id, size()).expect("reattach"); + reattached + .set_recv_timeout(Some(STREAM_WITHIN)) + .expect("bound the replay reads"); + let (_, exit) = collect_until(&mut reattached, b"tty7_attach_replay"); + assert!(exit.is_none(), "the replayed pane is still running"); + + let refused = panes + .attach(u64::MAX, size()) + .expect_err("attaching to a pane that never existed must fail"); + assert!( + refused.to_string().contains("no such pane"), + "the refusal was {refused}" + ); + + reattached.kill().expect("kill the pane"); + let deadline = Instant::now() + STREAM_WITHIN; + loop { + let listed = panes.list().expect("list panes after the kill"); + let gone = !listed.iter().any(|p| p.pane_id == pane_id && p.alive); + if gone { + break; + } + assert!( + Instant::now() < deadline, + "pane {pane_id} was still listed alive after Kill: {listed:?}" + ); + std::thread::sleep(Duration::from_millis(50)); + } +}