diff --git a/README.md b/README.md index a360da7..80ca103 100644 --- a/README.md +++ b/README.md @@ -47,6 +47,25 @@ Waiting does **not** consume the restart budget: a slow dependency costs patience, not retries. This is what stops a Docker-backed server from being marked failed at login while the Docker daemon is still starting. +## Servers that read stdin + +MCP servers are stdio-first by convention, and several that also speak HTTP +still start their stdio transport unconditionally. Under the launchd agent the +daemon's stdin is `/dev/null`, so such a server sees EOF on its first read and +shuts down seconds after binding its port. + +`stdin` picks what the child gets on fd 0: + + stdin "keep-open" + +- `inherit` (default) — the child inherits the daemon's stdin. +- `null` — `/dev/null`; a read sees EOF immediately. +- `keep-open` — a pipe that `xy` holds open for the child's lifetime and never + writes to, so a read blocks instead of seeing EOF. + +Use `keep-open` for a server you want supervised in HTTP mode that insists on +running its stdio transport anyway. + ## Start on login (macOS) xy service install # write the LaunchAgent, load it, start the daemon diff --git a/crates/xy-protocol/src/config.rs b/crates/xy-protocol/src/config.rs index 66256b5..ab1af44 100644 --- a/crates/xy-protocol/src/config.rs +++ b/crates/xy-protocol/src/config.rs @@ -66,6 +66,19 @@ impl Default for StopConfig { use std::collections::BTreeMap; use std::path::PathBuf; +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum StdinMode { + /// The child inherits the daemon's stdin. Today's behaviour. + #[default] + Inherit, + /// The child gets `/dev/null`; a read sees EOF immediately. + Null, + /// The child gets a pipe that `xy` holds open and never writes to, + /// so a read blocks instead of seeing EOF. + KeepOpen, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct ServerConfig { pub name: String, @@ -83,6 +96,8 @@ pub struct ServerConfig { pub stop: StopConfig, #[serde(default)] pub wait_for: Option, + #[serde(default)] + pub stdin: StdinMode, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] diff --git a/crates/xy-protocol/src/kdl_parse.rs b/crates/xy-protocol/src/kdl_parse.rs index 14294ce..1a49643 100644 --- a/crates/xy-protocol/src/kdl_parse.rs +++ b/crates/xy-protocol/src/kdl_parse.rs @@ -1,5 +1,5 @@ use crate::{ - ConfigError, RestartConfig, RestartPolicy, ServerConfig, StopConfig, WaitCondition, + ConfigError, RestartConfig, RestartPolicy, ServerConfig, StdinMode, StopConfig, WaitCondition, WaitForConfig, default_wait_interval, default_wait_timeout, }; use kdl::{KdlDocument, KdlNode}; @@ -31,6 +31,7 @@ pub fn parse_server_config( let restart = parse_restart(&doc, source_path)?; let stop = parse_stop(&doc, source_path)?; let wait_for = parse_wait_for(&doc, source_path)?; + let stdin = parse_stdin(&doc, source_path)?; Ok(ServerConfig { name: name.to_string(), @@ -42,6 +43,7 @@ pub fn parse_server_config( restart, stop, wait_for, + stdin, }) } @@ -214,6 +216,27 @@ fn parse_restart(doc: &KdlDocument, path: &Path) -> Result Result { + let Some(node) = find_node(doc, "stdin") else { + return Ok(StdinMode::default()); + }; + + single_arg(node, "stdin", path)?; + + let s = string_arg(node, "stdin", path)?; + + match s.as_str() { + "inherit" => Ok(StdinMode::Inherit), + "null" => Ok(StdinMode::Null), + "keep-open" => Ok(StdinMode::KeepOpen), + other => Err(ConfigError::InvalidValue { + path: path.to_path_buf(), + field: "stdin", + message: format!("unknown stdin mode `{other}`"), + }), + } +} + fn parse_stop(doc: &KdlDocument, path: &Path) -> Result { let Some(node) = find_node(doc, "stop") else { return Ok(StopConfig::default()); @@ -810,4 +833,58 @@ stop { WaitCondition::Path(PathBuf::from("/x")) ); } + + #[test] + fn stdin_defaults_to_inherit() { + let text = "command \"/bin/x\"\nport 1"; + let cfg = parse_server_config("foo", text, p()).unwrap(); + + assert_eq!(cfg.stdin, StdinMode::Inherit); + } + + #[test] + fn parses_stdin_inherit() { + let text = "command \"/bin/x\"\nport 1\nstdin \"inherit\""; + let cfg = parse_server_config("foo", text, p()).unwrap(); + + assert_eq!(cfg.stdin, StdinMode::Inherit); + } + + #[test] + fn parses_stdin_null() { + let text = "command \"/bin/x\"\nport 1\nstdin \"null\""; + let cfg = parse_server_config("foo", text, p()).unwrap(); + + assert_eq!(cfg.stdin, StdinMode::Null); + } + + #[test] + fn parses_stdin_keep_open() { + let text = "command \"/bin/x\"\nport 1\nstdin \"keep-open\""; + let cfg = parse_server_config("foo", text, p()).unwrap(); + + assert_eq!(cfg.stdin, StdinMode::KeepOpen); + } + + #[test] + fn unknown_stdin_mode_fails() { + let text = "command \"/bin/x\"\nport 1\nstdin \"maybe\""; + let err = parse_server_config("foo", text, p()).unwrap_err(); + + assert!(matches!( + err, + ConfigError::InvalidValue { field: "stdin", .. } + )); + } + + #[test] + fn stdin_with_two_args_fails() { + let text = "command \"/bin/x\"\nport 1\nstdin \"null\" \"keep-open\""; + let err = parse_server_config("foo", text, p()).unwrap_err(); + + assert!(matches!( + err, + ConfigError::InvalidValue { field: "stdin", .. } + )); + } } diff --git a/crates/xy-protocol/src/lib.rs b/crates/xy-protocol/src/lib.rs index 669ed86..7b6d828 100644 --- a/crates/xy-protocol/src/lib.rs +++ b/crates/xy-protocol/src/lib.rs @@ -7,8 +7,8 @@ pub mod rpc; pub mod state; pub use config::{ - RestartConfig, RestartPolicy, ServerConfig, StopConfig, WaitCondition, WaitForConfig, - default_wait_interval, default_wait_timeout, + RestartConfig, RestartPolicy, ServerConfig, StdinMode, StopConfig, WaitCondition, + WaitForConfig, default_wait_interval, default_wait_timeout, }; pub use error::{ConfigError, RpcErrorCode}; pub use kdl_parse::{load_all_configs, parse_server_config}; diff --git a/crates/xy-supervisor/src/child.rs b/crates/xy-supervisor/src/child.rs index eb9c361..097cdaf 100644 --- a/crates/xy-supervisor/src/child.rs +++ b/crates/xy-supervisor/src/child.rs @@ -76,13 +76,18 @@ use nix::sys::signal::{Signal, kill}; use nix::unistd::Pid; use std::process::Stdio; use tokio::io::{AsyncBufReadExt, BufReader}; -use tokio::process::{Child as TokioChild, Command}; -use xy_protocol::{ServerConfig, rpc::LogStream}; +use tokio::process::{Child as TokioChild, ChildStdin, Command}; +use xy_protocol::{ServerConfig, StdinMode, rpc::LogStream}; pub struct RealChild { pid: u32, pgid: Pid, child: Option, + /// Write end of the child's stdin pipe under [`StdinMode::KeepOpen`]. + /// Never written to; held so the child blocks on read rather than + /// seeing EOF. `TokioChild::wait()` drops the handle it owns, so the + /// pipe must live out here to outlast supervision. + _stdin: Option, } impl RealChild { @@ -132,6 +137,16 @@ pub fn spawn_with_logs(cfg: &ServerConfig, sink: LogSink) -> std::io::Result {} + StdinMode::Null => { + cmd.stdin(Stdio::null()); + } + StdinMode::KeepOpen => { + cmd.stdin(Stdio::piped()); + } + } + cmd.stdout(Stdio::piped()); cmd.stderr(Stdio::piped()); cmd.kill_on_drop(true); @@ -149,6 +164,8 @@ pub fn spawn_with_logs(cfg: &ServerConfig, sink: LogSink) -> std::io::Result std::io::Result ServerConfig { + ServerConfig { + name: "reader".to_string(), + command: "/bin/sh".into(), + args: vec!["-c".into(), script.to_string()], + port: 1, + env: Default::default(), + working_dir: None, + restart: RestartConfig::default(), + stop: StopConfig::default(), + wait_for: None, + stdin, + } + } + + fn test_sink() -> LogSink { + let dir = tempdir().unwrap(); + let writer = RotatingLogWriter::open(&dir.path().join("s.log"), 1024, 3).unwrap(); + std::mem::forget(dir); + LogSink::new("reader".to_string(), writer, 1024) + } + + /// Exits 9 straight away unless fd 0 is a FIFO, so the assertion below + /// cannot be satisfied by whatever stdin the test runner happened to have. + const REQUIRE_PIPE_THEN_READ: &str = "[ -p /dev/stdin ] || exit 9; read line; exit 7"; + + #[tokio::test] + async fn keep_open_gives_the_child_a_pipe_that_survives_wait() { + let cfg = reader_cfg(StdinMode::KeepOpen, REQUIRE_PIPE_THEN_READ); + let mut child = spawn_with_logs(&cfg, test_sink()).unwrap(); + + // `TokioChild::wait()` drops its own stdin handle on entry, so this + // blocks only if `RealChild` took the handle and is still holding it. + let outcome = tokio::time::timeout(Duration::from_millis(500), child.wait()).await; + + assert!( + outcome.is_err(), + "child should still be blocked on read, got {outcome:?}" + ); + + child.kill().unwrap(); + } + + #[tokio::test] + async fn null_stdin_child_sees_eof_immediately() { + let cfg = reader_cfg(StdinMode::Null, "read line; exit 7"); + let mut child = spawn_with_logs(&cfg, test_sink()).unwrap(); + + let code = tokio::time::timeout(Duration::from_millis(2000), child.wait()) + .await + .expect("child should exit promptly on EOF") + .unwrap(); + + assert_eq!(code, Some(7)); + } } diff --git a/crates/xy-supervisor/src/supervisor.rs b/crates/xy-supervisor/src/supervisor.rs index c05c456..d76d0bc 100644 --- a/crates/xy-supervisor/src/supervisor.rs +++ b/crates/xy-supervisor/src/supervisor.rs @@ -536,6 +536,7 @@ mod tests { grace: Duration::from_millis(50), }, wait_for: None, + stdin: Default::default(), } } diff --git a/crates/xy/Cargo.toml b/crates/xy/Cargo.toml index 20c0fee..1d7deef 100644 --- a/crates/xy/Cargo.toml +++ b/crates/xy/Cargo.toml @@ -16,6 +16,10 @@ path = "src/bin/xy_test_sleep_server.rs" name = "xy-test-exit-failure" path = "src/bin/xy_test_exit_failure.rs" +[[bin]] +name = "xy-test-stdin-reader" +path = "src/bin/xy_test_stdin_reader.rs" + [dependencies] xy-protocol.workspace = true xy-supervisor.workspace = true diff --git a/crates/xy/src/bin/xy_test_stdin_reader.rs b/crates/xy/src/bin/xy_test_stdin_reader.rs new file mode 100644 index 0000000..97cbf58 --- /dev/null +++ b/crates/xy/src/bin/xy_test_stdin_reader.rs @@ -0,0 +1,21 @@ +/// Stands in for a stdio-first MCP server: it comes up, announces itself, then +/// blocks reading stdin and exits the moment it sees EOF. +fn main() { + use std::io::{BufRead, Write}; + + println!("ready"); + std::io::stdout().flush().ok(); + + let mut line = String::new(); + match std::io::stdin().lock().read_line(&mut line) { + Ok(0) => { + eprintln!("stdin closed (EOF), shutting down"); + std::process::exit(0); + } + Ok(_) => std::process::exit(1), + Err(err) => { + eprintln!("stdin read error: {err}"); + std::process::exit(2); + } + } +} diff --git a/crates/xy/tests/common/mod.rs b/crates/xy/tests/common/mod.rs index 90515a1..5797c33 100644 --- a/crates/xy/tests/common/mod.rs +++ b/crates/xy/tests/common/mod.rs @@ -37,8 +37,20 @@ impl Harness { } pub fn write_server(&self, name: &str, command: &str, port: u16, restart_policy: &str) { + self.write_server_with(name, command, port, restart_policy, ""); + } + + /// `extra` is appended verbatim as additional top-level KDL. + pub fn write_server_with( + &self, + name: &str, + command: &str, + port: u16, + restart_policy: &str, + extra: &str, + ) { let body = format!( - "command \"{command}\"\nport {port}\nrestart {{\n policy \"{restart_policy}\"\n backoff-initial \"10ms\"\n backoff-max \"50ms\"\n max-retries-per-minute 3\n}}\nstop {{ grace \"500ms\" }}\n" + "command \"{command}\"\nport {port}\nrestart {{\n policy \"{restart_policy}\"\n backoff-initial \"10ms\"\n backoff-max \"50ms\"\n max-retries-per-minute 3\n}}\nstop {{ grace \"500ms\" }}\n{extra}" ); std::fs::write(self.config_dir.join(format!("{name}.kdl")), body).unwrap(); } @@ -49,6 +61,7 @@ impl Harness { .env("XDG_CONFIG_HOME", self.tmp.path().join("config")) .env("XDG_STATE_HOME", self.tmp.path().join("state")) .env("XDG_RUNTIME_DIR", self.tmp.path().join("run")) + .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::inherit()) .kill_on_drop(true) @@ -99,6 +112,9 @@ pub fn sleep_server_bin() -> PathBuf { pub fn exit_failure_bin() -> PathBuf { artifact("xy-test-exit-failure") } +pub fn stdin_reader_bin() -> PathBuf { + artifact("xy-test-stdin-reader") +} fn artifact(name: &str) -> PathBuf { let mut p = std::env::current_exe().unwrap(); diff --git a/crates/xy/tests/stdin_mode.rs b/crates/xy/tests/stdin_mode.rs new file mode 100644 index 0000000..24bb4b6 --- /dev/null +++ b/crates/xy/tests/stdin_mode.rs @@ -0,0 +1,68 @@ +mod common; +use common::*; +use std::time::Duration; +use xy_protocol::ServerState; + +async fn state_of(h: &Harness, name: &str) -> ServerState { + h.list_rpc() + .await + .into_iter() + .find(|s| s.name == name) + .unwrap_or_else(|| panic!("no server named {name}")) + .state +} + +async fn settle(h: &Harness, name: &str) -> ServerState { + let mut last = ServerState::Stopped; + for _ in 0..40 { + last = state_of(h, name).await; + if last == ServerState::Running { + break; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + last +} + +#[tokio::test] +async fn keep_open_lets_a_stdin_reader_stay_running() { + let xy = xy_bin(); + let reader = stdin_reader_bin(); + let mut h = Harness::new(); + h.write_server_with( + "reader", + reader.to_str().unwrap(), + 19_010, + "never", + "stdin \"keep-open\"\n", + ); + h.start_daemon(&xy).await; + + assert_eq!(settle(&h, "reader").await, ServerState::Running); + + // Still up a beat later: it is blocked on read, not merely slow to die. + tokio::time::sleep(Duration::from_millis(500)).await; + assert_eq!(state_of(&h, "reader").await, ServerState::Running); +} + +/// Characterizes the reported bug: the daemon's stdin is `/dev/null` under +/// launchd, so by default a stdio-first server sees EOF and exits at once. +#[tokio::test] +async fn without_keep_open_a_stdin_reader_exits_on_eof() { + let xy = xy_bin(); + let reader = stdin_reader_bin(); + let mut h = Harness::new(); + h.write_server("reader", reader.to_str().unwrap(), 19_011, "never"); + h.start_daemon(&xy).await; + + let mut last = ServerState::Running; + for _ in 0..40 { + last = state_of(&h, "reader").await; + if last == ServerState::Stopped { + break; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + + assert_eq!(last, ServerState::Stopped); +}