From fbb58ebfeffe8c95dcab99d00de8970dfd089b90 Mon Sep 17 00:00:00 2001 From: Anders Olsson Date: Fri, 7 Aug 2026 15:33:19 +0200 Subject: [PATCH] feat(protocol): wait-for schema and KDL parsing --- crates/xy-protocol/src/config.rs | 27 +++ crates/xy-protocol/src/kdl_parse.rs | 228 ++++++++++++++++++++++++- crates/xy-protocol/src/lib.rs | 5 +- crates/xy-supervisor/src/supervisor.rs | 7 +- 4 files changed, 262 insertions(+), 5 deletions(-) diff --git a/crates/xy-protocol/src/config.rs b/crates/xy-protocol/src/config.rs index 7780918..66256b5 100644 --- a/crates/xy-protocol/src/config.rs +++ b/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, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum WaitCondition { + Path(PathBuf), + Tcp(String), + Command { command: PathBuf, args: Vec }, +} + +#[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)] diff --git a/crates/xy-protocol/src/kdl_parse.rs b/crates/xy-protocol/src/kdl_parse.rs index 22d3a1f..8f7ca96 100644 --- a/crates/xy-protocol/src/kdl_parse.rs +++ b/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 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, 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 = Vec::new(); + let mut command: Option = None; + let mut args: Vec = 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 { 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", + .. + } + )); + } } diff --git a/crates/xy-protocol/src/lib.rs b/crates/xy-protocol/src/lib.rs index bf09517..669ed86 100644 --- a/crates/xy-protocol/src/lib.rs +++ b/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; diff --git a/crates/xy-supervisor/src/supervisor.rs b/crates/xy-supervisor/src/supervisor.rs index a32de08..094e6c0 100644 --- a/crates/xy-supervisor/src/supervisor.rs +++ b/crates/xy-supervisor/src/supervisor.rs @@ -22,7 +22,7 @@ pub enum SupervisorCmd { ack: oneshot::Sender<()>, }, Reconfigure { - new: ServerConfig, + new: Box, ack: oneshot::Sender<()>, }, Shutdown { @@ -170,7 +170,7 @@ impl SupervisorTask { 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 SupervisorTask { 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, } }