feat(protocol): wait-for schema and KDL parsing

crates/xy-protocol/src/config.rs

@@ -81,6 +81,33 @@ pub struct ServerConfig {
     pub restart: RestartConfig,
     #[serde(default)]
     pub stop: StopConfig,
+    #[serde(default)]
+    pub wait_for: Option<WaitForConfig>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum WaitCondition {
+    Path(PathBuf),
+    Tcp(String),
+    Command { command: PathBuf, args: Vec<String> },
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct WaitForConfig {
+    pub condition: WaitCondition,
+    #[serde(default = "default_wait_timeout", with = "humantime_serde")]
+    pub timeout: Duration,
+    #[serde(default = "default_wait_interval", with = "humantime_serde")]
+    pub interval: Duration,
+}
+
+pub fn default_wait_timeout() -> Duration {
+    Duration::from_secs(120)
+}
+
+pub fn default_wait_interval() -> Duration {
+    Duration::from_secs(1)
 }
 
 #[cfg(test)]

crates/xy-protocol/src/kdl_parse.rs

@@ -1,4 +1,7 @@
-use crate::{ConfigError, RestartConfig, RestartPolicy, ServerConfig, StopConfig};
+use crate::{
+    ConfigError, RestartConfig, RestartPolicy, ServerConfig, StopConfig, WaitCondition,
+    WaitForConfig, default_wait_interval, default_wait_timeout,
+};
 use kdl::{KdlDocument, KdlNode};
 use std::collections::BTreeMap;
 use std::path::{Path, PathBuf};
@@ -27,6 +30,7 @@ pub fn parse_server_config(
     let working_dir = optional_string_arg(&doc, "working-dir").map(PathBuf::from);
     let restart = parse_restart(&doc, source_path)?;
     let stop = parse_stop(&doc, source_path)?;
+    let wait_for = parse_wait_for(&doc, source_path)?;
 
     Ok(ServerConfig {
         name: name.to_string(),
@@ -37,6 +41,7 @@ pub fn parse_server_config(
         working_dir,
         restart,
         stop,
+        wait_for,
     })
 }
 
@@ -230,6 +235,97 @@ fn parse_stop(doc: &KdlDocument, path: &Path) -> Result<StopConfig, ConfigError>
     Ok(out)
 }
 
+fn expand_tilde(raw: &str, home: Option<&Path>) -> PathBuf {
+    let Some(rest) = raw.strip_prefix("~/") else {
+        return PathBuf::from(raw);
+    };
+
+    match home {
+        Some(home) => home.join(rest),
+        None => PathBuf::from(raw),
+    }
+}
+
+fn parse_wait_for(doc: &KdlDocument, path: &Path) -> Result<Option<WaitForConfig>, ConfigError> {
+    let Some(node) = find_node(doc, "wait-for") else {
+        return Ok(None);
+    };
+
+    let invalid = |message: String| ConfigError::InvalidValue {
+        path: path.to_path_buf(),
+        field: "wait-for",
+        message,
+    };
+
+    let Some(children) = node.children() else {
+        return Err(invalid(
+            "expected a block with exactly one condition".into(),
+        ));
+    };
+
+    let home = std::env::var_os("HOME").map(PathBuf::from);
+
+    let mut conditions: Vec<WaitCondition> = Vec::new();
+    let mut command: Option<PathBuf> = None;
+    let mut args: Vec<String> = Vec::new();
+    let mut timeout = default_wait_timeout();
+    let mut interval = default_wait_interval();
+
+    for child in children.nodes() {
+        match child.name().value() {
+            "path" => {
+                let raw = string_arg(child, "wait-for", path)?;
+
+                conditions.push(WaitCondition::Path(expand_tilde(&raw, home.as_deref())));
+            }
+            "tcp" => {
+                let addr = string_arg(child, "wait-for", path)?;
+
+                conditions.push(WaitCondition::Tcp(addr));
+            }
+            "command" => {
+                command = Some(PathBuf::from(string_arg(child, "wait-for", path)?));
+            }
+            "args" => {
+                args = child
+                    .entries()
+                    .iter()
+                    .filter_map(|e| e.value().as_string().map(str::to_string))
+                    .collect();
+            }
+            "timeout" => {
+                timeout = parse_duration_arg(child, "wait-for", path)?;
+            }
+            "interval" => {
+                interval = parse_duration_arg(child, "wait-for", path)?;
+            }
+            other => {
+                return Err(invalid(format!("unknown key `{other}`")));
+            }
+        }
+    }
+
+    if let Some(command) = command {
+        conditions.push(WaitCondition::Command { command, args });
+    } else if !args.is_empty() {
+        return Err(invalid("`args` requires `command`".into()));
+    }
+
+    match conditions.len() {
+        1 => Ok(Some(WaitForConfig {
+            condition: conditions.remove(0),
+            timeout,
+            interval,
+        })),
+        0 => Err(invalid(
+            "expected exactly one of `path`, `tcp` or `command`".into(),
+        )),
+        n => Err(invalid(format!(
+            "expected exactly one condition, found {n}"
+        ))),
+    }
+}
+
 fn string_arg(node: &KdlNode, field: &'static str, path: &Path) -> Result<String, ConfigError> {
     node.entries()
         .first()
@@ -444,4 +540,134 @@ stop {
             other => panic!("unexpected error: {other:?}"),
         }
     }
+
+    #[test]
+    fn parses_a_path_condition_with_defaults() {
+        let text = "command \"/bin/x\"\nport 1\nwait-for { path \"/var/run/d.sock\" }";
+        let cfg = parse_server_config("foo", text, p()).unwrap();
+
+        let wait = cfg.wait_for.unwrap();
+
+        assert_eq!(
+            wait.condition,
+            WaitCondition::Path(PathBuf::from("/var/run/d.sock"))
+        );
+        assert_eq!(wait.timeout, Duration::from_secs(120));
+        assert_eq!(wait.interval, Duration::from_secs(1));
+    }
+
+    #[test]
+    fn expands_a_leading_tilde_against_the_given_home() {
+        let home = PathBuf::from("/home/someone");
+
+        assert_eq!(
+            expand_tilde("~/.orbstack/run/docker.sock", Some(&home)),
+            PathBuf::from("/home/someone/.orbstack/run/docker.sock")
+        );
+    }
+
+    #[test]
+    fn leaves_non_tilde_paths_and_unknown_home_alone() {
+        let home = PathBuf::from("/home/someone");
+
+        assert_eq!(
+            expand_tilde("/var/run/d.sock", Some(&home)),
+            PathBuf::from("/var/run/d.sock")
+        );
+        assert_eq!(expand_tilde("~/a", None), PathBuf::from("~/a"));
+        assert_eq!(
+            expand_tilde("~weird/a", Some(&home)),
+            PathBuf::from("~weird/a")
+        );
+    }
+
+    #[test]
+    fn parses_a_tcp_condition_with_overrides() {
+        let text = "command \"/bin/x\"\nport 1\nwait-for { tcp \"127.0.0.1:5432\"\ntimeout \"30s\"\ninterval \"250ms\" }";
+        let cfg = parse_server_config("foo", text, p()).unwrap();
+
+        let wait = cfg.wait_for.unwrap();
+
+        assert_eq!(wait.condition, WaitCondition::Tcp("127.0.0.1:5432".into()));
+        assert_eq!(wait.timeout, Duration::from_secs(30));
+        assert_eq!(wait.interval, Duration::from_millis(250));
+    }
+
+    #[test]
+    fn parses_a_command_condition_from_sibling_nodes() {
+        let text = "command \"/bin/x\"\nport 1\nwait-for { command \"docker\"\nargs \"info\" }";
+        let cfg = parse_server_config("foo", text, p()).unwrap();
+
+        assert_eq!(
+            cfg.wait_for.unwrap().condition,
+            WaitCondition::Command {
+                command: PathBuf::from("docker"),
+                args: vec!["info".to_string()],
+            }
+        );
+    }
+
+    #[test]
+    fn absent_wait_for_leaves_no_gate() {
+        let text = "command \"/bin/x\"\nport 1";
+        let cfg = parse_server_config("foo", text, p()).unwrap();
+
+        assert!(cfg.wait_for.is_none());
+    }
+
+    #[test]
+    fn wait_for_without_a_condition_fails() {
+        let text = "command \"/bin/x\"\nport 1\nwait-for { timeout \"5s\" }";
+        let err = parse_server_config("foo", text, p()).unwrap_err();
+
+        assert!(matches!(
+            err,
+            ConfigError::InvalidValue {
+                field: "wait-for",
+                ..
+            }
+        ));
+    }
+
+    #[test]
+    fn wait_for_with_two_conditions_fails() {
+        let text = "command \"/bin/x\"\nport 1\nwait-for { path \"/a\"\ntcp \"127.0.0.1:1\" }";
+        let err = parse_server_config("foo", text, p()).unwrap_err();
+
+        assert!(matches!(
+            err,
+            ConfigError::InvalidValue {
+                field: "wait-for",
+                ..
+            }
+        ));
+    }
+
+    #[test]
+    fn wait_for_args_without_command_fails() {
+        let text = "command \"/bin/x\"\nport 1\nwait-for { args \"info\" }";
+        let err = parse_server_config("foo", text, p()).unwrap_err();
+
+        assert!(matches!(
+            err,
+            ConfigError::InvalidValue {
+                field: "wait-for",
+                ..
+            }
+        ));
+    }
+
+    #[test]
+    fn unknown_wait_for_key_fails() {
+        let text = "command \"/bin/x\"\nport 1\nwait-for { path \"/a\"\nnope \"x\" }";
+        let err = parse_server_config("foo", text, p()).unwrap_err();
+
+        assert!(matches!(
+            err,
+            ConfigError::InvalidValue {
+                field: "wait-for",
+                ..
+            }
+        ));
+    }
 }

crates/xy-protocol/src/lib.rs

@@ -6,7 +6,10 @@ pub mod kdl_parse;
 pub mod rpc;
 pub mod state;
 
-pub use config::{RestartConfig, RestartPolicy, ServerConfig, StopConfig};
+pub use config::{
+    RestartConfig, RestartPolicy, ServerConfig, StopConfig, WaitCondition, WaitForConfig,
+    default_wait_interval, default_wait_timeout,
+};
 pub use error::{ConfigError, RpcErrorCode};
 pub use kdl_parse::{load_all_configs, parse_server_config};
 pub use state::ServerState;

crates/xy-supervisor/src/supervisor.rs

@@ -22,7 +22,7 @@ pub enum SupervisorCmd {
         ack: oneshot::Sender<()>,
     },
     Reconfigure {
-        new: ServerConfig,
+        new: Box<ServerConfig>,
         ack: oneshot::Sender<()>,
     },
     Shutdown {
@@ -170,7 +170,7 @@ impl<S: Spawner> SupervisorTask<S> {
                             let _ = ack.send(());
                         }
                         SupervisorCmd::Reconfigure { new, ack } => {
-                            self.cfg = new;
+                            self.cfg = *new;
                             self.backoff =
                                 Backoff::new(self.cfg.restart.backoff_initial, self.cfg.restart.backoff_max);
                             self.retry_window = RetryWindow::new(
@@ -248,7 +248,7 @@ impl<S: Spawner> SupervisorTask<S> {
                                         Action::RetryNow
                                     }
                                     Some(SupervisorCmd::Reconfigure { new, ack }) => {
-                                        self.cfg = new;
+                                        self.cfg = *new;
                                         self.backoff = Backoff::new(
                                             self.cfg.restart.backoff_initial,
                                             self.cfg.restart.backoff_max,
@@ -390,6 +390,7 @@ mod tests {
             stop: StopConfig {
                 grace: Duration::from_millis(50),
             },
+            wait_for: None,
         }
     }