feat(supervisor): evaluate wait-for conditions

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;
 

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"
+        );
+    }
+}