diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d856ea..aa00c7a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,15 @@ All notable changes to `devloop` will be recorded in this file. ## [Unreleased] +## [0.10.4] - 2026-09-01 + +### Fixed + +- Made startup stop with a clear non-zero error when a managed process's local + readiness address is already occupied, without restarting the failed process + or disturbing the existing listener. Concurrent sessions remain supported + when they use different ports. + ## [0.10.3] - 2026-08-26 ### Fixed diff --git a/Cargo.lock b/Cargo.lock index 2f44a52..53e298a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -235,7 +235,7 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" [[package]] name = "devloop" -version = "0.10.3" +version = "0.10.4" dependencies = [ "anyhow", "axum", diff --git a/Cargo.toml b/Cargo.toml index 59c9e60..77a8b98 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "devloop" -version = "0.10.3" +version = "0.10.4" edition = "2024" [dependencies] diff --git a/docs/behavior.md b/docs/behavior.md index c767286..2c01e98 100644 --- a/docs/behavior.md +++ b/docs/behavior.md @@ -43,6 +43,14 @@ When `devloop run` starts, it: 7. runs each workflow named in `startup_workflows` in order 8. starts watching the configured `root` +Before starting a managed process with a loopback HTTP readiness probe, +`devloop` checks whether it can bind that process-owned readiness address. An +occupied address, including one that is bound but not yet accepting +connections, stops startup with a non-zero error that names the process and +address. The colliding session does not enter its restart loop or disturb the +existing listener. This check is per configured address: concurrent sessions +and worktrees remain independent when they use different ports. + The in-memory session state is authoritative for the running process. Edits made directly to the JSON file while `devloop` is running are not merged back into the live session. @@ -124,7 +132,10 @@ Managed processes are long-running child commands. process stops or exits, so a failed dependency cannot leave a stale URL or other process-derived value available to later workflows. - `restart = "always"` restarts a child after any exit unless - `devloop` is shutting down. + `devloop` is shutting down. A child that exits while a startup workflow is + waiting for its first readiness result is treated as a failed start and is + not restarted before that startup failure terminates the session. Later + runtime workflows preserve the configured restart policy. - `restart = "on_failure"` restarts only after unsuccessful exit. - `restart = "never"` never restarts automatically. - Restart policies use the managed command's exit status. A wrapper that diff --git a/src/core.rs b/src/core.rs index 24340fc..b6098b7 100644 --- a/src/core.rs +++ b/src/core.rs @@ -100,6 +100,7 @@ pub enum RuntimeEffect { RunWorkflow { workflow_name: String, changed_files: Vec, + origin: WorkflowRunOrigin, }, StartWatching, MaintainProcesses, @@ -115,6 +116,12 @@ pub enum RuntimeEffect { Exit, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WorkflowRunOrigin { + Startup, + Runtime, +} + #[derive(Debug, Clone)] struct ObservedHookRuntimeState { workflow_name: String, @@ -371,6 +378,7 @@ impl RuntimeMachine { self.pending_effects.push_back(RuntimeEffect::RunWorkflow { workflow_name, changed_files: Vec::new(), + origin: WorkflowRunOrigin::Startup, }); } self.pending_effects.push_back(RuntimeEffect::StartWatching); @@ -384,6 +392,7 @@ impl RuntimeMachine { self.pending_effects.push_back(RuntimeEffect::RunWorkflow { workflow_name, changed_files, + origin: WorkflowRunOrigin::Runtime, }); } } @@ -413,6 +422,7 @@ impl RuntimeMachine { self.pending_effects.push_front(RuntimeEffect::RunWorkflow { workflow_name, changed_files: Vec::new(), + origin: WorkflowRunOrigin::Runtime, }); } RuntimeEvent::CtrlC => { @@ -827,14 +837,16 @@ mod tests { runtime.next_effect(), Some(RuntimeEffect::RunWorkflow { workflow_name: "startup".into(), - changed_files: Vec::new() + changed_files: Vec::new(), + origin: WorkflowRunOrigin::Startup, }) ); assert_eq!( runtime.next_effect(), Some(RuntimeEffect::RunWorkflow { workflow_name: "publish".into(), - changed_files: Vec::new() + changed_files: Vec::new(), + origin: WorkflowRunOrigin::Startup, }) ); assert_eq!(runtime.next_effect(), Some(RuntimeEffect::StartWatching)); @@ -890,7 +902,8 @@ mod tests { runtime.next_effect(), Some(RuntimeEffect::RunWorkflow { workflow_name: "rust".into(), - changed_files: vec!["src/main.rs".into()] + changed_files: vec!["src/main.rs".into()], + origin: WorkflowRunOrigin::Runtime, }) ); assert_eq!( @@ -967,6 +980,7 @@ mod tests { Some(RuntimeEffect::RunWorkflow { workflow_name: "publish_post_url".into(), changed_files: Vec::new(), + origin: WorkflowRunOrigin::Runtime, }) ); assert_eq!(runtime.next_effect(), None); diff --git a/src/engine.rs b/src/engine.rs index d320bd6..93f57f7 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; -use anyhow::{Result, anyhow}; +use anyhow::{Context, Result, anyhow}; use notify::{ Config as NotifyConfig, Event, EventKind, PollWatcher, RecommendedWatcher, RecursiveMode, Watcher, @@ -18,9 +18,11 @@ use unicode_width::UnicodeWidthStr; use crate::browser_reload::{BrowserReloadSender, BrowserReloadServer, notify_browser_reload}; use crate::config::{CompiledWatchGroup, CompiledWatchTarget, Config, LogStyle, WatcherKind}; -use crate::core::{RuntimeEffect, RuntimeEvent, RuntimeMachine, WorkflowEffect, WorkflowMachine}; +use crate::core::{ + RuntimeEffect, RuntimeEvent, RuntimeMachine, WorkflowEffect, WorkflowMachine, WorkflowRunOrigin, +}; use crate::external_events::{ExternalEventMessage, ExternalEventServer}; -use crate::processes::ProcessManager; +use crate::processes::{ManagedAddressInUse, ProcessManager}; use crate::session_log::SessionLog; use crate::state::SessionState; use devloop::process_guardian::GuardianExecutable; @@ -54,7 +56,12 @@ trait RuntimeEffectAdapter { async fn start_external_event_server(&mut self) -> Result<()>; async fn start_browser_reload_server(&mut self) -> Result<()>; async fn start_autostart_processes(&mut self) -> Result<()>; - async fn run_workflow(&mut self, workflow_name: &str, changed_files: &[String]) -> Result<()>; + async fn run_workflow( + &mut self, + workflow_name: &str, + changed_files: &[String], + origin: WorkflowRunOrigin, + ) -> Result<()>; async fn start_watching(&mut self) -> Result<()>; async fn maintain_processes(&mut self) -> Result<()>; async fn poll_observed_hook(&mut self, hook: &str) -> Result; @@ -67,6 +74,7 @@ struct LiveWorkflowAdapter<'a, 'b> { processes: &'a mut ProcessManager<'b>, state: &'a SessionState, browser_reload_sender: Option, + restart_after_failed_start: bool, } struct LiveRuntimeAdapter<'a, 'b> { @@ -223,7 +231,9 @@ impl WorkflowEffectAdapter for LiveWorkflowAdapter<'_, '_> { } async fn wait_for_process(&mut self, process: &str) -> Result<()> { - self.processes.wait_for_named(process, self.state).await + self.processes + .wait_for_named(process, self.state, self.restart_after_failed_start) + .await } async fn run_hook( @@ -302,7 +312,12 @@ impl RuntimeEffectAdapter for LiveRuntimeAdapter<'_, '_> { self.processes.start_autostart(self.state).await } - async fn run_workflow(&mut self, workflow_name: &str, changed_files: &[String]) -> Result<()> { + async fn run_workflow( + &mut self, + workflow_name: &str, + changed_files: &[String], + origin: WorkflowRunOrigin, + ) -> Result<()> { info!("running workflow {}", workflow_name); let mut adapter = LiveWorkflowAdapter { processes: self.processes, @@ -311,6 +326,7 @@ impl RuntimeEffectAdapter for LiveRuntimeAdapter<'_, '_> { .browser_reload_server .as_ref() .map(BrowserReloadServer::sender), + restart_after_failed_start: origin == WorkflowRunOrigin::Runtime, }; execute_workflow(self.config, &mut adapter, workflow_name, changed_files).await } @@ -409,12 +425,32 @@ async fn execute_runtime_effects( RuntimeEffect::StartBrowserReloadServer => { adapter.start_browser_reload_server().await? } - RuntimeEffect::StartAutostartProcesses => adapter.start_autostart_processes().await?, + RuntimeEffect::StartAutostartProcesses => { + if let Err(error) = adapter.start_autostart_processes().await { + adapter + .stop_all_processes() + .await + .context("failed to clean up after autostart failure")?; + return Err(error); + } + } RuntimeEffect::RunWorkflow { workflow_name, changed_files, + origin, } => { - if let Err(error) = adapter.run_workflow(&workflow_name, &changed_files).await { + if let Err(error) = adapter + .run_workflow(&workflow_name, &changed_files, origin) + .await + { + if origin == WorkflowRunOrigin::Startup + && error.downcast_ref::().is_some() + { + adapter.stop_all_processes().await.with_context(|| { + format!("failed to clean up after startup workflow '{workflow_name}'") + })?; + return Err(error); + } error!( workflow = %workflow_name, error = %workflow_failure_chain(&error), @@ -496,6 +532,7 @@ async fn run_workflow( processes, state, browser_reload_sender, + restart_after_failed_start: true, }; execute_workflow(config, &mut adapter, workflow_name, changed_files).await } @@ -1601,6 +1638,7 @@ mod tests { calls: Vec, changed_hooks: BTreeMap, workflow_errors: BTreeMap, + workflow_address_conflicts: BTreeSet, stop_watching_error: Option, watching: bool, } @@ -1611,6 +1649,7 @@ mod tests { calls: Vec::new(), changed_hooks: BTreeMap::new(), workflow_errors: BTreeMap::new(), + workflow_address_conflicts: BTreeSet::new(), stop_watching_error: None, watching: false, } @@ -1642,7 +1681,11 @@ mod tests { &mut self, workflow_name: &str, changed_files: &[String], + _origin: WorkflowRunOrigin, ) -> Result<()> { + if self.workflow_address_conflicts.contains(workflow_name) { + return Err(ManagedAddressInUse::new("server", "127.0.0.1:8787").into()); + } if let Some(message) = self.workflow_errors.get(workflow_name) { return Err(anyhow!(message.clone())); } @@ -2003,6 +2046,42 @@ mod tests { assert_eq!(adapter.calls, vec!["persist:root", "autostart", "watch"]); } + #[tokio::test] + async fn startup_address_collision_stops_and_cleans_up_before_watching() { + let config = Config { + root: PathBuf::from("."), + debounce_ms: 100, + watcher: crate::config::WatcherConfig::default(), + state_file: Some(PathBuf::from("./state.json")), + startup_workflows: vec!["startup".into()], + watch: BTreeMap::new(), + process: BTreeMap::new(), + hook: BTreeMap::new(), + event_server: crate::config::EventServerConfig::default(), + browser_reload_server: crate::config::BrowserReloadServerConfig::default(), + event: BTreeMap::new(), + workflow: BTreeMap::new(), + }; + let mut runtime = RuntimeMachine::new(&config); + runtime.handle_event(RuntimeEvent::Start { + root_display: "/tmp/example".into(), + startup_workflows: vec!["startup".into()], + }); + let mut adapter = MockRuntimeAdapter::new(); + adapter.workflow_address_conflicts.insert("startup".into()); + + let error = execute_runtime_effects(&mut runtime, &mut adapter) + .await + .expect_err("startup collision must stop the runtime"); + + assert_eq!( + error.to_string(), + "process 'server' cannot start: address 127.0.0.1:8787 is already in use" + ); + assert_eq!(adapter.calls, vec!["persist:root", "autostart", "stop_all"]); + assert!(!adapter.watching); + } + #[tokio::test] async fn missing_runtime_workflow_returns_error() { let config = Config { diff --git a/src/main.rs b/src/main.rs index 518509d..338e86e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,6 +13,7 @@ mod test_support; use std::io::{self, Write}; use std::path::PathBuf; +use std::process::ExitCode; use std::time::Duration; use anyhow::{Context, Result, anyhow}; @@ -20,7 +21,7 @@ use clap::{Parser, Subcommand, ValueEnum}; use pulldown_cmark::{ CodeBlockKind, Event as MarkdownEvent, HeadingLevel, Parser as MarkdownParser, Tag, TagEnd, }; -use tracing::{Event, Subscriber, error}; +use tracing::{Event, Subscriber}; use tracing_subscriber::EnvFilter; use tracing_subscriber::fmt::FmtContext; use tracing_subscriber::fmt::MakeWriter; @@ -77,7 +78,17 @@ enum DocsTopic { } #[tokio::main] -async fn main() -> Result<()> { +async fn main() -> ExitCode { + match run_cli().await { + Ok(exit_code) => exit_code, + Err(error) => { + eprintln!("devloop: {error:#}"); + ExitCode::FAILURE + } + } +} + +async fn run_cli() -> Result { let cli = Cli::parse(); match cli.command { Command::Validate { config } => { @@ -102,9 +113,9 @@ async fn main() -> Result<()> { .run() .await { - error!(error = %format!("{error:#}"), "devloop run failed"); + report_run_failure(&session_log, &error); flush_session_log_before_exit(&session_log).await; - return Err(error); + return Ok(ExitCode::FAILURE); } flush_session_log_before_exit(&session_log).await; } @@ -112,7 +123,15 @@ async fn main() -> Result<()> { print!("{}", render_docs_text(topic)); } } - Ok(()) + Ok(ExitCode::SUCCESS) +} + +fn report_run_failure(session_log: &SessionLog, error: &anyhow::Error) { + let message = format!("devloop run failed: {error:#}"); + if let Err(log_error) = session_log.write_labeled_line("devloop", message.as_bytes()) { + eprintln!("devloop: failed to persist run failure: {log_error}"); + } + eprintln!("devloop: {message}"); } async fn flush_session_log_before_exit(session_log: &SessionLog) { diff --git a/src/processes.rs b/src/processes.rs index 3529757..67a80c2 100644 --- a/src/processes.rs +++ b/src/processes.rs @@ -1,4 +1,6 @@ use std::collections::{BTreeMap, VecDeque}; +use std::fmt; +use std::net::{IpAddr, SocketAddr, TcpListener}; use std::os::fd::AsRawFd; use std::os::unix::net::UnixStream; use std::path::{Path, PathBuf}; @@ -81,6 +83,37 @@ const TERMINAL_OUTPUT_QUEUE_CAPACITY: usize = 256; const SESSION_LOG_FLUSH_TIMEOUT: Duration = Duration::from_secs(5); const PROCESS_STOP_TIMEOUT: Duration = Duration::from_secs(2); const GUARDIAN_REAP_TIMEOUT: Duration = Duration::from_secs(2); +#[derive(Debug)] +pub(crate) struct ManagedAddressInUse { + process: String, + address: String, +} + +impl ManagedAddressInUse { + pub(crate) fn new(process: &str, address: impl ToString) -> Self { + Self { + process: process.to_owned(), + address: address.to_string(), + } + } +} + +struct LocalReadinessAddress { + display: String, + candidates: Vec, +} + +impl fmt::Display for ManagedAddressInUse { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + formatter, + "process '{}' cannot start: address {} is already in use", + self.process, self.address + ) + } +} + +impl std::error::Error for ManagedAddressInUse {} struct CommandContext<'a> { env: &'a BTreeMap, @@ -136,11 +169,19 @@ impl<'a> ProcessManager<'a> { if self.shutting_down { return Ok(()); } + if self.children.contains_key(name) { + return Ok(()); + } let spec = self .config .process .get(name) .ok_or_else(|| anyhow!("unknown process '{name}'"))?; + if let Some(address) = local_readiness_address(name, spec)? + && address_is_occupied(&address) + { + return Err(ManagedAddressInUse::new(name, address.display).into()); + } self.start(name, spec, state).await } @@ -175,7 +216,12 @@ impl<'a> ProcessManager<'a> { self.start_named(name, state).await } - pub async fn wait_for_named(&mut self, name: &str, state: &SessionState) -> Result<()> { + pub async fn wait_for_named( + &mut self, + name: &str, + state: &SessionState, + restart_after_failed_start: bool, + ) -> Result<()> { let (readiness, output_rules) = self .config .process @@ -184,7 +230,7 @@ impl<'a> ProcessManager<'a> { .ok_or_else(|| anyhow!("unknown process '{name}'"))?; let Some(probe) = readiness else { return self - .ensure_process_running(name, &output_rules, state) + .ensure_process_running(name, &output_rules, state, restart_after_failed_start) .await; }; let probe = expand_probe_env(name, &probe)?; @@ -196,10 +242,10 @@ impl<'a> ProcessManager<'a> { }; let interval = Duration::from_millis(probe.interval()); loop { - self.ensure_process_running(name, &output_rules, state) + self.ensure_process_running(name, &output_rules, state, restart_after_failed_start) .await?; if check_probe(&self.client, name, &probe, state).await.is_ok() { - self.ensure_process_running(name, &output_rules, state) + self.ensure_process_running(name, &output_rules, state, restart_after_failed_start) .await?; return Ok(()); } @@ -215,6 +261,7 @@ impl<'a> ProcessManager<'a> { name: &str, output_rules: &[OutputRule], state: &SessionState, + restart_after_failed_start: bool, ) -> Result<()> { let child = self .children @@ -229,13 +276,29 @@ impl<'a> ProcessManager<'a> { self.retire_output_state(name, output_rules, state); signal_process_group(name, managed.guarded.process_group, Signal::KILL)?; self.spawn_output_cleanup(name.to_owned(), managed.output_tasks); - let now_ms = self.clock_start.elapsed().as_millis() as u64; - for effect in self.supervisor.on_tick( - self.config, - now_ms, - vec![(name.to_owned(), status.success())], - ) { - self.apply_process_effect(effect, state).await?; + let occupied_address = if let Some(spec) = self.config.process.get(name) + && let Some(address) = local_readiness_address(name, spec)? + && address_is_occupied(&address) + { + Some(address.display) + } else { + None + }; + if let Some(address) = occupied_address { + self.supervisor.on_process_stopped(name); + return Err(ManagedAddressInUse::new(name, address).into()); + } + if restart_after_failed_start { + let now_ms = self.clock_start.elapsed().as_millis() as u64; + for effect in self.supervisor.on_tick( + self.config, + now_ms, + vec![(name.to_owned(), status.success())], + ) { + self.apply_process_effect(effect, state).await?; + } + } else { + self.supervisor.on_process_stopped(name); } return Err(anyhow!( "process '{name}' exited with {status} before it became ready" @@ -630,6 +693,59 @@ impl<'a> ProcessManager<'a> { } } +fn local_readiness_address( + name: &str, + spec: &ProcessSpec, +) -> Result> { + let Some(readiness) = &spec.readiness else { + return Ok(None); + }; + let ProbeSpec::Http { url, .. } = expand_probe_env(name, readiness)? else { + return Ok(None); + }; + let url = reqwest::Url::parse(&url) + .with_context(|| format!("invalid HTTP readiness URL for process '{name}'"))?; + let port = url + .port_or_known_default() + .ok_or_else(|| anyhow!("HTTP readiness URL for process '{name}' has no port"))?; + let Some(host) = url.host_str() else { + return Ok(None); + }; + if host == "localhost" { + return Ok(Some(LocalReadinessAddress { + display: format!("localhost:{port}"), + candidates: vec![ + SocketAddr::new(IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), port), + SocketAddr::new(IpAddr::V6(std::net::Ipv6Addr::LOCALHOST), port), + ], + })); + } + let host = host.trim_start_matches('[').trim_end_matches(']'); + let ip = if let Ok(ip) = host.parse::() + && ip.is_loopback() + { + ip + } else { + return Ok(None); + }; + let address = SocketAddr::new(ip, port); + Ok(Some(LocalReadinessAddress { + display: address.to_string(), + candidates: vec![address], + })) +} + +fn address_is_occupied(address: &LocalReadinessAddress) -> bool { + for candidate in &address.candidates { + match TcpListener::bind(candidate) { + Ok(listener) => drop(listener), + Err(error) if error.kind() == std::io::ErrorKind::AddrInUse => return true, + Err(_) => {} + } + } + false +} + /// Spawns a managed command behind devloop's internal Rust guardian. /// /// The guardian stays outside the target process group, so ordinary TERM/KILL @@ -1966,7 +2082,7 @@ mod tests { ); let error = manager - .wait_for_named("tunnel", &state) + .wait_for_named("tunnel", &state, false) .await .expect_err("stale state must not imply readiness"); @@ -1994,7 +2110,7 @@ mod tests { timeout_ms: 1000, }), liveness: None, - restart: crate::config::RestartPolicy::Never, + restart: crate::config::RestartPolicy::Always, env: BTreeMap::new(), output: OutputConfig { rules: vec![OutputRule { @@ -2041,7 +2157,7 @@ mod tests { .expect("capture readiness before observing server exit"); let error = manager - .wait_for_named("server", &state) + .wait_for_named("server", &state, false) .await .expect_err("exited process must not become ready"); @@ -2408,7 +2524,7 @@ exec sleep 600 .await .expect("start process"); manager - .wait_for_named("server", &state) + .wait_for_named("server", &state, true) .await .expect("wait for first process readiness"); log.fail_for_test( @@ -2422,7 +2538,7 @@ exec sleep 600 .expect("restart process"); manager - .wait_for_named("server", &state) + .wait_for_named("server", &state, true) .await .expect("wait for restarted process readiness"); assert!(manager.children.contains_key("server")); diff --git a/tests/port_collisions.rs b/tests/port_collisions.rs new file mode 100644 index 0000000..d2793ef --- /dev/null +++ b/tests/port_collisions.rs @@ -0,0 +1,226 @@ +#![cfg(unix)] + +use std::io::Write; +use std::net::{SocketAddr, TcpListener, TcpStream}; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +use rustix::process::{Pid, Signal, kill_process}; +use tempfile::TempDir; + +#[test] +#[ignore] +fn serve_ephemeral_port() { + if std::env::var_os("DEVLOOP_TEST_SERVER_MODE").is_none() { + return; + } + let listener = TcpListener::bind("127.0.0.1:0").expect("bind ephemeral test server"); + println!( + "listening {}", + listener.local_addr().expect("read server address").port() + ); + std::io::stdout().flush().expect("flush server address"); + loop { + let _ = listener.accept().expect("accept test connection"); + } +} + +#[test] +fn separate_sessions_can_run_on_distinct_ports() { + let first = SessionFixture::ephemeral(); + let second = SessionFixture::ephemeral(); + + let mut first_session = DevloopChild::spawn(&first); + let mut second_session = DevloopChild::spawn(&second); + + let first_address = wait_for_reported_listener(&first, &mut first_session.child); + let second_address = wait_for_reported_listener(&second, &mut second_session.child); + assert_ne!(first_address, second_address); + first_session.assert_running(); + second_session.assert_running(); +} + +#[test] +fn startup_fails_once_when_a_managed_address_is_occupied() { + let incumbent = TcpListener::bind("127.0.0.1:0").expect("bind incumbent listener"); + assert_startup_collision(&incumbent); +} + +#[test] +fn startup_detects_an_occupied_ipv6_loopback_address() { + let Ok(incumbent) = TcpListener::bind("[::1]:0") else { + return; + }; + assert_startup_collision(&incumbent); +} + +fn assert_startup_collision(incumbent: &TcpListener) { + let address = incumbent.local_addr().expect("read incumbent address"); + let fixture = SessionFixture::new(address); + + let output = Command::new(env!("CARGO_BIN_EXE_devloop")) + .arg("run") + .arg("--config") + .arg(fixture.config_path()) + .current_dir(fixture.path()) + .env("RUST_LOG", "off") + .output() + .expect("run colliding devloop session"); + + assert!(!output.status.success(), "collision must exit non-zero"); + let stderr = String::from_utf8(output.stderr).expect("stderr is UTF-8"); + let diagnostic = format!("process 'server' cannot start: address {address} is already in use"); + assert_eq!( + stderr.matches(&diagnostic).count(), + 1, + "collision must produce one concise diagnostic; stderr: {stderr}" + ); + assert!(!stderr.contains("started process server"), "{stderr}"); + TcpStream::connect(address).expect("incumbent listener remains reachable"); +} + +fn wait_for_reported_listener(fixture: &SessionFixture, child: &mut Child) -> SocketAddr { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + if let Ok(raw_state) = std::fs::read_to_string(fixture.state_path()) + && let Ok(state) = serde_json::from_str::(&raw_state) + && let Some(port) = state + .get("server_port") + .and_then(serde_json::Value::as_str) + .and_then(|port| port.parse::().ok()) + { + let address = SocketAddr::from(([127, 0, 0, 1], port)); + if TcpStream::connect(address).is_ok() { + return address; + } + } + if let Some(status) = child.try_wait().expect("read devloop status") { + panic!("devloop exited before its ephemeral port became ready: {status}"); + } + assert!( + Instant::now() < deadline, + "timed out waiting for devloop's reported listener" + ); + std::thread::yield_now(); + } +} + +struct SessionFixture { + dir: TempDir, +} + +impl SessionFixture { + fn new(address: SocketAddr) -> Self { + let dir = tempfile::tempdir().expect("create session fixture"); + let fixture = Self { dir }; + let config = format!( + r#"root = "." +state_file = "./.devloop/state.json" +startup_workflows = ["startup"] + +[watch.config] +paths = ["devloop.toml"] +workflow = "startup" + +[process.server] +command = ["devloop-test-command-that-must-not-run"] +autostart = false +readiness = {{ kind = "http", url = "http://{address}/", interval_ms = 20, timeout_ms = 5000 }} +restart = "always" + +[workflow.startup] +steps = [ + {{ action = "start_process", process = "server" }}, + {{ action = "wait_for_process", process = "server" }}, +] +"#, + ); + std::fs::write(fixture.config_path(), config).expect("write session config"); + fixture + } + + fn ephemeral() -> Self { + let dir = tempfile::tempdir().expect("create session fixture"); + let fixture = Self { dir }; + let test_binary = std::env::current_exe().expect("resolve integration test executable"); + let config = format!( + r#"root = "." +state_file = "./.devloop/state.json" +startup_workflows = ["startup"] + +[watch.config] +paths = ["devloop.toml"] +workflow = "startup" + +[process.server] +command = ["{test_binary}", "--exact", "serve_ephemeral_port", "--ignored", "--nocapture"] +autostart = false +readiness = {{ kind = "state_key", key = "server_port", interval_ms = 20, timeout_ms = 5000 }} +restart = "always" +env = {{ DEVLOOP_TEST_SERVER_MODE = "1" }} +output = {{ inherit = false, rules = [{{ state_key = "server_port", pattern = "^listening ([0-9]+)$", extract = "regex", capture_group = 1 }}] }} + +[workflow.startup] +steps = [ + {{ action = "start_process", process = "server" }}, + {{ action = "wait_for_process", process = "server" }}, +] +"#, + test_binary = test_binary.display() + ); + std::fs::write(fixture.config_path(), config).expect("write session config"); + fixture + } + + fn path(&self) -> &std::path::Path { + self.dir.path() + } + + fn config_path(&self) -> std::path::PathBuf { + self.path().join("devloop.toml") + } + + fn state_path(&self) -> std::path::PathBuf { + self.path().join(".devloop/state.json") + } +} + +struct DevloopChild { + child: Child, +} + +impl DevloopChild { + fn spawn(fixture: &SessionFixture) -> Self { + let child = Command::new(env!("CARGO_BIN_EXE_devloop")) + .arg("run") + .arg("--config") + .arg(fixture.config_path()) + .current_dir(fixture.path()) + .env("RUST_LOG", "info") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn devloop session"); + Self { child } + } + + fn assert_running(&mut self) { + assert!( + self.child + .try_wait() + .expect("read devloop status") + .is_none(), + "devloop session exited" + ); + } +} + +impl Drop for DevloopChild { + fn drop(&mut self) { + if self.child.try_wait().ok().flatten().is_none() { + let pid = Pid::from_raw(self.child.id() as i32).expect("devloop pid"); + let _ = kill_process(pid, Signal::INT); + } + let _ = self.child.wait(); + } +}