diff --git a/crates/xy-supervisor/src/lib.rs b/crates/xy-supervisor/src/lib.rs index 4807b59..95fcd6e 100644 --- a/crates/xy-supervisor/src/lib.rs +++ b/crates/xy-supervisor/src/lib.rs @@ -4,6 +4,7 @@ pub mod backoff; pub mod child; pub mod logs; pub mod policy; +mod ready; pub mod retry_window; pub mod supervisor; diff --git a/crates/xy-supervisor/src/ready.rs b/crates/xy-supervisor/src/ready.rs new file mode 100644 index 0000000..d7e8c01 --- /dev/null +++ b/crates/xy-supervisor/src/ready.rs @@ -0,0 +1,139 @@ +use std::process::Stdio; +use std::time::Duration; +use xy_protocol::WaitCondition; + +pub(crate) async fn is_ready(condition: &WaitCondition, budget: Duration) -> bool { + match condition { + WaitCondition::Path(p) => p.exists(), + WaitCondition::Tcp(addr) => { + let connect = tokio::net::TcpStream::connect(addr); + + matches!(tokio::time::timeout(budget, connect).await, Ok(Ok(_))) + } + WaitCondition::Command { command, args } => { + let mut cmd = tokio::process::Command::new(command); + + cmd.args(args) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .kill_on_drop(true); + + match tokio::time::timeout(budget, cmd.status()).await { + Ok(Ok(status)) => status.success(), + _ => false, + } + } + } +} + +pub(crate) fn describe(condition: &WaitCondition) -> String { + match condition { + WaitCondition::Path(p) => format!("path {}", p.display()), + WaitCondition::Tcp(addr) => format!("tcp {addr}"), + WaitCondition::Command { command, args } => { + let mut out = format!("command {}", command.display()); + + for arg in args { + out.push(' '); + out.push_str(arg); + } + + out + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::path::PathBuf; + + fn ms(n: u64) -> Duration { + Duration::from_millis(n) + } + + #[tokio::test] + async fn an_existing_path_is_ready() { + let tmp = tempfile::tempdir().unwrap(); + let file = tmp.path().join("sock"); + + std::fs::write(&file, b"").unwrap(); + + assert!(is_ready(&WaitCondition::Path(file), ms(50)).await); + } + + #[tokio::test] + async fn a_missing_path_is_not_ready() { + let condition = WaitCondition::Path(PathBuf::from("/definitely/not/here.sock")); + + assert!(!is_ready(&condition, ms(50)).await); + } + + #[tokio::test] + async fn a_listening_port_is_ready() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + assert!(is_ready(&WaitCondition::Tcp(addr), ms(500)).await); + } + + #[tokio::test] + async fn a_closed_port_is_not_ready() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + drop(listener); + + assert!(!is_ready(&WaitCondition::Tcp(addr), ms(500)).await); + } + + #[tokio::test] + async fn a_command_exiting_zero_is_ready() { + let condition = WaitCondition::Command { + command: PathBuf::from("/usr/bin/true"), + args: vec![], + }; + + assert!(is_ready(&condition, ms(500)).await); + } + + #[tokio::test] + async fn a_command_exiting_nonzero_is_not_ready() { + let condition = WaitCondition::Command { + command: PathBuf::from("/usr/bin/false"), + args: vec![], + }; + + assert!(!is_ready(&condition, ms(500)).await); + } + + #[tokio::test] + async fn a_command_that_outlasts_its_budget_is_not_ready() { + let condition = WaitCondition::Command { + command: PathBuf::from("/bin/sleep"), + args: vec!["5".to_string()], + }; + + assert!(!is_ready(&condition, ms(100)).await); + } + + #[test] + fn descriptions_name_the_condition() { + assert_eq!( + describe(&WaitCondition::Path(PathBuf::from("/a/b.sock"))), + "path /a/b.sock" + ); + assert_eq!( + describe(&WaitCondition::Tcp("127.0.0.1:5432".into())), + "tcp 127.0.0.1:5432" + ); + assert_eq!( + describe(&WaitCondition::Command { + command: PathBuf::from("docker"), + args: vec!["info".to_string()], + }), + "command docker info" + ); + } +}